pcep/impl/src/main/java/org/opendaylight/protocol/pcep/impl/PCEPProtocolSessionPromise.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.pcep.impl;
   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.protocol.framework.ReconnectStrategy;
  24 import org.opendaylight.protocol.pcep.PCEPSession;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28 @ThreadSafe
  29 public final class PCEPProtocolSessionPromise<S extends PCEPSession> extends DefaultPromise<S> {
  30     private static final Logger LOG = LoggerFactory.getLogger(PCEPProtocolSessionPromise.class);
  31     private final ReconnectStrategy strategy;
  32     private final Bootstrap b;
  33     private InetSocketAddress address;
  34     @GuardedBy("this")
  35     private Future<?> pending;
  36
  37     PCEPProtocolSessionPromise(final EventExecutor executor, final InetSocketAddress address, final ReconnectStrategy
  38         strategy, 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 PCEPProtocolSessionPromise lock = this;
  47
  48         try {
  49             final int e = this.strategy.getConnectTimeout();
  50             LOG.debug("Promise {} attempting connect for {}ms", lock, Integer.valueOf(e));
  51             if (this.address.isUnresolved()) {
  52                 this.address = new InetSocketAddress(this.address.getHostName(), this.address.getPort());
  53             }
  54
  55             this.b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, e);
  56             final ChannelFuture connectFuture = this.b.connect(this.address);
  57             connectFuture.addListener(new PCEPProtocolSessionPromise.BootstrapConnectListener(lock));
  58             this.pending = connectFuture;
  59         } catch (Exception e) {
  60             LOG.info("Failed to connect to {}", this.address, e);
  61             this.setFailure(e);
  62         }
  63
  64     }
  65
  66     @Override
  67     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
  68         if (super.cancel(mayInterruptIfRunning)) {
  69             this.pending.cancel(mayInterruptIfRunning);
  70             return true;
  71         } else {
  72             return false;
  73         }
  74     }
  75
  76     @Override
  77     public synchronized Promise<S> setSuccess(final S result) {
  78         LOG.debug("Promise {} completed", this);
  79         this.strategy.reconnectSuccessful();
  80         return super.setSuccess(result);
  81     }
  82
  83     private class BootstrapConnectListener implements ChannelFutureListener {
  84         private final Object lock;
  85
  86         public BootstrapConnectListener(final Object lock) {
  87             this.lock = lock;
  88         }
  89
  90         @Override
  91         public void operationComplete(final ChannelFuture cf) throws Exception {
  92             synchronized (this.lock) {
  93                 PCEPProtocolSessionPromise.LOG.debug("Promise {} connection resolved", this.lock);
  94                 Preconditions.checkState(PCEPProtocolSessionPromise.this.pending.equals(cf));
  95                 if (PCEPProtocolSessionPromise.this.isCancelled()) {
  96                     if (cf.isSuccess()) {
  97                         PCEPProtocolSessionPromise.LOG.debug("Closing channel for cancelled promise {}", this.lock);
  98                         cf.channel().close();
  99                     }
 100
 101                 } else if (cf.isSuccess()) {
 102                     PCEPProtocolSessionPromise.LOG.debug("Promise {} connection successful", this.lock);
 103                 } else {
 104                     PCEPProtocolSessionPromise.LOG.debug("Attempt to connect to {} failed", PCEPProtocolSessionPromise.this.address, cf.cause());
 105                     final Future rf = PCEPProtocolSessionPromise.this.strategy.scheduleReconnect(cf.cause());
 106                     rf.addListener(new PCEPProtocolSessionPromise.BootstrapConnectListener.ReconnectingStrategyListener());
 107                     PCEPProtocolSessionPromise.this.pending = rf;
 108                 }
 109             }
 110         }
 111
 112         private final class ReconnectingStrategyListener implements FutureListener<Void> {
 113             private ReconnectingStrategyListener() {
 114             }
 115
 116             @Override
 117             public void operationComplete(final Future<Void> sf) {
 118                 synchronized (BootstrapConnectListener.this.lock) {
 119                     Preconditions.checkState(PCEPProtocolSessionPromise.this.pending.equals(sf));
 120                     if (!PCEPProtocolSessionPromise.this.isCancelled()) {
 121                         if (sf.isSuccess()) {
 122                             PCEPProtocolSessionPromise.this.connect();
 123                         } else {
 124                             PCEPProtocolSessionPromise.this.setFailure(sf.cause());
 125                         }
 126                     }
 127
 128                 }
 129             }
 130         }
 131     }
 132 }