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.netconf.nettyutil;
10 import static java.util.Objects.requireNonNull;
12 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
13 import io.netty.bootstrap.Bootstrap;
14 import io.netty.channel.ChannelHandlerContext;
15 import io.netty.channel.ChannelInboundHandlerAdapter;
16 import io.netty.util.concurrent.DefaultPromise;
17 import io.netty.util.concurrent.EventExecutor;
18 import io.netty.util.concurrent.Future;
19 import java.net.InetSocketAddress;
20 import org.opendaylight.netconf.api.NetconfSession;
21 import org.opendaylight.netconf.api.NetconfSessionListener;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
26 final class ReconnectPromise<S extends NetconfSession, L extends NetconfSessionListener<? super S>>
27 extends DefaultPromise<Void> {
28 private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
30 private final AbstractNetconfDispatcher<S, L> dispatcher;
31 private final InetSocketAddress address;
32 private final ReconnectStrategyFactory strategyFactory;
33 private final Bootstrap bootstrap;
34 private final AbstractNetconfDispatcher.PipelineInitializer<S> initializer;
35 private Future<?> pending;
37 ReconnectPromise(final EventExecutor executor, final AbstractNetconfDispatcher<S, L> dispatcher,
38 final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
39 final Bootstrap bootstrap, final AbstractNetconfDispatcher.PipelineInitializer<S> initializer) {
41 this.bootstrap = bootstrap;
42 this.initializer = requireNonNull(initializer);
43 this.dispatcher = requireNonNull(dispatcher);
44 this.address = requireNonNull(address);
45 this.strategyFactory = requireNonNull(connectStrategyFactory);
48 synchronized void connect() {
49 final ReconnectStrategy cs = this.strategyFactory.createReconnectStrategy();
51 // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support
53 pending = this.dispatcher.createClient(this.address, cs, bootstrap, (channel, promise) -> {
54 initializer.initializeChannel(channel, promise);
55 // add closed channel handler
56 // This handler has to be added as last channel handler and the channel inactive event has to be caught by
58 // Handlers in front of it can react to channelInactive event, but have to forward the event or the
59 // reconnect will not work
60 // This handler is last so all handlers in front of it can handle channel inactive (to e.g. resource
61 // cleanup) before a new connection is started
62 channel.pipeline().addLast(new ClosedChannelHandler(ReconnectPromise.this));
65 pending.addListener(future -> {
66 if (!future.isSuccess() && !ReconnectPromise.this.isDone()) {
67 ReconnectPromise.this.setFailure(future.cause());
73 * Indicate if the initial connection succeeded.
75 * @return true if initial connection was established successfully, false if initial connection failed due to e.g.
76 * Connection refused, Negotiation failed
78 @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
79 justification = "https://github.com/spotbugs/spotbugs/issues/811")
80 private synchronized boolean isInitialConnectFinished() {
81 requireNonNull(pending);
82 return pending.isDone() && pending.isSuccess();
86 public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
87 if (super.cancel(mayInterruptIfRunning)) {
88 requireNonNull(pending);
89 this.pending.cancel(mayInterruptIfRunning);
97 * Channel handler that responds to channelInactive event and reconnects the session.
98 * Only if the promise was not canceled.
100 private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
101 private final ReconnectPromise<?, ?> promise;
103 ClosedChannelHandler(final ReconnectPromise<?, ?> promise) {
104 this.promise = promise;
108 public void channelInactive(final ChannelHandlerContext ctx) {
109 // This is the ultimate channel inactive handler, not forwarding
110 if (promise.isCancelled()) {
114 if (promise.isInitialConnectFinished() == false) {
115 LOG.debug("Connection to {} was dropped during negotiation, reattempting", promise.address);
118 LOG.debug("Reconnecting after connection to {} was dropped", promise.address);