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.Promise;
  19 import java.net.InetSocketAddress;
  20 import org.slf4j.Logger;
  21 import org.slf4j.LoggerFactory;
  22
  23 final class ReconnectPromise<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> extends DefaultPromise<Void> {
  24     private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
  25
  26     private final AbstractDispatcher<S, L> dispatcher;
  27     private final InetSocketAddress address;
  28     private final ReconnectStrategyFactory strategyFactory;
  29     private final Bootstrap b;
  30     private final AbstractDispatcher.PipelineInitializer<S> initializer;
  31     private Future<?> pending;
  32
  33     public ReconnectPromise(final EventExecutor executor, final AbstractDispatcher<S, L> dispatcher, final InetSocketAddress address,
  34                             final ReconnectStrategyFactory connectStrategyFactory, final Bootstrap b, final AbstractDispatcher.PipelineInitializer<S> initializer) {
  35         super(executor);
  36         this.b = b;
  37         this.initializer = Preconditions.checkNotNull(initializer);
  38         this.dispatcher = Preconditions.checkNotNull(dispatcher);
  39         this.address = Preconditions.checkNotNull(address);
  40         this.strategyFactory = Preconditions.checkNotNull(connectStrategyFactory);
  41     }
  42
  43     synchronized void connect() {
  44         final ReconnectStrategy cs = this.strategyFactory.createReconnectStrategy();
  45
  46         // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
  47         pending = this.dispatcher.createClient(this.address, cs, b, new AbstractDispatcher.PipelineInitializer<S>() {
  48             @Override
  49             public void initializeChannel(final SocketChannel channel, final Promise<S> promise) {
  50                 initializer.initializeChannel(channel, promise);
  51
  52                 // add closed channel handler
  53                 channel.pipeline().addFirst(new ClosedChannelHandler(ReconnectPromise.this));
  54             }
  55         });
  56     }
  57
  58     /**
  59      *
  60      * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
  61      */
  62     private boolean isInitialConnectFinished() {
  63         Preconditions.checkNotNull(pending);
  64         return pending.isDone() && pending.isSuccess();
  65     }
  66
  67     @Override
  68     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
  69         if (super.cancel(mayInterruptIfRunning)) {
  70             Preconditions.checkNotNull(pending);
  71             this.pending.cancel(mayInterruptIfRunning);
  72             return true;
  73         }
  74
  75         return false;
  76     }
  77
  78     /**
  79      * Channel handler that responds to channelInactive event and reconnects the session.
  80      * Only if the initial connection was successfully established and promise was not canceled.
  81      */
  82     private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
  83         private final ReconnectPromise<?, ?> promise;
  84
  85         public ClosedChannelHandler(final ReconnectPromise<?, ?> promise) {
  86             this.promise = promise;
  87         }
  88
  89         @Override
  90         public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
  91             if (promise.isCancelled()) {
  92                 return;
  93             }
  94
  95             // Check if initial connection was fully finished. If the session was dropped during negotiation, reconnect will not happen.
  96             // Session can be dropped during negotiation on purpose by the client side and would make no sense to initiate reconnect
  97             if (promise.isInitialConnectFinished() == false) {
  98                 return;
  99             }
 100
 101             LOG.debug("Reconnecting after connection to {} was dropped", promise.address);
 102             promise.connect();
 103         }
 104     }
 105
 106 }