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 java.net.InetSocketAddress;
  20 import javax.annotation.Nonnull;
  21 import org.opendaylight.protocol.bgp.rib.impl.spi.BGPPeerRegistry;
  22 import org.opendaylight.protocol.bgp.rib.impl.spi.ChannelPipelineInitializer;
  23 import org.opendaylight.protocol.bgp.rib.spi.BGPSession;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 public class BGPReconnectPromise<S extends BGPSession> extends DefaultPromise<Void> {
  28     private static final Logger LOG = LoggerFactory.getLogger(BGPReconnectPromise.class);
  29
  30     private final InetSocketAddress address;
  31     private final int retryTimer;
  32     private final Bootstrap bootstrap;
  33     private final BGPPeerRegistry peerRegistry;
  34     private final ChannelPipelineInitializer<S> initializer;
  35     private BGPProtocolSessionPromise<S> pending;
  36
  37     public BGPReconnectPromise(@Nonnull final EventExecutor executor, @Nonnull final InetSocketAddress address,
  38         final int retryTimer, @Nonnull final Bootstrap bootstrap, @Nonnull final BGPPeerRegistry peerRegistry,
  39         @Nonnull final ChannelPipelineInitializer<S> 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         this.peerRegistry = Preconditions.checkNotNull(peerRegistry);
  46     }
  47
  48     public synchronized void connect() {
  49         if (this.pending != null) {
  50             this.pending.cancel(true);
  51         }
  52
  53         // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
  54         this.pending = connectSessionPromise(this.address, this.retryTimer, this.bootstrap, this.peerRegistry, (channel, promise) -> {
  55             this.initializer.initializeChannel(channel, promise);
  56             // add closed channel handler
  57             // This handler has to be added as last channel handler and the channel inactive event has to be caught by it
  58             // Handlers in front of it can react to channelInactive event, but have to forward the event or the reconnect will not work
  59             // 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
  60             channel.pipeline().addLast(new ClosedChannelHandler(this));
  61         });
  62
  63         this.pending.addListener(future -> {
  64             if (!future.isSuccess() && !this.isDone()) {
  65                 this.setFailure(future.cause());
  66             }
  67         });
  68     }
  69
  70     private BGPProtocolSessionPromise<S> connectSessionPromise(final InetSocketAddress address, final int retryTimer,
  71             final Bootstrap bootstrap, final BGPPeerRegistry peerRegistry,
  72             final ChannelPipelineInitializer<S> initializer) {
  73         final BGPProtocolSessionPromise<S> sessionPromise = new BGPProtocolSessionPromise<>(address, retryTimer,
  74                 bootstrap, peerRegistry);
  75         final ChannelHandler chInit = new ChannelInitializer<SocketChannel>() {
  76             @Override
  77             protected void initChannel(final SocketChannel channel) {
  78                 initializer.initializeChannel(channel, sessionPromise);
  79             }
  80         };
  81
  82         bootstrap.handler(chInit);
  83         sessionPromise.connect();
  84         LOG.debug("Client created.");
  85         return sessionPromise;
  86     }
  87
  88     /**
  89      * @return true if initial connection was established successfully, false if initial connection failed due
  90      *         to e.g. Connection refused, Negotiation failed
  91      */
  92     private  synchronized boolean isInitialConnectFinished() {
  93         Preconditions.checkNotNull(this.pending);
  94         return this.pending.isDone() && this.pending.isSuccess();
  95     }
  96
  97     private synchronized void reconnect() {
  98         Preconditions.checkNotNull(this.pending);
  99         this.pending.reconnect();
 100     }
 101
 102     @Override
 103     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
 104         if (super.cancel(mayInterruptIfRunning)) {
 105             Preconditions.checkNotNull(this.pending);
 106             this.pending.cancel(mayInterruptIfRunning);
 107             return true;
 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         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 (this.promise.isCancelled()) {
 127                 return;
 128             }
 129
 130             if (!this.promise.isInitialConnectFinished()) {
 131                 LOG.debug("Connection to {} was dropped during negotiation, reattempting", this.promise.address);
 132                 this.promise.reconnect();
 133                 return;
 134             }
 135
 136             LOG.debug("Reconnecting after connection to {} was dropped", this.promise.address);
 137             this.promise.connect();
 138         }
 139     }
 140 }