netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/NetconfSessionPromise.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 com.google.common.base.Verify.verify;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import io.netty.bootstrap.Bootstrap;
  14 import io.netty.channel.ChannelFuture;
  15 import io.netty.channel.ChannelFutureListener;
  16 import io.netty.util.concurrent.DefaultPromise;
  17 import io.netty.util.concurrent.EventExecutor;
  18 import io.netty.util.concurrent.Future;
  19 import io.netty.util.concurrent.Promise;
  20 import java.net.InetSocketAddress;
  21 import org.checkerframework.checker.lock.qual.GuardedBy;
  22 import org.opendaylight.netconf.api.NetconfSession;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 @Deprecated
  27 final class NetconfSessionPromise<S extends NetconfSession> extends DefaultPromise<S> {
  28     private static final Logger LOG = LoggerFactory.getLogger(NetconfSessionPromise.class);
  29     private final ReconnectStrategy strategy;
  30     private InetSocketAddress address;
  31     private final Bootstrap bootstrap;
  32
  33     @GuardedBy("this")
  34     private Future<?> pending;
  35
  36     NetconfSessionPromise(final EventExecutor executor, final InetSocketAddress address,
  37             final ReconnectStrategy strategy, final Bootstrap bootstrap) {
  38         super(executor);
  39         this.strategy = requireNonNull(strategy);
  40         this.address = requireNonNull(address);
  41         this.bootstrap = requireNonNull(bootstrap);
  42     }
  43
  44     @SuppressWarnings("checkstyle:illegalCatch")
  45     synchronized void connect() {
  46         final int timeout;
  47         try {
  48             timeout = strategy.getConnectTimeout();
  49         } catch (Exception e) {
  50             LOG.info("Connection to {} aborted due to strategy decision", address, e);
  51             setFailure(e);
  52             return;
  53         }
  54
  55         LOG.debug("Promise {} attempting connect for {}ms", this, timeout);
  56
  57         final ChannelFuture connectFuture;
  58         try {
  59             if (address.isUnresolved()) {
  60                 address = new InetSocketAddress(address.getHostName(), address.getPort());
  61             }
  62             connectFuture = bootstrap.connect(address);
  63         } catch (final Exception e) {
  64             LOG.info("Failed to connect to {}", address, e);
  65             setFailure(e);
  66             return;
  67         }
  68
  69         pending = connectFuture;
  70         // Add listener that attempts reconnect by invoking this method again.
  71         connectFuture.addListener((ChannelFutureListener) this::channelConnectComplete);
  72     }
  73
  74     @Override
  75     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
  76         if (super.cancel(mayInterruptIfRunning)) {
  77             pending.cancel(mayInterruptIfRunning);
  78             return true;
  79         }
  80
  81         return false;
  82     }
  83
  84     @Override
  85     public synchronized Promise<S> setSuccess(final S result) {
  86         LOG.debug("Promise {} completed", this);
  87         strategy.reconnectSuccessful();
  88         return super.setSuccess(result);
  89     }
  90
  91     // Triggered when a connection attempt is resolved.
  92     private synchronized void channelConnectComplete(final ChannelFuture cf) {
  93         LOG.debug("Promise {} connection resolved", this);
  94         verify(pending == cf, "Completed channel future %s while pending %s", cf, pending);
  95
  96         /*
  97          * The promise we gave out could have been cancelled,
  98          * which cascades to the connect getting cancelled,
  99          * but there is a slight race window, where the connect
 100          * is already resolved, but the listener has not yet
 101          * been notified -- cancellation at that point won't
 102          * stop the notification arriving, so we have to close
 103          * the race here.
 104          */
 105         if (isCancelled()) {
 106             if (cf.isSuccess()) {
 107                 LOG.debug("Closing channel for cancelled promise {}", this);
 108                 cf.channel().close();
 109             }
 110             return;
 111         }
 112
 113         if (cf.isSuccess()) {
 114             LOG.debug("Promise {} connection successful", this);
 115             return;
 116         }
 117
 118         LOG.debug("Attempt to connect to {} failed", address, cf.cause());
 119
 120         final Future<Void> rf = strategy.scheduleReconnect(cf.cause());
 121         pending = rf;
 122         rf.addListener(this::reconnectFutureComplete);
 123     }
 124
 125     // Triggered when a connection attempt is to be made.
 126     private synchronized void reconnectFutureComplete(final Future<?> sf) {
 127         LOG.debug("Promise {} strategy triggered reconnect", this);
 128         verify(pending == sf, "Completed strategy future %s while pending %s", sf, pending);
 129
 130         /*
 131          * The promise we gave out could have been cancelled,
 132          * which cascades to the reconnect attempt getting
 133          * cancelled, but there is a slight race window, where
 134          * the reconnect attempt is already enqueued, but the
 135          * listener has not yet been notified -- if cancellation
 136          * happens at that point, we need to catch it here.
 137          */
 138         if (!isCancelled()) {
 139             if (sf.isSuccess()) {
 140                 connect();
 141             } else {
 142                 setFailure(sf.cause());
 143             }
 144         }
 145     }
 146 }