2 * Copyright (c) 2015 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.bgp.rib.impl.protocol;
10 import static java.util.Objects.requireNonNull;
12 import io.netty.bootstrap.Bootstrap;
13 import io.netty.channel.ChannelHandler;
14 import io.netty.channel.ChannelHandlerContext;
15 import io.netty.channel.ChannelInboundHandlerAdapter;
16 import io.netty.channel.ChannelInitializer;
17 import io.netty.channel.socket.SocketChannel;
18 import io.netty.util.concurrent.DefaultPromise;
19 import io.netty.util.concurrent.EventExecutor;
20 import java.net.InetSocketAddress;
21 import org.eclipse.jdt.annotation.NonNull;
22 import org.opendaylight.protocol.bgp.rib.impl.spi.BGPPeerRegistry;
23 import org.opendaylight.protocol.bgp.rib.impl.spi.ChannelPipelineInitializer;
24 import org.opendaylight.protocol.bgp.rib.spi.BGPSession;
25 import org.slf4j.Logger;
26 import org.slf4j.LoggerFactory;
28 public class BGPReconnectPromise<S extends BGPSession> extends DefaultPromise<Void> {
29 private static final Logger LOG = LoggerFactory.getLogger(BGPReconnectPromise.class);
31 private final InetSocketAddress address;
32 private final int retryTimer;
33 private final Bootstrap bootstrap;
34 private final BGPPeerRegistry peerRegistry;
35 private final ChannelPipelineInitializer<S> initializer;
36 private BGPProtocolSessionPromise<S> pending;
38 public BGPReconnectPromise(final @NonNull EventExecutor executor, final @NonNull InetSocketAddress address,
39 final int retryTimer, final @NonNull Bootstrap bootstrap, final @NonNull BGPPeerRegistry peerRegistry,
40 final @NonNull ChannelPipelineInitializer<S> initializer) {
42 this.bootstrap = bootstrap;
43 this.initializer = requireNonNull(initializer);
44 this.address = requireNonNull(address);
45 this.retryTimer = retryTimer;
46 this.peerRegistry = requireNonNull(peerRegistry);
49 public synchronized void connect() {
50 if (this.pending != null) {
51 this.pending.cancel(true);
54 // Set up a client with pre-configured bootstrap, but add a closed channel handler
55 // into the pipeline to support reconnect attempts
56 this.pending = connectSessionPromise(this.address, this.retryTimer, this.bootstrap, this.peerRegistry,
57 (channel, promise) -> {
58 this.initializer.initializeChannel(channel, promise);
59 // add closed channel handler
60 // This handler has to be added as last channel handler and the channel inactive event has to be
62 // Handlers in front of it can react to channelInactive event, but have to forward the event or
63 // the reconnect will not work. This handler is last so all handlers in front of it can handle
64 // channel inactive (to e.g. resource cleanup) before a new connection is started
65 channel.pipeline().addLast(new ClosedChannelHandler(this));
68 this.pending.addListener(future -> {
69 if (!future.isSuccess() && !this.isDone()) {
70 this.setFailure(future.cause());
75 private static <S extends BGPSession> BGPProtocolSessionPromise<S> connectSessionPromise(
76 final InetSocketAddress address, final int retryTimer, final Bootstrap bootstrap,
77 final BGPPeerRegistry peerRegistry, final ChannelPipelineInitializer<S> initializer) {
78 final BGPProtocolSessionPromise<S> sessionPromise = new BGPProtocolSessionPromise<>(address, retryTimer,
79 bootstrap, peerRegistry);
80 final ChannelHandler chInit = new ChannelInitializer<SocketChannel>() {
82 protected void initChannel(final SocketChannel channel) {
83 LOG.info("Initializing channel with {}", channel.remoteAddress());
84 initializer.initializeChannel(channel, sessionPromise);
88 bootstrap.handler(chInit);
89 sessionPromise.connect();
90 LOG.debug("Client created.");
91 return sessionPromise;
95 * Indicate whether the initial connection was established successfully.
97 * @return true if initial connection was established successfully, false if initial connection failed due
98 * to e.g. Connection refused, Negotiation failed
100 private synchronized boolean isInitialConnectFinished() {
101 requireNonNull(this.pending);
102 return this.pending.isDone() && this.pending.isSuccess();
105 private synchronized void reconnect() {
106 requireNonNull(this.pending);
107 this.pending.reconnect();
111 public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
112 if (super.cancel(mayInterruptIfRunning)) {
113 requireNonNull(this.pending);
114 this.pending.cancel(mayInterruptIfRunning);
121 * Channel handler that responds to channelInactive event and reconnects the session.
122 * Only if the promise was not canceled.
124 private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
125 private final BGPReconnectPromise<?> promise;
127 ClosedChannelHandler(final BGPReconnectPromise<?> promise) {
128 this.promise = promise;
132 public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
133 // This is the ultimate channel inactive handler, not forwarding
134 if (this.promise.isCancelled()) {
138 if (!this.promise.isInitialConnectFinished()) {
139 LOG.debug("Connection to {} was dropped during negotiation, reattempting", this.promise.address);
140 this.promise.reconnect();
144 LOG.debug("Reconnecting after connection to {} was dropped", this.promise.address);
145 this.promise.connect();