Bug 933: Changed Topology Compatibility to use DataChangeEvent

Topology Compatibility adapter used commit handler to receive
change events and also reading from datastore, which may leaded
to some desynchronization issues. Changed topology Compatibility
to data change listener, which ensures data store is up to date,
when read occurs.

Change-Id: I052c9c79d99bdc8ba70e32c4421073c581901adc
Signed-off-by: Tony Tkacik <ttkacik@cisco.com>

diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyCommitHandler.java b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyCommitHandler.java
new file mode 100644 (file)
index 0000000..a35c3ed
--- /dev/null
+++ b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyCommitHandler.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.sal.compatibility.topology;
+
+import static org.opendaylight.controller.sal.compatibility.topology.TopologyMapping.toAdEdge;
+import static org.opendaylight.controller.sal.compatibility.topology.TopologyMapping.toTopoEdgeUpdate;
+
+import java.util.Map.Entry;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.opendaylight.controller.md.sal.binding.util.TypeSafeDataReader;
+import org.opendaylight.controller.md.sal.common.api.data.DataChangeEvent;
+import org.opendaylight.controller.sal.binding.api.data.DataChangeListener;
+import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
+import org.opendaylight.controller.sal.core.UpdateType;
+import org.opendaylight.controller.sal.topology.IPluginOutTopologyService;
+import org.opendaylight.controller.sal.topology.TopoEdgeUpdate;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Link;
+import org.opendaylight.yangtools.yang.binding.DataObject;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TopologyCommitHandler implements DataChangeListener {
+    private static final Logger LOG = LoggerFactory.getLogger(TopologyCommitHandler.class);
+
+    private IPluginOutTopologyService topologyPublisher;
+
+    private final DataProviderService dataService;
+
+    public TopologyCommitHandler(final DataProviderService dataService, final IPluginOutTopologyService topologyPub) {
+        this.topologyPublisher = topologyPub;
+        this.dataService = dataService;
+    }
+
+    @Override
+    public void onDataChanged(final DataChangeEvent<InstanceIdentifier<?>, DataObject> modification) {
+        CopyOnWriteArrayList<TopoEdgeUpdate> msg = new CopyOnWriteArrayList<TopoEdgeUpdate>();
+        try {
+            TypeSafeDataReader reader = TypeSafeDataReader.forReader(dataService);
+            InstanceIdentifier<Topology> topologyPath = InstanceIdentifier.builder(NetworkTopology.class)
+                    .child(Topology.class, new TopologyKey(new TopologyId("flow:1"))).build();
+            Topology topology = reader.readOperationalData(topologyPath);
+
+            for (Entry<InstanceIdentifier<? extends DataObject>, DataObject> entry : modification
+                    .getCreatedOperationalData().entrySet()) {
+                if (entry.getValue() instanceof Link
+                        && modification.getCreatedOperationalData().containsKey(entry.getKey())) {
+                    msg.add(toTopoEdgeUpdate(toAdEdge((Link) entry.getValue(), topology), UpdateType.ADDED, reader));
+                }
+            }
+
+            for (Entry<InstanceIdentifier<? extends DataObject>, DataObject> entry : modification
+                    .getUpdatedOperationalData().entrySet()) {
+                if (entry.getValue() instanceof Link) {
+                    msg.add(toTopoEdgeUpdate(toAdEdge((Link) entry.getValue(), topology), UpdateType.CHANGED, reader));
+                }
+            }
+            for (InstanceIdentifier<? extends DataObject> path : modification.getRemovedOperationalData()) {
+                if (path.getTargetType() == Link.class) {
+                    Link link = (Link) modification.getOriginalOperationalData().get(path);
+                    msg.add(toTopoEdgeUpdate(toAdEdge(link, topology), UpdateType.CHANGED, reader));
+                }
+
+            }
+
+            if (topologyPublisher != null && msg != null && !msg.isEmpty()) {
+                topologyPublisher.edgeUpdate(msg);
+            }
+
+        } catch (Exception e) {
+            LOG.error("Exception caught", e);
+        }
+    }
+
+    protected IPluginOutTopologyService getTopologyPublisher() {
+        return topologyPublisher;
+    }
+
+    protected void setTopologyPublisher(final IPluginOutTopologyService topologyPublisher) {
+        this.topologyPublisher = topologyPublisher;
+    }
+
+}

diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyCommitHandler.xtend b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyCommitHandler.xtend
deleted file mode 100644 (file)
index fcf86f2..0000000
--- a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyCommitHandler.xtend
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.compatibility.topology
-
-import com.google.common.collect.FluentIterable
-import java.util.concurrent.CopyOnWriteArrayList
-import org.opendaylight.controller.md.sal.binding.util.TypeSafeDataReader
-import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler
-import org.opendaylight.controller.md.sal.common.api.data.DataModification
-import org.opendaylight.controller.sal.binding.api.data.DataProviderService
-import org.opendaylight.controller.sal.core.UpdateType
-import org.opendaylight.controller.sal.topology.IPluginOutTopologyService
-import org.opendaylight.controller.sal.topology.TopoEdgeUpdate
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Link
-import org.opendaylight.yangtools.yang.binding.DataObject
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier
-import static extension org.opendaylight.controller.sal.compatibility.topology.TopologyMapping.*
-import org.slf4j.LoggerFactory
-
-class TopologyCommitHandler implements DataCommitHandler<InstanceIdentifier<? extends DataObject>, DataObject> {
-    static val LOG = LoggerFactory.getLogger(TopologyCommitHandler);
-    @Property
-    IPluginOutTopologyService topologyPublisher;
-    
-    @Property
-    DataProviderService dataService;
-    
-    new(DataProviderService dataService) {
-        _topologyPublisher = topologyPublisher
-        _dataService = dataService
-    }
-    
-    override requestCommit(DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification) {
-        val msg = new CopyOnWriteArrayList<TopoEdgeUpdate>()
-        try {
-            val reader = TypeSafeDataReader.forReader(dataService)
-            val topologyPath = InstanceIdentifier.builder(NetworkTopology).child(Topology, new TopologyKey(new TopologyId("flow:1"))).toInstance
-            val topology = reader.readOperationalData(topologyPath)
-            val adds = FluentIterable.from(modification.createdOperationalData.entrySet)
-                .filter[value instanceof Link]
-                .transform[(value as Link).toAdEdge(topology).toTopoEdgeUpdate(UpdateType.ADDED,reader)]
-                .toList
-            val updates = FluentIterable.from(modification.updatedOperationalData.entrySet)
-                .filter[!modification.createdOperationalData.containsKey(key) && (value instanceof Link)]
-                .transform[(value as Link).toAdEdge(topology).toTopoEdgeUpdate(UpdateType.ADDED,reader)] // Evidently the ADSAL does not expect edge 'CHANGED"
-                .toList
-            val removes = FluentIterable.from(modification.removedOperationalData)
-                .transform[reader.readOperationalData(it as InstanceIdentifier<DataObject>)]
-                .filter[it instanceof Link]
-                .transform[(it as Link).toAdEdge(topology).toTopoEdgeUpdate(UpdateType.REMOVED,reader)]
-                .toList
-            msg.addAll(adds)
-            msg.addAll(updates)
-            msg.addAll(removes)
-         } catch (Exception e) {
-            LOG.error("Exception caught",e)
-         }
-        return new TopologyTransaction(modification,topologyPublisher,dataService,msg)
-    }
-}

diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyProvider.xtend b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyProvider.xtend
index 4aef75d9916565dc20c3c18a87149669116f4753..21f2b35f40a82c846df07e7ce05fab2ae540029c 100644 (file)
--- a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyProvider.xtend
+++ b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyProvider.xtend
@@ -15,10 +15,10 @@ import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey
 import org.opendaylight.yangtools.yang.binding.DataObject
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier
-import org.opendaylight.yangtools.concepts.Registration
-import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Link
 import org.slf4j.LoggerFactory
