openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/connection/listener/SystemNotificationsListenerImpl.java

   1 /**
   2  * Copyright (c) 2015 Cisco Systems, Inc. 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 package org.opendaylight.openflowplugin.impl.connection.listener;
   9
  10 import org.opendaylight.openflowplugin.api.openflow.connection.HandshakeContext;
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.Futures;
  14 import com.google.common.util.concurrent.JdkFutureAdapters;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import java.net.InetSocketAddress;
  17 import java.util.concurrent.Future;
  18 import java.util.concurrent.TimeUnit;
  19 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  20 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  21 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoInputBuilder;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoOutput;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FeaturesReply;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.DisconnectEvent;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SwitchIdleEvent;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SystemNotificationsListener;
  28 import org.opendaylight.yangtools.yang.common.RpcError;
  29 import org.opendaylight.yangtools.yang.common.RpcResult;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 /**
  34  *
  35  */
  36 public class SystemNotificationsListenerImpl implements SystemNotificationsListener {
  37
  38     private ConnectionContext connectionContext;
  39     HandshakeContext handshakeContext;
  40     private static final Logger LOG = LoggerFactory.getLogger(SystemNotificationsListenerImpl.class);
  41     @VisibleForTesting
  42     static final long MAX_ECHO_REPLY_TIMEOUT = 2000;
  43
  44     public SystemNotificationsListenerImpl(final ConnectionContext connectionContext,
  45             final HandshakeContext handshakeContext) {
  46         this.connectionContext = connectionContext;
  47         this.handshakeContext = handshakeContext;
  48     }
  49
  50     @Override
  51     public void onDisconnectEvent(DisconnectEvent notification) {
  52         disconnect();
  53     }
  54
  55     @Override
  56     public void onSwitchIdleEvent(SwitchIdleEvent notification) {
  57         new Thread(new Runnable() {
  58             @Override
  59             public void run() {
  60                 boolean shouldBeDisconnected = true;
  61
  62                 final InetSocketAddress remoteAddress = connectionContext.getConnectionAdapter().getRemoteAddress();
  63
  64                 if (ConnectionContext.CONNECTION_STATE.WORKING.equals(connectionContext.getConnectionState())) {
  65                     FeaturesReply features = connectionContext.getFeatures();
  66                     LOG.debug(
  67                             "first idle state occured, node={}|auxId={}",
  68                             remoteAddress, features.getAuxiliaryId());
  69                     connectionContext.setConnectionState(ConnectionContext.CONNECTION_STATE.TIMEOUTING);
  70                     EchoInputBuilder builder = new EchoInputBuilder();
  71                     builder.setVersion(features.getVersion());
  72                     Xid xid = new Xid(0);
  73                     builder.setXid(xid.getValue());
  74
  75                     Future<RpcResult<EchoOutput>> echoReplyFuture = connectionContext.getConnectionAdapter()
  76                             .echo(builder.build());
  77
  78                     try {
  79                         RpcResult<EchoOutput> echoReplyValue = echoReplyFuture.get(MAX_ECHO_REPLY_TIMEOUT, TimeUnit.MILLISECONDS);
  80                         if (echoReplyValue.isSuccessful()) {
  81                             connectionContext.setConnectionState(ConnectionContext.CONNECTION_STATE.WORKING);
  82                             shouldBeDisconnected = false;
  83                         } else {
  84                             for (RpcError replyError : echoReplyValue
  85                                     .getErrors()) {
  86                                 Throwable cause = replyError.getCause();
  87                                 LOG.warn("while receiving echoReply [{}] in TIMEOUTING state {} ",
  88                                         remoteAddress,
  89                                         cause.getMessage());
  90                                 LOG.trace("while receiving echoReply [{}] in TIMEOUTING state ..", remoteAddress, cause);
  91                             }
  92                         }
  93                     } catch (Exception e) {
  94                         LOG.warn("while waiting for echoReply in TIMEOUTING state: {}", e.getMessage());
  95                         LOG.trace("while waiting for echoReply in TIMEOUTING state ..", remoteAddress, e);
  96                     }
  97                 }
  98                 if (shouldBeDisconnected) {
  99                     disconnect();
 100                 }
 101             }
 102         }).start();
 103     }
 104
 105     private void disconnect() {
 106         final ConnectionAdapter connectionAdapter = connectionContext.getConnectionAdapter();
 107         short auxId = -1;
 108         if (null != connectionContext.getFeatures() && null != connectionContext.getFeatures().getAuxiliaryId()) {
 109             auxId = connectionContext.getFeatures().getAuxiliaryId();
 110         }
 111         final Short auxiliaryId = auxId;
 112         final InetSocketAddress remoteAddress = connectionAdapter.getRemoteAddress();
 113
 114         LOG.trace("disconnecting: node={}|auxId={}|connection state = {}",
 115                 remoteAddress,
 116                 auxiliaryId,
 117                 connectionContext.getConnectionState());
 118
 119         ListenableFuture<Boolean> result = null;
 120         if (connectionAdapter.isAlive()) {
 121             result = JdkFutureAdapters.listenInPoolThread(connectionAdapter.disconnect());
 122         } else {
 123             LOG.debug("connection already disconnected");
 124             result = Futures.immediateFuture(true);
 125         }
 126         connectionContext.setConnectionState(ConnectionContext.CONNECTION_STATE.RIP);
 127         Futures.addCallback(result, new FutureCallback<Boolean>() {
 128             @Override
 129             public void onSuccess(final Boolean aBoolean) {
 130                 LOG.debug("Connection node={}|auxId={}|connection state = {}, closed successfully:{}.",
 131                         remoteAddress,
 132                         auxiliaryId,
 133                         connectionContext.getConnectionState(),
 134                         aBoolean);
 135             }
 136
 137             @Override
 138             public void onFailure(final Throwable throwable) {
 139                 LOG.debug("Connection node={}|auxId={}|connection state = {} close failed.",
 140                         remoteAddress,
 141                         auxiliaryId,
 142                         connectionContext.getConnectionState());
 143             }
 144         });
 145
 146         connectionContext.propagateClosingConnection();
 147         try {
 148             handshakeContext.close();
 149         } catch (Exception e) {
 150             LOG.debug("Closing of handshake context wasn't successfull. {}", e);
 151         }
 152     }
 153 }