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

   1 /*
   2  * Copyright (c) 2013 Pantheon Technologies s.r.o. and others. All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9
  10 package org.opendaylight.openflowjava.protocol.impl.core.connection;
  11
  12 import com.google.common.base.Preconditions;
  13 import io.netty.channel.Channel;
  14 import java.net.InetSocketAddress;
  15 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
  16 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandler;
  17 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
  18 import org.opendaylight.openflowjava.protocol.api.extensibility.AlienMessageListener;
  19 import org.opendaylight.openflowjava.protocol.impl.core.OFVersionDetector;
  20 import org.opendaylight.openflowjava.protocol.impl.core.PipelineHandlers;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoOutput;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoRequestMessage;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ErrorMessage;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemovedMessage;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.HelloMessage;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReplyMessage;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OpenflowProtocolListener;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.DisconnectEvent;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SwitchIdleEvent;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SystemNotificationsListener;
  35 import org.opendaylight.yangtools.yang.binding.DataObject;
  36 import org.opendaylight.yangtools.yang.binding.Notification;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39
  40 /**
  41  * Handles messages (notifications + rpcs) and connections.
  42  * @author mirehak
  43  * @author michal.polkorab
  44  */
  45 public class ConnectionAdapterImpl extends AbstractConnectionAdapterStatistics implements ConnectionFacade {
  46
  47     private static final Logger LOG = LoggerFactory.getLogger(ConnectionAdapterImpl.class);
  48
  49     private ConnectionReadyListener connectionReadyListener;
  50     private OpenflowProtocolListener messageListener;
  51     private SystemNotificationsListener systemListener;
  52     private AlienMessageListener alienMessageListener;
  53     private AbstractOutboundQueueManager<?, ?> outputManager;
  54     private OFVersionDetector versionDetector;
  55
  56     private final boolean useBarrier;
  57
  58     /**
  59      * Default constructor.
  60      * @param channel the channel to be set - used for communication
  61      * @param address client address (used only in case of UDP communication,
  62      *                as there is no need to store address over tcp (stable channel))
  63      * @param useBarrier value is configurable by configSubsytem
  64      */
  65     public ConnectionAdapterImpl(final Channel channel, final InetSocketAddress address, final boolean useBarrier) {
  66         super(channel, address);
  67         this.useBarrier = useBarrier;
  68         LOG.debug("ConnectionAdapter created");
  69     }
  70
  71     @Override
  72     public void setMessageListener(final OpenflowProtocolListener messageListener) {
  73         this.messageListener = messageListener;
  74     }
  75
  76     @Override
  77     public void setConnectionReadyListener(final ConnectionReadyListener connectionReadyListener) {
  78         this.connectionReadyListener = connectionReadyListener;
  79     }
  80
  81     @Override
  82     public void setSystemListener(final SystemNotificationsListener systemListener) {
  83         this.systemListener = systemListener;
  84     }
  85
  86     @Override
  87     public void setAlienMessageListener(final AlienMessageListener alienMessageListener) {
  88         this.alienMessageListener = alienMessageListener;
  89     }
  90
  91     @Override
  92     public void consumeDeviceMessage(final DataObject message) {
  93         LOG.debug("ConsumeIntern msg on {}", channel);
  94         if (disconnectOccured) {
  95             return;
  96         }
  97         if (message instanceof Notification) {
  98
  99             // System events
 100             if (message instanceof DisconnectEvent) {
 101                 systemListener.onDisconnectEvent((DisconnectEvent) message);
 102                 responseCache.invalidateAll();
 103                 disconnectOccured = true;
 104             } else if (message instanceof SwitchIdleEvent) {
 105                 systemListener.onSwitchIdleEvent((SwitchIdleEvent) message);
 106             // OpenFlow messages
 107             } else if (message instanceof EchoRequestMessage) {
 108                 if (outputManager != null) {
 109                     outputManager.onEchoRequest((EchoRequestMessage) message);
 110                 } else {
 111                     messageListener.onEchoRequestMessage((EchoRequestMessage) message);
 112                 }
 113             } else if (message instanceof ErrorMessage) {
 114                 // Send only unmatched errors
 115                 if (outputManager == null || !outputManager.onMessage((OfHeader) message)) {
 116                     messageListener.onErrorMessage((ErrorMessage) message);
 117                 }
 118             } else if (message instanceof ExperimenterMessage) {
 119                 if (outputManager != null) {
 120                     outputManager.onMessage((OfHeader) message);
 121                 }
 122                 messageListener.onExperimenterMessage((ExperimenterMessage) message);
 123             } else if (message instanceof FlowRemovedMessage) {
 124                 messageListener.onFlowRemovedMessage((FlowRemovedMessage) message);
 125             } else if (message instanceof HelloMessage) {
 126                 LOG.info("Hello received");
 127                 messageListener.onHelloMessage((HelloMessage) message);
 128             } else if (message instanceof MultipartReplyMessage) {
 129                 if (outputManager != null) {
 130                     outputManager.onMessage((OfHeader) message);
 131                 }
 132                 messageListener.onMultipartReplyMessage((MultipartReplyMessage) message);
 133             } else if (message instanceof PacketInMessage) {
 134                 messageListener.onPacketInMessage((PacketInMessage) message);
 135             } else if (message instanceof PortStatusMessage) {
 136                 messageListener.onPortStatusMessage((PortStatusMessage) message);
 137             } else {
 138                 LOG.warn("message listening not supported for type: {}", message.getClass());
 139             }
 140         } else if (message instanceof OfHeader) {
 141             LOG.debug("OF header msg received");
 142
 143             if (alienMessageListener != null && alienMessageListener.onAlienMessage((OfHeader) message)) {
 144                 LOG.debug("Alien message {} received", message.getImplementedInterface());
 145             } else if (outputManager == null || !outputManager.onMessage((OfHeader) message) || message instanceof
 146                     EchoOutput) {
 147                 final RpcResponseKey key = createRpcResponseKey((OfHeader) message);
 148                 final ResponseExpectedRpcListener<?> listener = findRpcResponse(key);
 149                 if (listener != null) {
 150                     LOG.debug("Corresponding rpcFuture found");
 151                     listener.completed((OfHeader) message);
 152                     LOG.debug("After setting rpcFuture");
 153                     responseCache.invalidate(key);
 154                 }
 155             }
 156         } else {
 157             LOG.warn("message listening not supported for type: {}", message.getClass());
 158         }
 159     }
 160
 161     private static RpcResponseKey createRpcResponseKey(final OfHeader message) {
 162         return new RpcResponseKey(message.getXid(), message.getImplementedInterface().getName());
 163     }
 164
 165     @Override
 166     public void checkListeners() {
 167         final StringBuilder buffer =  new StringBuilder();
 168         if (systemListener == null) {
 169             buffer.append("SystemListener ");
 170         }
 171         if (messageListener == null) {
 172             buffer.append("MessageListener ");
 173         }
 174         if (connectionReadyListener == null) {
 175             buffer.append("ConnectionReadyListener ");
 176         }
 177
 178         Preconditions.checkState(buffer.length() == 0, "Missing listeners: %s", buffer.toString());
 179     }
 180
 181     @Override
 182     public void fireConnectionReadyNotification() {
 183         versionDetector = (OFVersionDetector) channel.pipeline().get(PipelineHandlers.OF_VERSION_DETECTOR.name());
 184         Preconditions.checkState(versionDetector != null);
 185
 186         new Thread(new Runnable() {
 187             @Override
 188             public void run() {
 189                 connectionReadyListener.onConnectionReady();
 190             }
 191         }).start();
 192     }
 193
 194     @Override
 195     public <T extends OutboundQueueHandler> OutboundQueueHandlerRegistration<T> registerOutboundQueueHandler(
 196             final T handler, final int maxQueueDepth, final long maxBarrierNanos) {
 197         Preconditions.checkState(outputManager == null, "Manager %s already registered", outputManager);
 198
 199         final AbstractOutboundQueueManager<T, ?> ret;
 200         if (useBarrier) {
 201             ret = new OutboundQueueManager<>(this, address, handler, maxQueueDepth, maxBarrierNanos);
 202         } else {
 203             LOG.warn("OutboundQueueManager without barrier is started.");
 204             ret = new OutboundQueueManagerNoBarrier<>(this, address, handler);
 205         }
 206
 207         outputManager = ret;
 208         /* we don't need it anymore */
 209         channel.pipeline().remove(output);
 210         // OutboundQueueManager is put before DelegatingInboundHandler because otherwise channelInactive event would
 211         // be first processed in OutboundQueueManager and then in ConnectionAdapter (and Openflowplugin). This might
 212         // cause problems because we are shutting down the queue before Openflowplugin knows about it.
 213         channel.pipeline().addBefore(PipelineHandlers.DELEGATING_INBOUND_HANDLER.name(),
 214                 PipelineHandlers.CHANNEL_OUTBOUND_QUEUE_MANAGER.name(), outputManager);
 215
 216         return new OutboundQueueHandlerRegistrationImpl<T>(handler) {
 217             @Override
 218             protected void removeRegistration() {
 219                 outputManager.close();
 220                 channel.pipeline().remove(outputManager);
 221                 outputManager = null;
 222             }
 223         };
 224     }
 225
 226     Channel getChannel() {
 227         return channel;
 228     }
 229
 230     @Override
 231     public void setPacketInFiltering(final boolean enabled) {
 232         versionDetector.setFilterPacketIns(enabled);
 233         LOG.debug("PacketIn filtering {}abled", enabled ? "en" : "dis");
 234     }
 235 }