+import org.opendaylight.yangtools.concepts.ListenerRegistration
+import org.opendaylight.controller.sal.binding.api.data.DataChangeListener
 
 class TopologyProvider implements AutoCloseable{
     static val LOG = LoggerFactory.getLogger(TopologyProvider);
@@ -30,8 +30,9 @@ class TopologyProvider implements AutoCloseable{
     @Property
     DataProviderService dataService;
     
-    Registration<DataCommitHandler<InstanceIdentifier<? extends DataObject>,DataObject>> commitHandlerRegistration;
-
+    ListenerRegistration<DataChangeListener> listenerRegistration
+    
+    
     def void start() {
 
     }
@@ -40,18 +41,17 @@ class TopologyProvider implements AutoCloseable{
             LOG.error("dataService not set");
             return;
         }
-        commitHandler = new TopologyCommitHandler(dataService)
-        commitHandler.setTopologyPublisher(topologyPublisher)
+        commitHandler = new TopologyCommitHandler(dataService,topologyPublisher);
         val InstanceIdentifier<? extends DataObject> path = InstanceIdentifier.builder(NetworkTopology)
             .child(Topology,new TopologyKey(new TopologyId("flow:1")))
             .child(Link)
             .toInstance();
-        commitHandlerRegistration = dataService.registerCommitHandler(path,commitHandler);
+        listenerRegistration = dataService.registerDataChangeListener(path,commitHandler);
         LOG.info("TopologyProvider started")
     }
     
     override close() throws Exception {
-        commitHandlerRegistration.close
+        listenerRegistration.close
     }
     
     def setTopologyPublisher(IPluginOutTopologyService topologyPublisher) {

diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyTransaction.xtend b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyTransaction.xtend
deleted file mode 100644 (file)
index 4de78ca..0000000
--- a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/topology/TopologyTransaction.xtend
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.compatibility.topology
-
-import java.util.Collections
-import java.util.List
-import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction
-import org.opendaylight.controller.md.sal.common.api.data.DataModification
-import org.opendaylight.controller.sal.binding.api.data.DataProviderService
-import org.opendaylight.controller.sal.topology.IPluginOutTopologyService
-import org.opendaylight.controller.sal.topology.TopoEdgeUpdate
-import org.opendaylight.yangtools.yang.binding.DataObject
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier
-import org.opendaylight.yangtools.yang.common.RpcResult
-import org.slf4j.LoggerFactory
-
-class TopologyTransaction implements DataCommitTransaction<InstanceIdentifier<?extends DataObject>, DataObject> {
-    static val LOG = LoggerFactory.getLogger(TopologyTransaction);
-    @Property
-    val DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification;
-    
-    @Property
-    IPluginOutTopologyService topologyPublisher;
-    
-    @Property
-    DataProviderService dataService;
-    @Property
-    List<TopoEdgeUpdate> edgeUpdates;
-    
-    new(DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification,IPluginOutTopologyService topologyPublisher,
-        DataProviderService dataService,List<TopoEdgeUpdate> edgeUpdates) {
-        _modification = modification;
-        _topologyPublisher = topologyPublisher
-        _dataService = dataService
-        _edgeUpdates = edgeUpdates
-    }
-    override finish() throws IllegalStateException {
-        
-        if(topologyPublisher != null && _edgeUpdates != null && !edgeUpdates.empty) {
-            topologyPublisher.edgeUpdate(edgeUpdates)
-        }
-         
-         return new RpcResultTo()
-    }
-    
-    override getModification() {
-        return _modification;
-    }
-    
-    override rollback() throws IllegalStateException {
-        // NOOP
-    }
-}
-class RpcResultTo implements RpcResult<Void> {
-    
-    override getErrors() {
-        return Collections.emptySet
-    }
-    
-    override getResult() {
-        return null;
-    }
-    
-    override isSuccessful() {
-        return true;
-    }
-    
-}