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 io.netty.bootstrap.Bootstrap;
13 import io.netty.channel.ChannelHandlerContext;
14 import io.netty.channel.ChannelInboundHandlerAdapter;
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.Promise;
19 import java.net.InetSocketAddress;
20 import org.checkerframework.checker.lock.qual.GuardedBy;
21 import org.checkerframework.checker.lock.qual.Holding;
22 import org.opendaylight.netconf.api.NetconfSession;
23 import org.opendaylight.netconf.api.NetconfSessionListener;
24 import org.opendaylight.netconf.nettyutil.AbstractNetconfDispatcher.PipelineInitializer;
25 import org.opendaylight.yangtools.yang.common.Empty;
26 import org.slf4j.Logger;
27 import org.slf4j.LoggerFactory;
30 final class ReconnectPromise<S extends NetconfSession, L extends NetconfSessionListener<? super S>>
31 extends DefaultPromise<Empty> implements ReconnectFuture {
32 private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
34 private final AbstractNetconfDispatcher<S, L> dispatcher;
35 private final InetSocketAddress address;
36 private final ReconnectStrategyFactory strategyFactory;
37 private final Bootstrap bootstrap;
38 private final PipelineInitializer<S> initializer;
39 private final Promise<Empty> firstSessionFuture;
42 private Future<?> pending;
44 ReconnectPromise(final EventExecutor executor, final AbstractNetconfDispatcher<S, L> dispatcher,
45 final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
46 final Bootstrap bootstrap, final PipelineInitializer<S> initializer) {
48 this.firstSessionFuture = new DefaultPromise<>(executor);
49 this.bootstrap = requireNonNull(bootstrap);
50 this.initializer = requireNonNull(initializer);
51 this.dispatcher = requireNonNull(dispatcher);
52 this.address = requireNonNull(address);
53 this.strategyFactory = requireNonNull(connectStrategyFactory);
57 public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
58 if (super.cancel(mayInterruptIfRunning)) {
59 firstSessionFuture.cancel(mayInterruptIfRunning);
60 pending.cancel(mayInterruptIfRunning);
67 public Future<?> firstSessionFuture() {
68 return firstSessionFuture;
71 synchronized void connect() {
76 private void lockedConnect() {
77 final ReconnectStrategy cs = strategyFactory.createReconnectStrategy();
79 // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support
81 pending = dispatcher.createClient(address, cs, bootstrap, (channel, promise) -> {
82 initializer.initializeChannel(channel, promise);
83 // add closed channel handler
84 // This handler has to be added as last channel handler and the channel inactive event has to be caught by
86 // Handlers in front of it can react to channelInactive event, but have to forward the event or the
87 // reconnect will not work
88 // This handler is last so all handlers in front of it can handle channel inactive (to e.g. resource
89 // cleanup) before a new connection is started
90 channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
92 public void channelInactive(final ChannelHandlerContext ctx) {
98 if (!firstSessionFuture.isDone()) {
99 pending.addListener(future -> {
100 if (!future.isSuccess() && !firstSessionFuture.isDone()) {
101 firstSessionFuture.setFailure(future.cause());
107 private void onChannelInactive() {
108 // This is the ultimate channel inactive handler, not forwarding
113 synchronized (this) {
114 final Future<?> attempt = pending;
115 if (!attempt.isDone() || !attempt.isSuccess()) {
116 // Connection refused, negotiation failed, or similar
117 LOG.debug("Connection to {} was dropped during negotiation, reattempting", address);
120 LOG.debug("Reconnecting after connection to {} was dropped", address);