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 io.netty.bootstrap.Bootstrap;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelInboundHandlerAdapter;
  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.checkerframework.checker.lock.qual.GuardedBy;
  21 import org.checkerframework.checker.lock.qual.Holding;
  22 import org.opendaylight.netconf.api.NetconfSession;
  23 import org.opendaylight.netconf.api.NetconfSessionListener;
  24 import org.opendaylight.netconf.nettyutil.AbstractNetconfDispatcher.PipelineInitializer;
  25 import org.opendaylight.yangtools.yang.common.Empty;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 @Deprecated
  30 final class ReconnectPromise<S extends NetconfSession, L extends NetconfSessionListener<? super S>>
  31         extends DefaultPromise<Empty> implements ReconnectFuture {
  32     private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
  33
  34     private final AbstractNetconfDispatcher<S, L> dispatcher;
  35     private final InetSocketAddress address;
  36     private final ReconnectStrategyFactory strategyFactory;
  37     private final Bootstrap bootstrap;
  38     private final PipelineInitializer<S> initializer;
  39     private final Promise<Empty> firstSessionFuture;
  40
  41     @GuardedBy("this")
  42     private Future<?> pending;
  43
  44     ReconnectPromise(final EventExecutor executor, final AbstractNetconfDispatcher<S, L> dispatcher,
  45             final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
  46             final Bootstrap bootstrap, final PipelineInitializer<S> initializer) {
  47         super(executor);
  48         this.firstSessionFuture = new DefaultPromise<>(executor);
  49         this.bootstrap = requireNonNull(bootstrap);
  50         this.initializer = requireNonNull(initializer);
  51         this.dispatcher = requireNonNull(dispatcher);
  52         this.address = requireNonNull(address);
  53         this.strategyFactory = requireNonNull(connectStrategyFactory);
  54     }
  55
  56     @Override
  57     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
  58         if (super.cancel(mayInterruptIfRunning)) {
  59             firstSessionFuture.cancel(mayInterruptIfRunning);
  60             pending.cancel(mayInterruptIfRunning);
  61             return true;
  62         }
  63         return false;
  64     }
  65
  66     @Override
  67     public Future<?> firstSessionFuture() {
  68         return firstSessionFuture;
  69     }
  70
  71     synchronized void connect() {
  72         lockedConnect();
  73     }
  74
  75     @Holding("this")
  76     private void lockedConnect() {
  77         final ReconnectStrategy cs = strategyFactory.createReconnectStrategy();
  78
  79         // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support
  80         // reconnect attempts
  81         pending = dispatcher.createClient(address, cs, bootstrap, (channel, promise) -> {
  82             initializer.initializeChannel(channel, promise);
  83             // add closed channel handler
  84             // This handler has to be added as last channel handler and the channel inactive event has to be caught by
  85             // it
  86             // Handlers in front of it can react to channelInactive event, but have to forward the event or the
  87             // reconnect will not work
  88             // This handler is last so all handlers in front of it can handle channel inactive (to e.g. resource
  89             // cleanup) before a new connection is started
  90             channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  91                 @Override
  92                 public void channelInactive(final ChannelHandlerContext ctx) {
  93                     onChannelInactive();
  94                 }
  95             });
  96         });
  97
  98         if (!firstSessionFuture.isDone()) {
  99             pending.addListener(future -> {
 100                 if (!future.isSuccess() && !firstSessionFuture.isDone()) {
 101                     firstSessionFuture.setFailure(future.cause());
 102                 }
 103             });
 104         }
 105     }
 106
 107     private void onChannelInactive() {
 108         // This is the ultimate channel inactive handler, not forwarding
 109         if (isCancelled()) {
 110             return;
 111         }
 112
 113         synchronized (this) {
 114             final Future<?> attempt = pending;
 115             if (!attempt.isDone() || !attempt.isSuccess()) {
 116                 // Connection refused, negotiation failed, or similar
 117                 LOG.debug("Connection to {} was dropped during negotiation, reattempting", address);
 118             }
 119
 120             LOG.debug("Reconnecting after connection to {} was dropped", address);
 121             lockedConnect();
 122         }
 123     }
 124 }