2 * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.protocol.framework;
10 import io.netty.bootstrap.Bootstrap;
11 import io.netty.channel.ChannelFuture;
12 import io.netty.channel.ChannelFutureListener;
13 import io.netty.channel.ChannelOption;
14 import io.netty.util.concurrent.DefaultPromise;
15 import io.netty.util.concurrent.EventExecutor;
16 import io.netty.util.concurrent.Future;
17 import io.netty.util.concurrent.FutureListener;
18 import io.netty.util.concurrent.Promise;
20 import java.net.InetSocketAddress;
22 import javax.annotation.concurrent.GuardedBy;
23 import javax.annotation.concurrent.ThreadSafe;
25 import org.slf4j.Logger;
26 import org.slf4j.LoggerFactory;
28 import com.google.common.base.Preconditions;
31 final class ProtocolSessionPromise<S extends ProtocolSession<?>> extends DefaultPromise<S> {
32 private static final Logger LOG = LoggerFactory.getLogger(ProtocolSessionPromise.class);
33 private final ReconnectStrategy strategy;
34 private final InetSocketAddress address;
35 private final Bootstrap b;
38 private Future<?> pending;
40 ProtocolSessionPromise(final EventExecutor executor, final InetSocketAddress address, final ReconnectStrategy strategy,
43 this.strategy = Preconditions.checkNotNull(strategy);
44 this.address = Preconditions.checkNotNull(address);
45 this.b = Preconditions.checkNotNull(b);
48 synchronized void connect() {
49 final Object lock = this;
52 final int timeout = this.strategy.getConnectTimeout();
54 LOG.debug("Promise {} attempting connect for {}ms", lock, timeout);
56 this.b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout);
57 this.pending = this.b.connect(this.address).addListener(new ChannelFutureListener() {
59 public void operationComplete(final ChannelFuture cf) throws Exception {
62 LOG.debug("Promise {} connection resolved", lock);
64 // Triggered when a connection attempt is resolved.
65 Preconditions.checkState(ProtocolSessionPromise.this.pending.equals(cf));
68 * The promise we gave out could have been cancelled,
69 * which cascades to the connect getting cancelled,
70 * but there is a slight race window, where the connect
71 * is already resolved, but the listener has not yet
72 * been notified -- cancellation at that point won't
73 * stop the notification arriving, so we have to close
78 LOG.debug("Closing channel for cancelled promise {}", lock);
84 if (!cf.isSuccess()) {
85 LOG.warn("Attempt to connect to connect to {} failed", ProtocolSessionPromise.this.address, cf.cause());
86 final Future<Void> rf = ProtocolSessionPromise.this.strategy.scheduleReconnect(cf.cause());
87 rf.addListener(new FutureListener<Void>() {
89 public void operationComplete(final Future<Void> sf) {
91 // Triggered when a connection attempt is to be made.
92 Preconditions.checkState(ProtocolSessionPromise.this.pending.equals(sf));
95 * The promise we gave out could have been cancelled,
96 * which cascades to the reconnect attempt getting
97 * cancelled, but there is a slight race window, where
98 * the reconnect attempt is already enqueued, but the
99 * listener has not yet been notified -- if cancellation
100 * happens at that point, we need to catch it here.
102 if (!isCancelled()) {
103 if (sf.isSuccess()) {
106 setFailure(sf.cause());
113 ProtocolSessionPromise.this.pending = rf;
115 LOG.debug("Promise {} connection successful", lock);
120 } catch (final Exception e) {
121 LOG.info("Failed to connect to {}", e);
127 public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
128 if (super.cancel(mayInterruptIfRunning)) {
129 this.pending.cancel(mayInterruptIfRunning);
137 public synchronized Promise<S> setSuccess(final S result) {
138 LOG.debug("Promise {} completed", this);
139 this.strategy.reconnectSuccessful();
140 return super.setSuccess(result);