Change RpcImplementation contract to asynchronous

[controller.git] / opendaylight / md-sal / sal-netconf-connector / src / main / java / org / opendaylight / controller / sal / connect / netconf / NetconfDevice.xtend

diff --git a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend
index c9fb1fc0b895ffabcd609eed14424e7411856413..3eb0472b5c5bf3a4b1079e731875529f072da7d0 100644 (file)
--- a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend
+++ b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend
@@ -55,6 +55,7 @@ import static com.google.common.base.Preconditions.*
 import static org.opendaylight.controller.sal.connect.netconf.InventoryUtils.*
 
 import static extension org.opendaylight.controller.sal.connect.netconf.NetconfMapping.*
+import com.google.common.util.concurrent.Futures
 
 class NetconfDevice implements Provider, // 
 DataReader<InstanceIdentifier, CompositeNode>, //
@@ -127,12 +128,8 @@ AutoCloseable {
 
         val listener = new NetconfDeviceListener(this);
         val task = startClientTask(dispatcher, listener)
-        if (mountInstance != null) {
-            commitHandlerReg = mountInstance.registerCommitHandler(ROOT_PATH, this)
-        }
         return processingExecutor.submit(task) as Future<Void>;
 
-    //commitHandlerReg = mountInstance.registerCommitHandler(path,this);
     }
 
     def Optional<SchemaContext> getSchemaContext() {
@@ -162,11 +159,16 @@ AutoCloseable {
                 deviceContextProvider.createContextFromCapabilities(initialCapabilities);
                 if (mountInstance != null && schemaContext.isPresent) {
                     mountInstance.schemaContext = schemaContext.get();
+                    val operations = schemaContext.get().operations;
+                    for (rpc : operations) {
+                        mountInstance.addRpcImplementation(rpc.QName, this);
+                    }
                 }
                 updateDeviceState()
                 if (mountInstance != null && confReaderReg == null && operReaderReg == null) {
                     confReaderReg = mountInstance.registerConfigurationReader(ROOT_PATH, this);
                     operReaderReg = mountInstance.registerOperationalReader(ROOT_PATH, this);
+                    commitHandlerReg = mountInstance.registerCommitHandler(ROOT_PATH, this);
                 }
             } catch (Exception e) {
                 logger.error("Netconf client NOT started. ", e)
@@ -202,13 +204,13 @@ AutoCloseable {
 
     override readConfigurationData(InstanceIdentifier path) {
         val result = invokeRpc(NETCONF_GET_CONFIG_QNAME,
-            wrap(NETCONF_GET_CONFIG_QNAME, CONFIG_SOURCE_RUNNING, path.toFilterStructure()));
+            wrap(NETCONF_GET_CONFIG_QNAME, CONFIG_SOURCE_RUNNING, path.toFilterStructure())).get();
         val data = result.result.getFirstCompositeByName(NETCONF_DATA_QNAME);
         return data?.findNode(path) as CompositeNode;
     }
 
     override readOperationalData(InstanceIdentifier path) {
-        val result = invokeRpc(NETCONF_GET_QNAME, wrap(NETCONF_GET_QNAME, path.toFilterStructure()));
+        val result = invokeRpc(NETCONF_GET_QNAME, wrap(NETCONF_GET_QNAME, path.toFilterStructure())).get();
         val data = result.result.getFirstCompositeByName(NETCONF_DATA_QNAME);
         return data?.findNode(path) as CompositeNode;
     }
@@ -217,19 +219,18 @@ AutoCloseable {
         Collections.emptySet;
     }
 
-    def createSubscription(String streamName) {
-        val it = ImmutableCompositeNode.builder()
-        QName = NETCONF_CREATE_SUBSCRIPTION_QNAME
-        addLeaf("stream", streamName);
-        invokeRpc(QName, toInstance())
-    }
+//    def createSubscription(String streamName) {
+//        val it = ImmutableCompositeNode.builder()
+//        QName = NETCONF_CREATE_SUBSCRIPTION_QNAME
+//        addLeaf("stream", streamName);
+//        invokeRpc(QName, toInstance())
+//    }
 
     override invokeRpc(QName rpc, CompositeNode input) {
         try {
             val message = rpc.toRpcMessage(input,schemaContext);
             val result = sendMessageImpl(message, messegeRetryCount, messageTimeoutCount);
-            return result.toRpcResult(rpc, schemaContext);
-
+            return Futures.immediateFuture(result.toRpcResult(rpc, schemaContext));
         } catch (Exception e) {
             logger.error("Rpc was not processed correctly.", e)
             throw e;
@@ -341,7 +342,6 @@ AutoCloseable {
         operReaderReg?.close()
         client?.close()
     }
-
 }
 
 package class NetconfDeviceSchemaContextProvider {