BUG-3219: Handle EchoRequests directly in ConnectionAdapter

[openflowjava.git] / openflow-protocol-impl / src / main / java / org / opendaylight / openflowjava / protocol / impl / core / connection / ConnectionAdapterImpl.java

diff --git a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java
index c2854bec95d35e7bcb5ca718124375ae2b1d08e8..cf2693c80276a4c587b3bc0f45d47af17d8078e5 100644 (file)
--- a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java
+++ b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java
@@ -9,16 +9,26 @@
 
 package org.opendaylight.openflowjava.protocol.impl.core.connection;
 
+import com.google.common.base.Preconditions;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalCause;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.util.concurrent.GenericFutureListener;
-
 import java.net.InetSocketAddress;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
-
 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
+import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandler;
+import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
+import org.opendaylight.openflowjava.protocol.impl.core.OFVersionDetector;
+import org.opendaylight.openflowjava.protocol.impl.core.PipelineHandlers;
 import org.opendaylight.openflowjava.statistics.CounterEventTypes;
 import org.opendaylight.openflowjava.statistics.StatisticsCounters;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
@@ -66,15 +76,6 @@ import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.base.Preconditions;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.RemovalCause;
-import com.google.common.cache.RemovalListener;
-import com.google.common.cache.RemovalNotification;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-
 /**
  * Handles messages (notifications + rpcs) and connections
  * @author mirehak
@@ -115,8 +116,11 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
     private ConnectionReadyListener connectionReadyListener;
     private OpenflowProtocolListener messageListener;
     private SystemNotificationsListener systemListener;
+    private OutboundQueueManager<?> outputManager;
     private boolean disconnectOccured = false;
-    private StatisticsCounters statisticsCounters;
+    private final StatisticsCounters statisticsCounters;
+    private OFVersionDetector versionDetector;
+    private final InetSocketAddress address;
 
     /**
      * default ctor
@@ -131,8 +135,10 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
                 .removalListener(REMOVAL_LISTENER).build();
         this.channel = Preconditions.checkNotNull(channel);
         this.output = new ChannelOutboundQueue(channel, DEFAULT_QUEUE_DEPTH, address);
+        this.address = address;
         channel.pipeline().addLast(output);
         statisticsCounters = StatisticsCounters.getInstance();
+
         LOG.debug("ConnectionAdapter created");
     }
 
@@ -263,7 +269,7 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
 
     @Override
     public void consume(final DataObject message) {
-        LOG.debug("ConsumeIntern msg");
+        LOG.debug("ConsumeIntern msg on {}", channel);
         if (disconnectOccured ) {
             return;
         }
@@ -278,12 +284,22 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
                 systemListener.onSwitchIdleEvent((SwitchIdleEvent) message);
                 // OpenFlow messages
             } else if (message instanceof EchoRequestMessage) {
-                messageListener.onEchoRequestMessage((EchoRequestMessage) message);
+                if (outputManager != null) {
+                    outputManager.onEchoRequest((EchoRequestMessage) message);
+                } else {
+                    messageListener.onEchoRequestMessage((EchoRequestMessage) message);
+                }
                 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
             } else if (message instanceof ErrorMessage) {
-                messageListener.onErrorMessage((ErrorMessage) message);
+                // Send only unmatched errors
+                if (outputManager == null || !outputManager.onMessage((OfHeader) message)) {
+                    messageListener.onErrorMessage((ErrorMessage) message);
+                }
                 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
             } else if (message instanceof ExperimenterMessage) {
+                if (outputManager != null) {
+                    outputManager.onMessage((OfHeader) message);
+                }
                 messageListener.onExperimenterMessage((ExperimenterMessage) message);
                 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
             } else if (message instanceof FlowRemovedMessage) {
@@ -294,6 +310,9 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
                 messageListener.onHelloMessage((HelloMessage) message);
                 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
             } else if (message instanceof MultipartReplyMessage) {
+                if (outputManager != null) {
+                    outputManager.onMessage((OfHeader) message);
+                }
                 messageListener.onMultipartReplyMessage((MultipartReplyMessage) message);
                 statisticsCounters.incrementCounter(CounterEventTypes.US_MESSAGE_PASS);
             } else if (message instanceof PacketInMessage) {
@@ -305,9 +324,10 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
             } else {
                 LOG.warn("message listening not supported for type: {}", message.getClass());
             }
-        } else {
-            if (message instanceof OfHeader) {
-                LOG.debug("OFheader msg received");
+        } else if (message instanceof OfHeader) {
+            LOG.debug("OFheader msg received");
+
+            if (outputManager == null || !outputManager.onMessage((OfHeader) message)) {
                 RpcResponseKey key = createRpcResponseKey((OfHeader) message);
                 final ResponseExpectedRpcListener<?> listener = findRpcResponse(key);
                 if (listener != null) {
@@ -319,9 +339,9 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
                 } else {
                     LOG.warn("received unexpected rpc response: {}", key);
                 }
-            } else {
-                LOG.warn("message listening not supported for type: {}", message.getClass());
             }
+        } else {
+            LOG.warn("message listening not supported for type: {}", message.getClass());
         }
     }
 
@@ -443,6 +463,9 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
 
     @Override
     public void fireConnectionReadyNotification() {
+        versionDetector = (OFVersionDetector) channel.pipeline().get(PipelineHandlers.OF_VERSION_DETECTOR.name());
+        Preconditions.checkState(versionDetector != null);
+
         new Thread(new Runnable() {
             @Override
             public void run() {
@@ -466,7 +489,46 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
      * Used only for testing purposes
      * @param cache
      */
-    public void setResponseCache(Cache<RpcResponseKey, ResponseExpectedRpcListener<?>> cache) {
+    public void setResponseCache(final Cache<RpcResponseKey, ResponseExpectedRpcListener<?>> cache) {
         this.responseCache = cache;
     }
+
+       @Override
+    public boolean isAutoRead() {
+       return channel.config().isAutoRead();
+    }
+
+       @Override
+    public void setAutoRead(final boolean autoRead) {
+       channel.config().setAutoRead(autoRead);
+    }
+
+    @Override
+    public <T extends OutboundQueueHandler> OutboundQueueHandlerRegistration<T> registerOutboundQueueHandler(
+            final T handler, final int maxQueueDepth, final long maxBarrierNanos) {
+        Preconditions.checkState(outputManager == null, "Manager %s already registered", outputManager);
+
+        final OutboundQueueManager<T> ret = new OutboundQueueManager<>(this, address, handler, maxQueueDepth, maxBarrierNanos);
+        outputManager = ret;
+        channel.pipeline().addLast(outputManager);
+
+        return new OutboundQueueHandlerRegistrationImpl<T>(handler) {
+            @Override
+            protected void removeRegistration() {
+                outputManager.close();
+                channel.pipeline().remove(outputManager);
+                outputManager = null;
+            }
+        };
+    }
+
+    Channel getChannel() {
+        return channel;
+    }
+
+    @Override
+    public void setPacketInFiltering(final boolean enabled) {
+        versionDetector.setFilterPacketIns(enabled);
+        LOG.debug("PacketIn filtering {}abled", enabled ? "en" : "dis");
+    }
 }