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 com.google.common.base.Preconditions;
11 import io.netty.bootstrap.Bootstrap;
12 import io.netty.channel.ChannelHandlerContext;
13 import io.netty.channel.ChannelInboundHandlerAdapter;
14 import io.netty.util.concurrent.DefaultPromise;
15 import io.netty.util.concurrent.EventExecutor;
16 import io.netty.util.concurrent.Future;
17 import java.net.InetSocketAddress;
18 import org.slf4j.Logger;
19 import org.slf4j.LoggerFactory;
22 final class ReconnectPromise<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> extends DefaultPromise<Void> {
23 private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
25 private final AbstractDispatcher<S, L> dispatcher;
26 private final InetSocketAddress address;
27 private final ReconnectStrategyFactory strategyFactory;
28 private final Bootstrap b;
29 private final AbstractDispatcher.PipelineInitializer<S> initializer;
30 private Future<?> pending;
32 public ReconnectPromise(final EventExecutor executor, final AbstractDispatcher<S, L> dispatcher, final InetSocketAddress address,
33 final ReconnectStrategyFactory connectStrategyFactory, final Bootstrap b, final AbstractDispatcher.PipelineInitializer<S> initializer) {
36 this.initializer = Preconditions.checkNotNull(initializer);
37 this.dispatcher = Preconditions.checkNotNull(dispatcher);
38 this.address = Preconditions.checkNotNull(address);
39 this.strategyFactory = Preconditions.checkNotNull(connectStrategyFactory);
42 synchronized void connect() {
43 final ReconnectStrategy cs = this.strategyFactory.createReconnectStrategy();
45 // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
46 pending = this.dispatcher.createClient(this.address, cs, b, (channel, promise) -> {
47 initializer.initializeChannel(channel, promise);
48 // add closed channel handler
49 // This handler has to be added as last channel handler and the channel inactive event has to be caught by it
50 // Handlers in front of it can react to channelInactive event, but have to forward the event or the reconnect will not work
51 // This handler is last so all handlers in front of it can handle channel inactive (to e.g. resource cleanup) before a new connection is started
52 channel.pipeline().addLast(new ClosedChannelHandler(ReconnectPromise.this));
55 pending.addListener(future -> {
56 if (!future.isSuccess() && !ReconnectPromise.this.isDone()) {
57 ReconnectPromise.this.setFailure(future.cause());
64 * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
66 private boolean isInitialConnectFinished() {
67 Preconditions.checkNotNull(pending);
68 return pending.isDone() && pending.isSuccess();
72 public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
73 if (super.cancel(mayInterruptIfRunning)) {
74 Preconditions.checkNotNull(pending);
75 this.pending.cancel(mayInterruptIfRunning);
83 * Channel handler that responds to channelInactive event and reconnects the session.
84 * Only if the promise was not canceled.
86 private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
87 private final ReconnectPromise<?, ?> promise;
89 public ClosedChannelHandler(final ReconnectPromise<?, ?> promise) {
90 this.promise = promise;
94 public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
95 // This is the ultimate channel inactive handler, not forwarding
96 if (promise.isCancelled()) {
100 if (promise.isInitialConnectFinished() == false) {
101 LOG.debug("Connection to {} was dropped during negotiation, reattempting", promise.address);
104 LOG.debug("Reconnecting after connection to {} was dropped", promise.address);