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