Fixed advanced Netconf client functionality. 38/4038/2
authorTony Tkacik <ttkacik@cisco.com>
Mon, 6 Jan 2014 19:47:49 +0000 (20:47 +0100)
committerGerrit Code Review <gerrit@opendaylight.org>
Tue, 7 Jan 2014 03:37:46 +0000 (03:37 +0000)
  - Added two-stage model for remote device
     - Schema-unaware - allows only for baseline netconf functionality
     - Schema-aware - allows for deeper get operations
  - Performance fixes for message sending and retrieval
  - Fixed received notification upstreaming by proper reimplementation
    of NetconfClientSessionListener

Change-Id: Ib80bd49eae28c3f9602ca105503edcd3381bcf22
Signed-off-by: Tony Tkacik <ttkacik@cisco.com>
opendaylight/md-sal/sal-netconf-connector/pom.xml
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/config/yang/md/sal/connector/netconf/NetconfConnectorModule.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.xtend
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfMapping.xtend
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/XmlDocumentUtils.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/YangModelInputStreamAdapter.java [new file with mode: 0644]
opendaylight/md-sal/sal-netconf-connector/src/main/yang/odl-sal-netconf-connector-cfg.yang

index fe61356..2323b09 100644 (file)
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.opendaylight.controller</groupId>
+            <artifactId>sal-binding-broker-impl</artifactId>
+            <version>1.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+           <groupId>org.opendaylight.controller</groupId>
+           <artifactId>sal-binding-broker-impl</artifactId>
+           <version>1.0-SNAPSHOT</version>
+           <type>test-jar</type>
+           <scope>test</scope>
+        </dependency>
+        <dependency>
+        <groupId>org.opendaylight.controller</groupId>
+        <artifactId>ietf-netconf-monitoring</artifactId>
+        <version>0.2.3-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.opendaylight.yangtools.model</groupId>
             <artifactId>ietf-inet-types</artifactId>
index 55a1fbf..f5d592c 100644 (file)
@@ -12,8 +12,12 @@ package org.opendaylight.controller.config.yang.md.sal.connector.netconf;
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.concurrent.GlobalEventExecutor;
 
+import java.io.File;
+import java.io.InputStream;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import javax.net.ssl.SSLContext;
 
@@ -24,6 +28,10 @@ import org.opendaylight.controller.netconf.util.handler.ssh.authentication.Login
 import org.opendaylight.controller.sal.connect.netconf.NetconfDevice;
 import org.opendaylight.protocol.framework.ReconnectStrategy;
 import org.opendaylight.protocol.framework.TimedReconnectStrategy;
+import org.opendaylight.yangtools.yang.model.util.repo.AbstractCachingSchemaSourceProvider;
+import org.opendaylight.yangtools.yang.model.util.repo.FilesystemSchemaCachingProvider;
+import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProvider;
+import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProviders;
 import org.osgi.framework.BundleContext;
 
 import static com.google.common.base.Preconditions.*;
@@ -37,6 +45,8 @@ import com.google.common.net.InetAddresses;
 public final class NetconfConnectorModule extends org.opendaylight.controller.config.yang.md.sal.connector.netconf.AbstractNetconfConnectorModule
 {
 
+    private static ExecutorService GLOBAL_PROCESSING_EXECUTOR = null;
+    private static AbstractCachingSchemaSourceProvider<String, InputStream> GLOBAL_NETCONF_SOURCE_PROVIDER = null;
     private BundleContext bundleContext;
 
     public NetconfConnectorModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
@@ -74,31 +84,56 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         } else {
             addressValue = getAddress().getIpv6Address().getValue();
         }
-        
         */
         ReconnectStrategy strategy = new TimedReconnectStrategy(GlobalEventExecutor.INSTANCE, attemptMsTimeout, 1000, 1.0, null,
                 Long.valueOf(connectionAttempts), null);
         
-        
-        device.setStrategy(strategy);
+        device.setReconnectStrategy(strategy);
         
         InetAddress addr = InetAddresses.forString(addressValue);
         InetSocketAddress socketAddress = new InetSocketAddress(addr , getPort().intValue());
+
+        
+        device.setProcessingExecutor(getGlobalProcessingExecutor());
+        
         device.setSocketAddress(socketAddress);
+        device.setEventExecutor(getEventExecutorDependency());
+        device.setDispatcher(createDispatcher());
+        device.setSchemaSourceProvider(getGlobalNetconfSchemaProvider(bundleContext));
         
