bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/protocol/BGPReconnectPromise.java

   1 /*
   2  * Copyright (c) 2015 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.bgp.rib.impl.protocol;
   9
  10 import com.google.common.base.Preconditions;
  11 import io.netty.bootstrap.Bootstrap;
  12 import io.netty.channel.ChannelHandler;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelInboundHandlerAdapter;
  15 import io.netty.channel.ChannelInitializer;
  16 import io.netty.channel.socket.SocketChannel;
  17 import io.netty.util.concurrent.DefaultPromise;
  18 import io.netty.util.concurrent.EventExecutor;
  19 import io.netty.util.concurrent.Future;
  20 import io.netty.util.concurrent.GenericFutureListener;
  21 import io.netty.util.concurrent.Promise;
  22 import java.net.InetSocketAddress;
  23 import org.opendaylight.protocol.bgp.rib.impl.spi.ChannelPipelineInitializer;
  24 import org.opendaylight.protocol.bgp.rib.spi.BGPSession;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28 public class BGPReconnectPromise<S extends BGPSession> extends DefaultPromise<Void> {
  29     private static final Logger LOG = LoggerFactory.getLogger(BGPReconnectPromise.class);
  30
  31     private final InetSocketAddress address;
  32     private final int retryTimer;
  33     private final Bootstrap bootstrap;
  34     private final ChannelPipelineInitializer initializer;
  35     private BGPProtocolSessionPromise<S> pending;
  36
  37     public BGPReconnectPromise(final EventExecutor executor, final InetSocketAddress address,
  38                                final int retryTimer, final Bootstrap bootstrap,
  39                                final ChannelPipelineInitializer initializer) {
  40         super(executor);
  41         this.bootstrap = bootstrap;
  42         this.initializer = Preconditions.checkNotNull(initializer);
  43         this.address = Preconditions.checkNotNull(address);
  44         this.retryTimer = retryTimer;
  45     }
  46
  47     public synchronized void connect() {
  48         // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
  49         this.pending = connectSessionPromise(this.address, this.retryTimer, this.bootstrap, new ChannelPipelineInitializer<S>() {
  50             @Override
  51             public void initializeChannel(final SocketChannel channel, final Promise<S> promise) {
  52                 BGPReconnectPromise.this.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(BGPReconnectPromise.this));
  58             }
  59         });
  60
  61         this.pending.addListener(new GenericFutureListener<Future<Object>>() {
  62             @Override
  63             public void operationComplete(final Future<Object> future) throws Exception {
  64                 if (!future.isSuccess()) {
  65                     BGPReconnectPromise.this.setFailure(future.cause());
  66                 }
  67             }
  68         });
  69     }
  70
  71     public BGPProtocolSessionPromise<S> connectSessionPromise(final InetSocketAddress address, final int retryTimer, final Bootstrap bootstrap,
  72                                   final ChannelPipelineInitializer initializer) {
  73         final BGPProtocolSessionPromise sessionPromise = new BGPProtocolSessionPromise(address, retryTimer, bootstrap);
  74         final ChannelHandler chInit = new ChannelInitializer<SocketChannel>() {
  75             @Override
  76             protected void initChannel(final SocketChannel channel) {
  77                 initializer.initializeChannel(channel, sessionPromise);
  78             }
  79         };
  80
  81         bootstrap.handler(chInit);
  82         sessionPromise.connect();
  83         LOG.debug("Client created.");
  84         return sessionPromise;
  85     }
  86
  87     /**
  88      * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
  89      */
  90     private boolean isInitialConnectFinished() {
  91         Preconditions.checkNotNull(this.pending);
  92         return this.pending.isDone() && this.pending.isSuccess();
  93     }
  94
  95     @Override
  96     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
  97         if (super.cancel(mayInterruptIfRunning)) {
  98             Preconditions.checkNotNull(this.pending);
  99             this.pending.cancel(mayInterruptIfRunning);
 100             return true;
 101         }
 102         return false;
 103     }
 104
 105     /**
 106      * Channel handler that responds to channelInactive event and reconnects the session.
 107      * Only if the promise was not canceled.
 108      */
 109     private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
 110         private final BGPReconnectPromise promise;
 111
 112         public ClosedChannelHandler(final BGPReconnectPromise promise) {
 113             this.promise = promise;
 114         }
 115
 116         @Override
 117         public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
 118             // This is the ultimate channel inactive handler, not forwarding
 119             if (this.promise.isCancelled()) {
 120                 return;
 121             }
 122
 123             if (!this.promise.isInitialConnectFinished()) {
 124                 LOG.debug("Connection to {} was dropped during negotiation, reattempting", this.promise.address);
 125                 this.promise.pending.reconnect();
 126                 return;
 127             }
 128
 129             LOG.debug("Reconnecting after connection to {} was dropped", this.promise.address);
 130             this.promise.connect();
 131         }
 132     }
 133 }