X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fprotocol_plugins%2Fopenflow%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fprotocol_plugin%2Fopenflow%2Finternal%2FDataPacketMuxDemux.java;h=7a01d3bd19bce2060a6749bb017e08026a84aaea;hb=d4b6addab23cf24f20cd7969a7f1d800fda2bac1;hp=7614a4d128eae867ed4378f78e346da447c4d081;hpb=600e7a7bf7a59029e85a4811d835133d77b6d8d2;p=controller.git

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java
index 7614a4d128..7a01d3bd19 100644
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java
@@ -32,6 +32,7 @@ import org.openflow.protocol.action.OFActionOutput;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.ConstructionException;
 import org.opendaylight.controller.sal.core.ContainerFlow;
 import org.opendaylight.controller.sal.core.IContainerListener;
@@ -60,6 +61,7 @@ public class DataPacketMuxDemux implements IContainerListener,
     private ConcurrentMap<String, List<ContainerFlow>> container2FlowSpecs = new ConcurrentHashMap<String, List<ContainerFlow>>();
     // Track local data packet listener
     private List<IDataPacketListen> iDataPacketListen = new CopyOnWriteArrayList<IDataPacketListen>();
+    private IPluginOutConnectionService connectionOutService;
 
     void setIDataPacketListen(IDataPacketListen s) {
         if (this.iDataPacketListen != null) {
@@ -128,10 +130,20 @@ public class DataPacketMuxDemux implements IContainerListener,
         }
     }
 
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
+
     /**
      * Function called by the dependency manager when all the required
      * dependencies are satisfied
-     * 
+     *
      */
     void init() {
         this.controller.addMessageListener(OFType.PACKET_IN, this);
@@ -141,7 +153,7 @@ public class DataPacketMuxDemux implements IContainerListener,
      * Function called by the dependency manager when at least one dependency
      * become unsatisfied or when the component is shutting down because for
      * example bundle is being stopped.
-     * 
+     *
      */
     void destroy() {
         this.controller.removeMessageListener(OFType.PACKET_IN, this);
@@ -164,9 +176,21 @@ public class DataPacketMuxDemux implements IContainerListener,
                     new Object[] { sw, msg, this.pluginOutDataPacketServices });
             return;
         }
+
+        Long ofSwitchID = Long.valueOf(sw.getId());
+        try {
+            Node n = new Node(Node.NodeIDType.OPENFLOW, ofSwitchID);
+            if (!connectionOutService.isLocal(n)) {
+                logger.debug("Connection service refused DataPacketMuxDemux receive {} {}", sw, msg);
+                return;
+            }
+        }
+        catch (Exception e) {
+            return;
+        }
+
         if (msg instanceof OFPacketIn) {
             OFPacketIn ofPacket = (OFPacketIn) msg;
-            Long ofSwitchID = Long.valueOf(sw.getId());
             Short ofPortID = Short.valueOf(ofPacket.getInPort());
 
             try {
@@ -258,6 +282,12 @@ public class DataPacketMuxDemux implements IContainerListener,
             return;
         }
 
+        if (!connectionOutService.isLocal(outPort.getNode())) {
+            logger.debug("data packets will not be sent to {} in a non-master controller", outPort.toString());
+            return;
+        }
+
+
         if (!outPort.getType().equals(
                 NodeConnector.NodeConnectorIDType.OPENFLOW)) {
             // The output Port is not of type OpenFlow
@@ -290,7 +320,8 @@ public class DataPacketMuxDemux implements IContainerListener,
                 + data.length);
         po.setPacketData(data);
 
-        sw.asyncSend(po);
+        // send PACKET_OUT at high priority
+        sw.asyncFastSend(po);
         logger.trace("Transmitted a frame of size: {}", data.length);
     }