X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2FRemoteRpcProvider.java;h=d24ed5651aa8a5ffa0d554769fb389c5dc05ab78;hb=3e5bfba47ae5fe04360343073273a141730daefd;hp=1bb7ea451441683750dcd8e4a18051e61f829e01;hpb=98194e992296b9f8d9871126726a3e11db165b1f;p=controller.git diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcProvider.java b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcProvider.java index 1bb7ea4514..d24ed5651a 100644 --- a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcProvider.java +++ b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcProvider.java @@ -11,47 +11,45 @@ package org.opendaylight.controller.remote.rpc; import akka.actor.ActorRef; import akka.actor.ActorSystem; -import org.opendaylight.controller.remote.rpc.registry.ClusterWrapper; -import org.opendaylight.controller.remote.rpc.registry.ClusterWrapperImpl; -import org.opendaylight.controller.remote.rpc.registry.RpcRegistry; +import org.opendaylight.controller.remote.rpc.messages.UpdateSchemaContext; import org.opendaylight.controller.sal.core.api.Broker; import org.opendaylight.controller.sal.core.api.Provider; import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry; import org.opendaylight.controller.sal.core.api.model.SchemaService; -import org.opendaylight.yangtools.yang.common.QName; import org.opendaylight.yangtools.yang.model.api.SchemaContext; +import org.opendaylight.yangtools.yang.model.api.SchemaContextListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Collection; -import java.util.Set; /** * This is the base class which initialize all the actors, listeners and * default RPc implementation so remote invocation of rpcs. */ -public class RemoteRpcProvider implements AutoCloseable, Provider{ +public class RemoteRpcProvider implements AutoCloseable, Provider, SchemaContextListener { private static final Logger LOG = LoggerFactory.getLogger(RemoteRpcProvider.class); - private final ActorSystem actorSystem; - private ActorRef rpcBroker; - private ActorRef rpcRegistry; private final RpcProvisionRegistry rpcProvisionRegistry; + + private ActorSystem actorSystem; private Broker.ProviderSession brokerSession; - private RpcListener rpcListener; - private RoutedRpcListener routeChangeListener; - private RemoteRpcImplementation rpcImplementation; + private SchemaContext schemaContext; + private ActorRef rpcManager; + private RemoteRpcProviderConfig config; + + public RemoteRpcProvider(ActorSystem actorSystem, RpcProvisionRegistry rpcProvisionRegistry) { this.actorSystem = actorSystem; this.rpcProvisionRegistry = rpcProvisionRegistry; + this.config = new RemoteRpcProviderConfig(actorSystem.settings().config()); } @Override public void close() throws Exception { - this.actorSystem.shutdown(); - unregisterSupportedRpcs(); - unregisterSupportedRoutedRpcs(); + if (this.actorSystem != null) + this.actorSystem.shutdown(); } @Override @@ -66,64 +64,23 @@ public class RemoteRpcProvider implements AutoCloseable, Provider{ } private void start() { - LOG.debug("Starting all rpc listeners."); - // Create actor to handle and sync routing table in cluster - ClusterWrapper clusterWrapper = new ClusterWrapperImpl(actorSystem); - rpcRegistry = actorSystem.actorOf(RpcRegistry.props(clusterWrapper), "rpc-registry"); + LOG.info("Starting remote rpc service..."); - // Create actor to invoke and execute rpc SchemaService schemaService = brokerSession.getService(SchemaService.class); - SchemaContext schemaContext = schemaService.getGlobalContext(); - rpcBroker = actorSystem.actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext), "rpc-broker"); - String rpcBrokerPath = clusterWrapper.getAddress().toString() + "/user/rpc-broker"; - rpcListener = new RpcListener(rpcRegistry, rpcBrokerPath); - routeChangeListener = new RoutedRpcListener(rpcRegistry, rpcBrokerPath); - rpcImplementation = new RemoteRpcImplementation(rpcBroker, schemaContext); - brokerSession.addRpcRegistrationListener(rpcListener); - rpcProvisionRegistry.registerRouteChangeListener(routeChangeListener); - rpcProvisionRegistry.setRoutedRpcDefaultDelegate(rpcImplementation); - announceSupportedRpcs(); - announceSupportedRoutedRpcs(); + schemaContext = schemaService.getGlobalContext(); - } - - /** - * Add all the locally registered RPCs in the clustered routing table - */ - private void announceSupportedRpcs(){ - LOG.debug("Adding all supported rpcs to routing table"); - Set currentlySupported = brokerSession.getSupportedRpcs(); - for (QName rpc : currentlySupported) { - rpcListener.onRpcImplementationAdded(rpc); - } - } + rpcManager = actorSystem.actorOf(RpcManager.props(schemaContext, brokerSession, rpcProvisionRegistry), + config.getRpcManagerName()); - /** - * Add all the locally registered Routed RPCs in the clustered routing table - */ - private void announceSupportedRoutedRpcs(){ + LOG.debug("rpc manager started"); - //TODO: announce all routed RPCs as well - - } - - /** - * Un-Register all the supported RPCs from clustered routing table - */ - private void unregisterSupportedRpcs(){ - LOG.debug("removing all supported rpcs to routing table"); - Set currentlySupported = brokerSession.getSupportedRpcs(); - for (QName rpc : currentlySupported) { - rpcListener.onRpcImplementationRemoved(rpc); - } + schemaService.registerSchemaContextListener(this); } - /** - * Un-Register all the locally supported Routed RPCs from clustered routing table - */ - private void unregisterSupportedRoutedRpcs(){ - - //TODO: remove all routed RPCs as well + @Override + public void onGlobalContextUpdated(SchemaContext schemaContext) { + this.schemaContext = schemaContext; + rpcManager.tell(new UpdateSchemaContext(schemaContext), null); } }