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 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 final 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                 new ChannelFutureListener() {
  64                     @Override
  65                     public void operationComplete(final ChannelFuture f) {
  66                         if (!f.isSuccess()) {
  67                             LOG.info("Failed to send message {}", msg, f.cause());
  68                             negotiationFailed(f.cause());
  69                         } else {
  70                             LOG.trace("Message {} sent to socket", msg);
  71                         }
  72                     }
  73                 });
  74     }
  75
  76     @Override
  77     public final void channelActive(final ChannelHandlerContext ctx) {
  78         LOG.debug("Starting session negotiation on channel {}", channel);
  79         try {
  80             startNegotiation();
  81         } catch (Exception e) {
  82             LOG.warn("Unexpected negotiation failure", e);
  83             negotiationFailed(e);
  84         }
  85     }
  86
  87     @Override
  88     public final void channelRead(final ChannelHandlerContext ctx, final Object msg) {
  89         LOG.debug("Negotiation read invoked on channel {}", channel);
  90         try {
  91             handleMessage((M)msg);
  92         } catch (Exception e) {
  93             LOG.debug("Unexpected error while handling negotiation message {}", msg, e);
  94             negotiationFailed(e);
  95         }
  96     }
  97
  98     @Override
  99     public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
 100         LOG.info("Unexpected error during negotiation", cause);
 101         negotiationFailed(cause);
 102     }
 103 }