opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/AbstractSessionNegotiator.java

   1 /*
   2  * Copyright (c) 2013 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.protocol.framework;
   9
  10 import io.netty.channel.Channel;
  11 import io.netty.channel.ChannelFuture;
  12 import io.netty.channel.ChannelFutureListener;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelInboundHandlerAdapter;
  15 import io.netty.util.concurrent.Promise;
  16
  17 import org.slf4j.Logger;
  18 import org.slf4j.LoggerFactory;
  19
  20 import com.google.common.base.Preconditions;
  21
  22 /**
  23  * Abstract base class for session negotiators. It implements the basic
  24  * substrate to implement SessionNegotiator API specification, with subclasses
  25  * needing to provide only
  26  *
  27  * @param <M> Protocol message type
  28  * @param <S> Protocol session type, has to extend ProtocolSession<M>
  29  */
  30 @Deprecated
  31 public abstract class AbstractSessionNegotiator<M, S extends AbstractProtocolSession<?>> extends ChannelInboundHandlerAdapter implements SessionNegotiator<S> {
  32     private final Logger LOG = LoggerFactory.getLogger(AbstractSessionNegotiator.class);
  33     private final Promise<S> promise;
  34     protected final Channel channel;
  35
  36     public AbstractSessionNegotiator(final Promise<S> promise, final Channel channel) {
  37         this.promise = Preconditions.checkNotNull(promise);
  38         this.channel = Preconditions.checkNotNull(channel);
  39     }
  40
  41     protected abstract void startNegotiation() throws Exception;
  42     protected abstract void handleMessage(M msg) throws Exception;
  43
  44     protected final void negotiationSuccessful(final S session) {
  45         LOG.debug("Negotiation on channel {} successful with session {}", channel, session);
  46         channel.pipeline().replace(this, "session", session);
  47         promise.setSuccess(session);
  48     }
  49
  50     protected void negotiationFailed(final Throwable cause) {
  51         LOG.debug("Negotiation on channel {} failed", channel, cause);
  52         channel.close();
  53         promise.setFailure(cause);
  54     }
  55
  56     /**
  57      * Send a message to peer and fail negotiation if it does not reach
  58      * the peer.
  59      *
  60      * @param msg Message which should be sent.
  61      */
  62     protected final void sendMessage(final M msg) {
  63         this.channel.writeAndFlush(msg).addListener(
  64                 new ChannelFutureListener() {
  65                     @Override
  66                     public void operationComplete(final ChannelFuture f) {
  67                         if (!f.isSuccess()) {
  68                             LOG.info("Failed to send message {}", msg, f.cause());
  69                             negotiationFailed(f.cause());
  70                         } else {
  71                             LOG.trace("Message {} sent to socket", msg);
  72                         }
  73                     }
  74                 });
  75     }
  76
  77     @Override
  78     public final void channelActive(final ChannelHandlerContext ctx) {
  79         LOG.debug("Starting session negotiation on channel {}", channel);
  80         try {
  81             startNegotiation();
  82         } catch (Exception e) {
  83             LOG.warn("Unexpected negotiation failure", e);
  84             negotiationFailed(e);
  85         }
  86     }
  87
  88     @Override
  89     @SuppressWarnings("unchecked")
  90     public final void channelRead(final ChannelHandlerContext ctx, final Object msg) {
  91         LOG.debug("Negotiation read invoked on channel {}", channel);
  92         try {
  93             handleMessage((M)msg);
  94         } catch (Exception e) {
  95             LOG.debug("Unexpected error while handling negotiation message {}", msg, e);
  96             negotiationFailed(e);
  97         }
  98     }
  99
 100     @Override
 101     public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
 102         LOG.info("Unexpected error during negotiation", cause);
 103         negotiationFailed(cause);
 104     }
 105 }