opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/ReconnectPromise.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 com.google.common.base.Preconditions;
  11 import io.netty.bootstrap.Bootstrap;
  12 import io.netty.channel.ChannelHandlerContext;
  13 import io.netty.channel.ChannelInboundHandlerAdapter;
  14 import io.netty.channel.socket.SocketChannel;
  15 import io.netty.util.concurrent.DefaultPromise;
  16 import io.netty.util.concurrent.EventExecutor;
  17 import io.netty.util.concurrent.Future;
  18 import io.netty.util.concurrent.GenericFutureListener;
  19 import io.netty.util.concurrent.Promise;
  20 import java.net.InetSocketAddress;
  21 import org.slf4j.Logger;
  22 import org.slf4j.LoggerFactory;
  23
  24 @Deprecated
  25 final class ReconnectPromise<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> extends DefaultPromise<Void> {
  26     private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
  27
  28     private final AbstractDispatcher<S, L> dispatcher;
  29     private final InetSocketAddress address;
  30     private final ReconnectStrategyFactory strategyFactory;
  31     private final Bootstrap b;
  32     private final AbstractDispatcher.PipelineInitializer<S> initializer;
  33     private Future<?> pending;
  34
  35     public ReconnectPromise(final EventExecutor executor, final AbstractDispatcher<S, L> dispatcher, final InetSocketAddress address,
  36                             final ReconnectStrategyFactory connectStrategyFactory, final Bootstrap b, final AbstractDispatcher.PipelineInitializer<S> initializer) {
  37         super(executor);
  38         this.b = b;
  39         this.initializer = Preconditions.checkNotNull(initializer);
  40         this.dispatcher = Preconditions.checkNotNull(dispatcher);
  41         this.address = Preconditions.checkNotNull(address);
  42         this.strategyFactory = Preconditions.checkNotNull(connectStrategyFactory);
  43     }
  44
  45     synchronized void connect() {
  46         final ReconnectStrategy cs = this.strategyFactory.createReconnectStrategy();
  47
  48         // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
  49         pending = this.dispatcher.createClient(this.address, cs, b, new AbstractDispatcher.PipelineInitializer<S>() {
  50             @Override
  51             public void initializeChannel(final SocketChannel channel, final Promise<S> promise) {
  52                 initializer.initializeChannel(channel, promise);
  53                 // add closed channel handler
  54                 // This handler has to be added as last channel handler and the channel inactive event has to be caught by it
  55                 // Handlers in front of it can react to channelInactive event, but have to forward the event or the reconnect will not work
  56                 // This handler is last so all handlers in front of it can handle channel inactive (to e.g. resource cleanup) before a new connection is started
  57                 channel.pipeline().addLast(new ClosedChannelHandler(ReconnectPromise.this));
  58             }
  59         });
  60
  61         pending.addListener(new GenericFutureListener<Future<Object>>() {
  62             @Override
  63             public void operationComplete(Future<Object> future) throws Exception {
  64                 if (!future.isSuccess()) {
  65                     ReconnectPromise.this.setFailure(future.cause());
  66                 }
  67             }
  68         });
  69     }
  70
  71     /**
  72      *
  73      * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
  74      */
  75     private boolean isInitialConnectFinished() {
  76         Preconditions.checkNotNull(pending);
  77         return pending.isDone() && pending.isSuccess();
  78     }
  79
  80     @Override
  81     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
  82         if (super.cancel(mayInterruptIfRunning)) {
  83             Preconditions.checkNotNull(pending);
  84             this.pending.cancel(mayInterruptIfRunning);
  85             return true;
  86         }
  87
  88         return false;
  89     }
  90
  91     /**
  92      * Channel handler that responds to channelInactive event and reconnects the session.
  93      * Only if the promise was not canceled.
  94      */
  95     private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
  96         private final ReconnectPromise<?, ?> promise;
  97
  98         public ClosedChannelHandler(final ReconnectPromise<?, ?> promise) {
  99             this.promise = promise;
 100         }
 101
 102         @Override
 103         public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
 104             // This is the ultimate channel inactive handler, not forwarding
 105             if (promise.isCancelled()) {
 106                 return;
 107             }
 108
 109             if (promise.isInitialConnectFinished() == false) {
 110                 LOG.debug("Connection to {} was dropped during negotiation, reattempting", promise.address);
 111             }
 112
 113             LOG.debug("Reconnecting after connection to {} was dropped", promise.address);
 114             promise.connect();
 115         }
 116     }
 117
 118 }