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