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