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.opendaylight.protocol.framework.ReconnectStrategy;
  26 import org.opendaylight.protocol.framework.ReconnectStrategyFactory;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29
  30 public class BGPReconnectPromise<S extends BGPSession> extends DefaultPromise<Void> {
  31     private static final Logger LOG = LoggerFactory.getLogger(BGPReconnectPromise.class);
  32
  33     private final InetSocketAddress address;
  34     private final ReconnectStrategyFactory strategyFactory;
  35     private final Bootstrap b;
  36     private final ChannelPipelineInitializer initializer;
  37     private final EventExecutor executor;
  38     private Future<S> pending;
  39
  40     public BGPReconnectPromise(final EventExecutor executor, final InetSocketAddress address,
  41                                final ReconnectStrategyFactory connectStrategyFactory, final Bootstrap b,
  42                                final ChannelPipelineInitializer initializer) {
  43         super(executor);
  44         this.executor = executor;
  45         this.b = b;
  46         this.initializer = Preconditions.checkNotNull(initializer);
  47         this.address = Preconditions.checkNotNull(address);
  48         this.strategyFactory = Preconditions.checkNotNull(connectStrategyFactory);
  49     }
  50
  51     public synchronized void connect() {
  52         final ReconnectStrategy cs = this.strategyFactory.createReconnectStrategy();
  53
  54         // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
  55         pending = createClient(this.address, cs, b, new ChannelPipelineInitializer<S>() {
  56             @Override
  57             public void initializeChannel(final SocketChannel channel, final Promise<S> promise) {
  58                 initializer.initializeChannel(channel, promise);
  59                 // add closed channel handler
  60                 // This handler has to be added as last channel handler and the channel inactive event has to be caught by it
  61                 // Handlers in front of it can react to channelInactive event, but have to forward the event or the reconnect will not work
  62                 // 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
  63                 channel.pipeline().addLast(new ClosedChannelHandler(BGPReconnectPromise.this));
  64             }
  65         });
  66
  67         pending.addListener(new GenericFutureListener<Future<Object>>() {
  68             @Override
  69             public void operationComplete(Future<Object> future) throws Exception {
  70                 if (!future.isSuccess()) {
  71                     BGPReconnectPromise.this.setFailure(future.cause());
  72                 }
  73             }
  74         });
  75     }
  76
  77     public Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final Bootstrap bootstrap,
  78                                   final ChannelPipelineInitializer initializer) {
  79         final BGPProtocolSessionPromise p = new BGPProtocolSessionPromise(this.executor, address, strategy, bootstrap);
  80         final ChannelHandler chInit = new ChannelInitializer<SocketChannel>() {
  81             @Override
  82             protected void initChannel(SocketChannel ch) {
  83                 initializer.initializeChannel(ch, p);
  84             }
  85         };
  86
  87         bootstrap.handler(chInit);
  88         p.connect();
  89         LOG.debug("Client created.");
  90         return p;
  91     }
  92
  93     /**
  94      * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
  95      */
  96     private boolean isInitialConnectFinished() {
  97         Preconditions.checkNotNull(pending);
  98         return pending.isDone() && pending.isSuccess();
  99     }
 100
 101     @Override
 102     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
 103         if (super.cancel(mayInterruptIfRunning)) {
 104             Preconditions.checkNotNull(pending);
 105             this.pending.cancel(mayInterruptIfRunning);
 106             return true;
 107         }
 108
 109         return false;
 110     }
 111
 112     /**
 113      * Channel handler that responds to channelInactive event and reconnects the session.
 114      * Only if the promise was not canceled.
 115      */
 116     private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
 117         private final BGPReconnectPromise promise;
 118
 119         public ClosedChannelHandler(final BGPReconnectPromise promise) {
 120             this.promise = promise;
 121         }
 122
 123         @Override
 124         public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
 125             // This is the ultimate channel inactive handler, not forwarding
 126             if (promise.isCancelled()) {
 127                 return;
 128             }
 129
 130             if (!promise.isInitialConnectFinished()) {
 131                 LOG.debug("Connection to {} was dropped during negotiation, reattempting", promise.address);
 132             }
 133
 134             LOG.debug("Reconnecting after connection to {} was dropped", promise.address);
 135             promise.connect();
 136         }
 137     }
 138 }