+        getDomRegistryDependency().registerProvider(device, bundleContext);
+        device.start();
+        return device;
+    }
+
+    private ExecutorService getGlobalProcessingExecutor() {
+        if(GLOBAL_PROCESSING_EXECUTOR == null) {
+            
+            GLOBAL_PROCESSING_EXECUTOR = Executors.newCachedThreadPool();
+            
+        }
+        return GLOBAL_PROCESSING_EXECUTOR;
+    }
+
+    private synchronized AbstractCachingSchemaSourceProvider<String, InputStream> getGlobalNetconfSchemaProvider(BundleContext bundleContext) {
+        if(GLOBAL_NETCONF_SOURCE_PROVIDER == null) {
+            String storageFile = "cache/schema";
+            File directory = bundleContext.getDataFile(storageFile);
+            SchemaSourceProvider<String> defaultProvider = SchemaSourceProviders.noopProvider();
+            GLOBAL_NETCONF_SOURCE_PROVIDER = FilesystemSchemaCachingProvider.createFromStringSourceProvider(defaultProvider, directory);
+        }
+        return GLOBAL_NETCONF_SOURCE_PROVIDER;
+    }
+
+    private NetconfClientDispatcher createDispatcher() {
         EventLoopGroup bossGroup = getBossThreadGroupDependency();
         EventLoopGroup workerGroup = getWorkerThreadGroupDependency();
-        NetconfClientDispatcher dispatcher = null;
         if(getTcpOnly()) {
-            dispatcher = new NetconfClientDispatcher( bossGroup, workerGroup);
+            return new NetconfClientDispatcher( bossGroup, workerGroup);
         } else {
             AuthenticationHandler authHandler = new LoginPassword(getUsername(),getPassword());
-            dispatcher = new NetconfSshClientDispatcher(authHandler , bossGroup, workerGroup);
+            return new NetconfSshClientDispatcher(authHandler , bossGroup, workerGroup);
         }
-        getDomRegistryDependency().registerProvider(device, bundleContext);
-        
-        device.start(dispatcher);
-        return device;
     }
 
     public void setBundleContext(BundleContext bundleContext) {
index 7c4bf5f..bfe352a 100644 (file)
@@ -25,13 +25,43 @@ import org.opendaylight.yangtools.yang.data.impl.CompositeNodeTOImpl
 import org.opendaylight.protocol.framework.ReconnectStrategy
 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler
 import org.opendaylight.controller.md.sal.common.api.data.DataModification
+import com.google.common.collect.FluentIterable
+import org.opendaylight.yangtools.yang.model.api.SchemaContext
+import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.NetconfState
+import org.opendaylight.yangtools.yang.parser.impl.YangParserImpl
+import java.io.InputStream
+import org.slf4j.LoggerFactory
+import org.slf4j.Logger
+import org.opendaylight.controller.netconf.client.AbstractNetconfClientNotifySessionListener
+import org.opendaylight.controller.netconf.client.NetconfClientSession
+import org.opendaylight.controller.netconf.api.NetconfMessage
+import io.netty.util.concurrent.EventExecutor
 
-class NetconfDevice implements 
-    Provider, // 
-    DataReader<InstanceIdentifier, CompositeNode>, //
-    DataCommitHandler<InstanceIdentifier, CompositeNode>, //
-    RpcImplementation, //
-    AutoCloseable {
+import java.util.Map
+import java.util.Set
+import com.google.common.collect.ImmutableMap
+
+import org.opendaylight.yangtools.yang.model.util.repo.AbstractCachingSchemaSourceProvider
+import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProvider
+import com.google.common.base.Optional
+import com.google.common.collect.ImmutableList
+import org.opendaylight.yangtools.yang.model.util.repo.SchemaSourceProviders
+import static com.google.common.base.Preconditions.*;
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Future
+import org.opendaylight.controller.netconf.client.NetconfClientSessionListener
+import io.netty.util.concurrent.Promise
+import org.opendaylight.controller.netconf.util.xml.XmlElement
+import org.opendaylight.controller.netconf.util.xml.XmlNetconfConstants
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.locks.ReentrantLock
+
+class NetconfDevice implements Provider, // 
+DataReader<InstanceIdentifier, CompositeNode>, //
+DataCommitHandler<InstanceIdentifier, CompositeNode>, //
+RpcImplementation, //
+AutoCloseable {
 
     var NetconfClient client;
 
@@ -41,35 +71,97 @@ class NetconfDevice implements
     @Property
     var MountProvisionInstance mountInstance;
 
+    @Property
+    var EventExecutor eventExecutor;
+
+    @Property
+    var ExecutorService processingExecutor;
+
     @Property
     var InstanceIdentifier path;
 
     @Property
-    var ReconnectStrategy strategy;
+    var ReconnectStrategy reconnectStrategy;
+
+    @Property
+    var AbstractCachingSchemaSourceProvider<String, InputStream> schemaSourceProvider;
+
+    private NetconfDeviceSchemaContextProvider schemaContextProvider
+
+    protected val Logger logger
 
     Registration<DataReader<InstanceIdentifier, CompositeNode>> operReaderReg
     Registration<DataReader<InstanceIdentifier, CompositeNode>> confReaderReg
     Registration<DataCommitHandler<InstanceIdentifier, CompositeNode>> commitHandlerReg
-    
+
     val String name
     MountProvisionService mountService
+
+    int messegeRetryCount = 5;
+
+    int messageTimeoutCount = 5 * 1000;
+
+    Set<QName> cachedCapabilities
+
+    @Property
+    var NetconfClientDispatcher dispatcher
     
-    
+    static val InstanceIdentifier ROOT_PATH = InstanceIdentifier.builder().toInstance();
+
     public new(String name) {
         this.name = name;
+        this.logger = LoggerFactory.getLogger(NetconfDevice.name + "#" + name);
         this.path = InstanceIdentifier.builder(INVENTORY_PATH).nodeWithKey(INVENTORY_NODE,
             Collections.singletonMap(INVENTORY_ID, name)).toInstance;
     }
 
-    def start(NetconfClientDispatcher dispatcher) {
-        client = NetconfClient.clientFor(name, socketAddress, strategy, dispatcher);
-        confReaderReg = mountInstance.registerConfigurationReader(path, this);
-        operReaderReg = mountInstance.registerOperationalReader(path, this);
-        //commitHandlerReg = mountInstance.registerCommitHandler(path,this);
+    def start() {
+        checkState(dispatcher != null, "Dispatcher must be set.");
+        checkState(schemaSourceProvider != null, "Schema Source Provider must be set.")
+        checkState(eventExecutor != null, "Event executor must be set.");
+
+        val listener = new NetconfDeviceListener(this,eventExecutor);
+        val task = startClientTask(dispatcher, listener)
+        if(mountInstance != null) {
+            confReaderReg = mountInstance.registerConfigurationReader(ROOT_PATH, this);
+            operReaderReg = mountInstance.registerOperationalReader(ROOT_PATH, this);
+        }
+        return processingExecutor.submit(task) as Future<Void>;
+
+    //commitHandlerReg = mountInstance.registerCommitHandler(path,this);
+    }
+
+    def Optional<SchemaContext> getSchemaContext() {
+        if (schemaContextProvider == null) {
+            return Optional.absent();
+        }
+        return schemaContextProvider.currentContext;
+    }
+
+    private def Runnable startClientTask(NetconfClientDispatcher dispatcher, NetconfDeviceListener listener) {
+        return [ |
+            logger.info("Starting Netconf Client on: {}", socketAddress);
+            client = NetconfClient.clientFor(name, socketAddress, reconnectStrategy, dispatcher, listener);
+            logger.debug("Initial capabilities {}", initialCapabilities);
+            var SchemaSourceProvider<String> delegate;
+            if (initialCapabilities.contains(NetconfMapping.IETF_NETCONF_MONITORING_MODULE)) {
+                delegate = new NetconfDeviceSchemaSourceProvider(this);
+            } else {
+                logger.info("Device does not support IETF Netconf Monitoring.", socketAddress);
+                delegate = SchemaSourceProviders.<String>noopProvider();
+            }
+            val sourceProvider = schemaSourceProvider.createInstanceFor(delegate);
+            schemaContextProvider = new NetconfDeviceSchemaContextProvider(this, sourceProvider);
+            schemaContextProvider.createContextFromCapabilities(initialCapabilities);
+            if (mountInstance != null && schemaContext.isPresent) {
+                mountInstance.schemaContext = schemaContext.get();
+            }
+        ]
     }
 
     override readConfigurationData(InstanceIdentifier path) {
-        val result = invokeRpc(NETCONF_GET_CONFIG_QNAME, wrap(NETCONF_GET_CONFIG_QNAME, CONFIG_SOURCE_RUNNING, path.toFilterStructure()));
+        val result = invokeRpc(NETCONF_GET_CONFIG_QNAME,
+            wrap(NETCONF_GET_CONFIG_QNAME, CONFIG_SOURCE_RUNNING, path.toFilterStructure()));
         val data = result.result.getFirstCompositeByName(NETCONF_DATA_QNAME);
         return data?.findNode(path) as CompositeNode;
     }
@@ -83,10 +175,17 @@ class NetconfDevice implements
     override getSupportedRpcs() {
         Collections.emptySet;
     }
+    
+    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) {
         val message = rpc.toRpcMessage(input);
-        val result = client.sendMessage(message);
+        val result = client.sendMessage(message, messegeRetryCount, messageTimeoutCount);
         return result.toRpcResult();
     }
 
@@ -96,30 +195,28 @@ class NetconfDevice implements
 
     override onSessionInitiated(ProviderSession session) {
         val dataBroker = session.getService(DataBrokerService);
-        
-        
-        
+
         val transaction = dataBroker.beginTransaction
-        if(transaction.operationalNodeNotExisting) {
-            transaction.putOperationalData(path,nodeWithId)
+        if (transaction.operationalNodeNotExisting) {
+            transaction.putOperationalData(path, nodeWithId)
         }
-        if(transaction.configurationNodeNotExisting) {
-            transaction.putConfigurationData(path,nodeWithId)
+        if (transaction.configurationNodeNotExisting) {
+            transaction.putConfigurationData(path, nodeWithId)
         }
         transaction.commit().get();
         mountService = session.getService(MountProvisionService);
-        mountInstance = mountService.createOrGetMountPoint(path);
+        mountInstance = mountService?.createOrGetMountPoint(path);
     }
-    
+
     def getNodeWithId() {
-        val id = new SimpleNodeTOImpl(INVENTORY_ID,null,name);
-        return new CompositeNodeTOImpl(INVENTORY_NODE,null,Collections.singletonList(id));
+        val id = new SimpleNodeTOImpl(INVENTORY_ID, null, name);
+        return new CompositeNodeTOImpl(INVENTORY_NODE, null, Collections.singletonList(id));
     }
-    
+
     def boolean configurationNodeNotExisting(DataModificationTransaction transaction) {
         return null === transaction.readConfigurationData(path);
     }
-    
+
     def boolean operationalNodeNotExisting(DataModificationTransaction transaction) {
         return null === transaction.readOperationalData(path);
     }
@@ -133,9 +230,9 @@ class NetconfDevice implements
             } else if (current instanceof CompositeNode) {
                 val currentComposite = (current as CompositeNode);
 
-                current = currentComposite.getFirstCompositeByName(arg.nodeType);
+                current = currentComposite.getFirstCompositeByName(arg.nodeType.withoutRevision());
                 if (current == null) {
-                    current = currentComposite.getFirstSimpleByName(arg.nodeType);
+                    current = currentComposite.getFirstSimpleByName(arg.nodeType.withoutRevision());
                 }
                 if (current == null) {
                     return null;
@@ -149,6 +246,25 @@ class NetconfDevice implements
         throw new UnsupportedOperationException("TODO: auto-generated method stub")
     }
 
+    def getInitialCapabilities() {
+        val capabilities = client?.capabilities;
+        if (capabilities == null) {
+            return null;
+        }
+        if (cachedCapabilities == null) {
+            cachedCapabilities = FluentIterable.from(capabilities).filter[
+                contains("?") && contains("module=") && contains("revision=")].transform [
+                val parts = split("\\?");
+                val namespace = parts.get(0);
+                val queryParams = FluentIterable.from(parts.get(1).split("&"));
+                val revision = queryParams.findFirst[startsWith("revision=")].replaceAll("revision=", "");
+                val moduleName = queryParams.findFirst[startsWith("module=")].replaceAll("module=", "");
+                return QName.create(namespace, revision, moduleName);
+            ].toSet();
+        }
+        return cachedCapabilities;
+    }
+
     override close() {
         confReaderReg?.close()
         operReaderReg?.close()
@@ -156,3 +272,165 @@ class NetconfDevice implements
     }
 
 }
+
+package class NetconfDeviceListener extends NetconfClientSessionListener {
+
+    val NetconfDevice device
+    val EventExecutor eventExecutor
+
+    new(NetconfDevice device,EventExecutor eventExecutor) {
+        this.device = device
+        this.eventExecutor = eventExecutor
+    }
+
+    var Promise<NetconfMessage> messagePromise;
+    val promiseLock = new ReentrantLock;
+    
+    override onMessage(NetconfClientSession session, NetconfMessage message) {
+        if (isNotification(message)) {
+            onNotification(session, message);
+        } else try {
+            promiseLock.lock
+            if (messagePromise != null) {
+                messagePromise.setSuccess(message);
+                messagePromise = null;
+            }
+        } finally {
+            promiseLock.unlock
+        }
+    }
+
+    /**
+     * Method intended to customize notification processing.
+     * 
+     * @param session
+     *            {@see
+     *            NetconfClientSessionListener#onMessage(NetconfClientSession,
+     *            NetconfMessage)}
+     * @param message
+     *            {@see
+     *            NetconfClientSessionListener#onMessage(NetconfClientSession,
+     *            NetconfMessage)}
+     */
+    def void onNotification(NetconfClientSession session, NetconfMessage message) {
+        device.logger.debug("Received NETCONF notification.",message);
+        val domNotification = message?.toCompositeNode?.notificationBody;
+        if(domNotification != null) {
+            device?.mountInstance?.publish(domNotification);
+        }
+    }
+    
+    private static def CompositeNode getNotificationBody(CompositeNode node) {
+        for(child : node.children) {
+            if(child instanceof CompositeNode) {
+                return child as CompositeNode;
+            }
+        }
+    }
+
+    override getLastMessage(int attempts, int attemptMsDelay) throws InterruptedException {
+        val promise = promiseReply();
+        val messageAvailable = promise.await(attempts + attemptMsDelay);
+        if (messageAvailable) {
+            try {
+                return promise.get();
+            } catch (ExecutionException e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        throw new IllegalStateException("Unsuccessful after " + attempts + " attempts.");
+
+    // throw new TimeoutException("Message was not received on time.");
+    }
+
+    def Promise<NetconfMessage> promiseReply() {
+        promiseLock.lock
+        try {
+        if (messagePromise == null) {
+            messagePromise = eventExecutor.newPromise();
+            return messagePromise;
+        }
+        return messagePromise;
+        } finally {
+            promiseLock.unlock
+        }
+    }
+
+    def boolean isNotification(NetconfMessage message) {
+        val xmle = XmlElement.fromDomDocument(message.getDocument());
+        return XmlNetconfConstants.NOTIFICATION_ELEMENT_NAME.equals(xmle.getName());
+    }
+}
+
+package class NetconfDeviceSchemaContextProvider {
+
+    @Property
+    val NetconfDevice device;
+
+    @Property
+    val SchemaSourceProvider<InputStream> sourceProvider;
+
+    @Property
+    var Optional<SchemaContext> currentContext;
+
+    new(NetconfDevice device, SchemaSourceProvider<InputStream> sourceProvider) {
+        _device = device
+        _sourceProvider = sourceProvider
+    }
+
+    def createContextFromCapabilities(Iterable<QName> capabilities) {
+
+        val modelsToParse = ImmutableMap.<QName, InputStream>builder();
+        for (cap : capabilities) {
+            val source = sourceProvider.getSchemaSource(cap.localName, Optional.fromNullable(cap.formattedRevision));
+            if (source.present) {
+                modelsToParse.put(cap, source.get());
+            }
+        }
+        val context = tryToCreateContext(modelsToParse.build);
+        currentContext = Optional.fromNullable(context);
+    }
+
+    def SchemaContext tryToCreateContext(Map<QName, InputStream> modelsToParse) {
+        val parser = new YangParserImpl();
+        try {
+            val models = parser.parseYangModelsFromStreams(ImmutableList.copyOf(modelsToParse.values));
+            val result = parser.resolveSchemaContext(models);
+            return result;
+        } catch (Exception e) {
+            device.logger.debug("Error occured during parsing YANG schemas", e);
+            return null;
+        }
+    }
+}
+
+package class NetconfDeviceSchemaSourceProvider implements SchemaSourceProvider<String> {
+
+    val NetconfDevice device;
+
+    new(NetconfDevice device) {
+        this.device = device;
+    }
+
+    override getSchemaSource(String moduleName, Optional<String> revision) {
+        val it = ImmutableCompositeNode.builder() //
+        setQName(QName::create(NetconfState.QNAME, "get-schema")) //
+        addLeaf("format", "yang")
+        addLeaf("identifier", moduleName)
+        if (revision.present) {
+            addLeaf("version", revision.get())
+        }
+
+        device.logger.info("Loading YANG schema source for {}:{}", moduleName, revision)
+        val schemaReply = device.invokeRpc(getQName(), toInstance());
+
+        if (schemaReply.successful) {
+            val schemaBody = schemaReply.result.getFirstSimpleByName(
+                QName::create(NetconfState.QNAME.namespace, null, "data"))?.value;
+            device.logger.info("YANG Schema successfully received for: {}:{}", moduleName, revision);
+            return Optional.of(schemaBody as String);
+        }
+        return Optional.absent();
+    }
+}
index 78f6d59..794b582 100644 (file)
@@ -20,51 +20,63 @@ import org.w3c.dom.Element
 import org.opendaylight.controller.sal.common.util.Rpcs
 import java.util.List
 import com.google.common.collect.ImmutableList
+import org.opendaylight.yangtools.yang.data.api.SimpleNode
+import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode
 
 class NetconfMapping {
 
     public static val NETCONF_URI = URI.create("urn:ietf:params:xml:ns:netconf:base:1.0")
-    public static val NETCONF_QNAME = new QName(NETCONF_URI,null,"netconf");
-    public static val NETCONF_RPC_QNAME = new QName(NETCONF_QNAME,"rpc");
-    public static val NETCONF_GET_QNAME = new QName(NETCONF_QNAME,"get");
-    public static val NETCONF_GET_CONFIG_QNAME = new QName(NETCONF_QNAME,"get-config");
-    public static val NETCONF_SOURCE_QNAME = new QName(NETCONF_QNAME,"source");
-    public static val NETCONF_RUNNING_QNAME = new QName(NETCONF_QNAME,"running");
-    public static val NETCONF_RPC_REPLY_QNAME = new QName(NETCONF_QNAME,"rpc-reply");
-    public static val NETCONF_OK_QNAME = new QName(NETCONF_QNAME,"ok");
-    public static val NETCONF_DATA_QNAME = new QName(NETCONF_QNAME,"data");
+    public static val NETCONF_MONITORING_URI = "urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring"
+    public static val NETCONF_NOTIFICATION_URI = URI.create("urn:ietf:params:xml:ns:netconf:notification:1.0")
     
-     static List<Node<?>> RUNNING = Collections.<Node<?>>singletonList(new SimpleNodeTOImpl(NETCONF_RUNNING_QNAME,null,null));
-    public static val CONFIG_SOURCE_RUNNING = new CompositeNodeTOImpl(NETCONF_SOURCE_QNAME,null,RUNNING);
-
-    static val messageId = new AtomicInteger(0);
     
-   
+    public static val NETCONF_QNAME = QName.create(NETCONF_URI, null, "netconf");
+    public static val NETCONF_RPC_QNAME = QName.create(NETCONF_QNAME, "rpc");
+    public static val NETCONF_GET_QNAME = QName.create(NETCONF_QNAME, "get");
+    public static val NETCONF_FILTER_QNAME = QName.create(NETCONF_QNAME, "filter");
+    public static val NETCONF_TYPE_QNAME = QName.create(NETCONF_QNAME, "type");
+    public static val NETCONF_GET_CONFIG_QNAME = QName.create(NETCONF_QNAME, "get-config");
+    public static val NETCONF_SOURCE_QNAME = QName.create(NETCONF_QNAME, "source");
+    public static val NETCONF_RUNNING_QNAME = QName.create(NETCONF_QNAME, "running");
+    public static val NETCONF_RPC_REPLY_QNAME = QName.create(NETCONF_QNAME, "rpc-reply");
+    public static val NETCONF_OK_QNAME = QName.create(NETCONF_QNAME, "ok");
+    public static val NETCONF_DATA_QNAME = QName.create(NETCONF_QNAME, "data");
+    public static val NETCONF_CREATE_SUBSCRIPTION_QNAME = QName.create(NETCONF_NOTIFICATION_URI,null,"create-subscription");
+    public static val NETCONF_CANCEL_SUBSCRIPTION_QNAME = QName.create(NETCONF_NOTIFICATION_URI,null,"cancel-subscription");
+    public static val IETF_NETCONF_MONITORING_MODULE = QName.create(NETCONF_MONITORING_URI, "2010-10-04","ietf-netconf-monitoring");
 
+    static List<Node<?>> RUNNING = Collections.<Node<?>>singletonList(
+        new SimpleNodeTOImpl(NETCONF_RUNNING_QNAME, null, null));
+    public static val CONFIG_SOURCE_RUNNING = new CompositeNodeTOImpl(NETCONF_SOURCE_QNAME, null, RUNNING);
 
+    static val messageId = new AtomicInteger(0);
 
     static def Node<?> toFilterStructure(InstanceIdentifier identifier) {
         var Node<?> previous = null;
-        for (component : identifier.path.reverse) {
+        if(identifier.path.empty) {
+            return null;
+        }
+        
+        for (component : identifier.path.reverseView) {
             val Node<?> current = component.toNode(previous);
             previous = current;
         }
-        return previous;
+        return filter("subtree",previous);
     }
-    
+
     static def dispatch Node<?> toNode(NodeIdentifierWithPredicates argument, Node<?> node) {
         val list = new ArrayList<Node<?>>();
-        forarg : argument.keyValues.entrySet) {
-            list.add = new SimpleNodeTOImpl(arg.key,null,arg.value);
+        for (arg : argument.keyValues.entrySet) {
+            list.add = new SimpleNodeTOImpl(arg.key, null, arg.value);
         }
-        return new CompositeNodeTOImpl(argument.nodeType,null,list)
+        return new CompositeNodeTOImpl(argument.nodeType, null, list)
     }
-    
+
     static def dispatch Node<?> toNode(PathArgument argument, Node<?> node) {
-        if(node != null) {
-            return new CompositeNodeTOImpl(argument.nodeType,null,Collections.singletonList(node));
+        if (node != null) {
+            return new CompositeNodeTOImpl(argument.nodeType, null, Collections.singletonList(node));
         } else {
-            return new SimpleNodeTOImpl(argument.nodeType,null,null);
+            return new SimpleNodeTOImpl(argument.nodeType, null, null);
         }
     }
 
@@ -73,40 +85,62 @@ class NetconfMapping {
     }
 
     static def NetconfMessage toRpcMessage(QName rpc, CompositeNode node) {
-        val rpcPayload = wrap(NETCONF_RPC_QNAME,node);
+        val rpcPayload = wrap(NETCONF_RPC_QNAME, flattenInput(node));
         val w3cPayload = NodeUtils.buildShadowDomTree(rpcPayload);
-        w3cPayload.documentElement.setAttribute("message-id","m-"+ messageId.andIncrement);
+        w3cPayload.documentElement.setAttribute("message-id", "m-" + messageId.andIncrement);
         return new NetconfMessage(w3cPayload);
     }
+    
+    def static flattenInput(CompositeNode node) {
+        val inputQName = QName.create(node.nodeType,"input");
+        val input = node.getFirstCompositeByName(inputQName);
+        if(input == null) return node;
+        if(input instanceof CompositeNode) {
+            
+            val nodes = ImmutableList.builder() //
+                .addAll(input.children) //
+                .addAll(node.children.filter[nodeType != inputQName]) //
+                .build()
+            return ImmutableCompositeNode.create(node.nodeType,nodes);
+        } 
+        
+    }
 
     static def RpcResult<CompositeNode> toRpcResult(NetconfMessage message) {
         val rawRpc = message.document.toCompositeNode() as CompositeNode;
+
         //rawRpc.
-        
-        return Rpcs.getRpcResult(true,rawRpc,Collections.emptySet());
+        return Rpcs.getRpcResult(true, rawRpc, Collections.emptySet());
     }
-    
-    
-    static def wrap(QName name,Node<?> node) {
-        if(node != null) {
-            return new CompositeNodeTOImpl(name,null,Collections.singletonList(node));
-        }
-        else {
-            return new CompositeNodeTOImpl(name,null,Collections.emptyList());
+
+    static def wrap(QName name, Node<?> node) {
+        if (node != null) {
+            return new CompositeNodeTOImpl(name, null, Collections.singletonList(node));
+        } else {
+            return new CompositeNodeTOImpl(name, null, Collections.emptyList());
         }
     }
-    
-        static def wrap(QName name,Node<?> additional,Node<?> node) {
-        if(node != null) {
-            return new CompositeNodeTOImpl(name,null,ImmutableList.of(additional,node));
+
+    static def wrap(QName name, Node<?> additional, Node<?> node) {
+        if (node != null) {
+            return new CompositeNodeTOImpl(name, null, ImmutableList.of(additional, node));
+        } else {
+            return new CompositeNodeTOImpl(name, null, ImmutableList.of(additional));
         }
-        else {
-            return new CompositeNodeTOImpl(name,null,ImmutableList.of(additional));
+    }
+
+    static def filter(String type, Node<?> node) {
+        val it = ImmutableCompositeNode.builder(); //
+        setQName(NETCONF_FILTER_QNAME);
+        setAttribute(NETCONF_TYPE_QNAME,type);
+        if (node != null) {
+            return add(node).toInstance();
+        } else {
+            return toInstance();
         }
     }
-    
-    
+
     public static def Node<?> toCompositeNode(Document document) {
-        return XmlDocumentUtils.toCompositeNode(document) as Node<?>
+        return XmlDocumentUtils.toNode(document) as Node<?>
     }
 }
index 3f6b4e1..e151fca 100644 (file)
@@ -13,10 +13,12 @@ import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 import org.w3c.dom.NodeList;
 
+import com.google.common.base.Strings;
+
 public class XmlDocumentUtils {
 
-    public static CompositeNode toCompositeNode(Document doc) {
-        return (CompositeNode) toCompositeNode(doc.getDocumentElement());
+    public static Node<?> toNode(Document doc) {
+        return toCompositeNode(doc.getDocumentElement());
     }
 
     private static Node<?> toCompositeNode(Element element) {
@@ -29,7 +31,7 @@ public class XmlDocumentUtils {
 
         List<Node<?>> values = new ArrayList<>();
         NodeList nodes = element.getChildNodes();
-        boolean isSimpleObject = false;
+        boolean isSimpleObject = true;
         String value = null;
         for (int i = 0; i < nodes.getLength(); i++) {
             org.w3c.dom.Node child = nodes.item(i);
@@ -37,11 +39,10 @@ public class XmlDocumentUtils {
                 isSimpleObject = false;
                 values.add(toCompositeNode((Element) child));
             }
-            if (!isSimpleObject && child instanceof org.w3c.dom.Text) {
+            if (isSimpleObject && child instanceof org.w3c.dom.Text) {
                 value = element.getTextContent();
-                if (value.matches(".*\\w.*")) {
+                if (!Strings.isNullOrEmpty(value)) {
                     isSimpleObject = true;
-                    break;
                 }
             }
         }
diff --git a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/YangModelInputStreamAdapter.java b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/YangModelInputStreamAdapter.java
new file mode 100644 (file)
index 0000000..0c0070c
--- /dev/null
@@ -0,0 +1,88 @@
+package org.opendaylight.controller.sal.connect.netconf;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.StringBufferInputStream;
+import java.io.StringReader;
+
+import org.opendaylight.yangtools.concepts.Delegator;
+import org.opendaylight.yangtools.yang.common.QName;
+
+/**
+ * 
+ *
+ */
+public class YangModelInputStreamAdapter extends InputStream implements Delegator<InputStream> {
+
+    final String source;
+    final QName moduleIdentifier;
+    final InputStream delegate;
+    
+    
+    
+    private YangModelInputStreamAdapter(String source, QName moduleIdentifier, InputStream delegate) {
+        super();
+        this.source = source;
+        this.moduleIdentifier = moduleIdentifier;
+        this.delegate = delegate;
+    }
+
+    public int read() throws IOException {
+        return delegate.read();
+    }
+
+    public int hashCode() {
+        return delegate.hashCode();
+    }
+
+    public int read(byte[] b) throws IOException {
+        return delegate.read(b);
+    }
+
+    public boolean equals(Object obj) {
+        return delegate.equals(obj);
+    }
+
+    public int read(byte[] b, int off, int len) throws IOException {
+        return delegate.read(b, off, len);
+    }
+
+    public long skip(long n) throws IOException {
+        return delegate.skip(n);
+    }
+
+    public int available() throws IOException {
+        return delegate.available();
+    }
+
+    public void close() throws IOException {
+        delegate.close();
+    }
+
+    public void mark(int readlimit) {
+        delegate.mark(readlimit);
+    }
+
+    public void reset() throws IOException {
+        delegate.reset();
+    }
+
+    public boolean markSupported() {
+        return delegate.markSupported();
+    }
+
+    @Override
+    public InputStream getDelegate() {
+        return delegate;
+    }
+
+    @Override
+    public String toString() {
+        return "YangModelInputStreamAdapter [moduleIdentifier=" + moduleIdentifier + ", delegate=" + delegate + "]";
+    }
+
+    public static YangModelInputStreamAdapter create(QName name, String module) {
+        InputStream stringInput = new StringBufferInputStream(module);
+        return new YangModelInputStreamAdapter(null, name, stringInput );
+    }
+}
index 9238514..2fae7ee 100644 (file)
@@ -81,6 +81,14 @@ module odl-sal-netconf-connector-cfg {
                     }
                 }
             }
+
+            container event-executor {
+                uses config:service-ref {
+                    refine type {
+                        config:required-identity netty:netty-event-executor;
+                    }
+                }
+            }
         }
     }
 }
\ No newline at end of file