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