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

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.