bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/protocol/BGPProtocolSessionPromise.java

   1 /*
   2  * Copyright (c) 2015 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.bgp.rib.impl.protocol;
   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 org.opendaylight.protocol.bgp.rib.spi.BGPSession;
  23 import org.opendaylight.protocol.framework.ReconnectStrategy;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 public class BGPProtocolSessionPromise<S extends BGPSession> extends DefaultPromise<S> {
  28     private static final Logger LOG = LoggerFactory.getLogger(BGPProtocolSessionPromise.class);
  29     private final ReconnectStrategy strategy;
  30     private final Bootstrap b;
  31
  32     private InetSocketAddress address;
  33     @GuardedBy("this")
  34     private Future<?> pending;
  35
  36     public BGPProtocolSessionPromise(EventExecutor executor, InetSocketAddress address, ReconnectStrategy strategy, Bootstrap b) {
  37         super(executor);
  38         this.strategy = Preconditions.checkNotNull(strategy);
  39         this.address = Preconditions.checkNotNull(address);
  40         this.b = Preconditions.checkNotNull(b);
  41     }
  42
  43     public synchronized void connect() {
  44         final BGPProtocolSessionPromise lock = this;
  45
  46         try {
  47             int e = this.strategy.getConnectTimeout();
  48             LOG.debug("Promise {} attempting connect for {}ms", lock, Integer.valueOf(e));
  49             if (this.address.isUnresolved()) {
  50                 this.address = new InetSocketAddress(this.address.getHostName(), this.address.getPort());
  51             }
  52
  53             this.b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, e);
  54             final ChannelFuture connectFuture = this.b.connect(this.address);
  55             connectFuture.addListener(new BGPProtocolSessionPromise.BootstrapConnectListener(lock));
  56             this.pending = connectFuture;
  57         } catch (Exception e) {
  58             LOG.info("Failed to connect to {}", this.address, e);
  59             this.setFailure(e);
  60         }
  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         } else {
  70             return false;
  71         }
  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 (this.lock) {
  91                 BGPProtocolSessionPromise.LOG.debug("Promise {} connection resolved", this.lock);
  92                 Preconditions.checkState(BGPProtocolSessionPromise.this.pending.equals(cf));
  93                 if (BGPProtocolSessionPromise.this.isCancelled()) {
  94                     if (cf.isSuccess()) {
  95                         BGPProtocolSessionPromise.LOG.debug("Closing channel for cancelled promise {}", this.lock);
  96                         cf.channel().close();
  97                     }
  98
  99                 } else if (cf.isSuccess()) {
 100                     BGPProtocolSessionPromise.LOG.debug("Promise {} connection successful", this.lock);
 101                 } else {
 102                     BGPProtocolSessionPromise.LOG.debug("Attempt to connect to {} failed", BGPProtocolSessionPromise.this.address, cf.cause());
 103                     final Future rf = BGPProtocolSessionPromise.this.strategy.scheduleReconnect(cf.cause());
 104                     rf.addListener(new BGPProtocolSessionPromise.BootstrapConnectListener.ReconnectingStrategyListener());
 105                     BGPProtocolSessionPromise.this.pending = rf;
 106                 }
 107             }
 108         }
 109
 110         private final class ReconnectingStrategyListener implements FutureListener<Void> {
 111             private ReconnectingStrategyListener() {
 112             }
 113
 114             @Override
 115             public void operationComplete(final Future<Void> sf) {
 116                 synchronized (BootstrapConnectListener.this.lock) {
 117                     Preconditions.checkState(BGPProtocolSessionPromise.this.pending.equals(sf));
 118                     if (!BGPProtocolSessionPromise.this.isCancelled()) {
 119                         if (sf.isSuccess()) {
 120                             BGPProtocolSessionPromise.this.connect();
 121                         } else {
 122                             BGPProtocolSessionPromise.this.setFailure(sf.cause());
 123                         }
 124                     }
 125
 126                 }
 127             }
 128         }
 129     }
 130 }