- public def addModule(BrokerModule module) {
- log.info("Registering broker module " + module);
- if(modules.contains(module)) {
- log.error("Module already registered");
- throw new IllegalArgumentException("Module already exists.");
- }
-
- val provServices = module.getProvidedServices();
- for (Class<? extends BrokerService> serviceType : provServices) {
- log.info(" Registering session service implementation: " + serviceType.getCanonicalName());
- serviceProviders.put(serviceType, module);
- }
- }
-
- public def <T extends BrokerService> T serviceFor(Class<T> service, ConsumerContextImpl session) {
- val prov = serviceProviders.get(service);
- if(prov == null) {
- log.warn("Service " + service.toString() + " is not supported");
- return null;
- }
- return prov.getServiceForSession(service, session);
- }
-
- // RPC Functionality
- protected def void addRpcImplementation(QName rpcType, RpcImplementation implementation) {
- if(rpcImpls.get(rpcType) != null) {
- throw new IllegalStateException("Implementation for rpc " + rpcType + " is already registered.");
- }
-
- //TODO: Add notification for availability of Rpc Implementation
- rpcImpls.put(rpcType, implementation);
- }
-
- protected def void removeRpcImplementation(QName rpcType) {
- rpcImpls.remove(rpcType);
- }
-
- protected def Future<RpcResult<CompositeNode>> invokeRpc(QName rpc, CompositeNode input) {
- val impl = rpcImpls.get(rpc);
- val result = executor.submit([|impl.invokeRpc(rpc, input)] as Callable<RpcResult<CompositeNode>>);