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 com.google.common.base.Preconditions;
11 import io.netty.bootstrap.Bootstrap;
12 import io.netty.channel.ChannelHandler;
13 import io.netty.channel.ChannelHandlerContext;
14 import io.netty.channel.ChannelInboundHandlerAdapter;
15 import io.netty.channel.ChannelInitializer;
16 import io.netty.channel.socket.SocketChannel;
17 import io.netty.util.concurrent.DefaultPromise;
18 import io.netty.util.concurrent.EventExecutor;
19 import io.netty.util.concurrent.Future;
20 import io.netty.util.concurrent.GenericFutureListener;
21 import io.netty.util.concurrent.Promise;
22 import java.net.InetSocketAddress;
23 import org.opendaylight.protocol.bgp.rib.impl.spi.ChannelPipelineInitializer;
24 import org.opendaylight.protocol.bgp.rib.spi.BGPSession;
25 import org.opendaylight.protocol.framework.ReconnectStrategy;
26 import org.opendaylight.protocol.framework.ReconnectStrategyFactory;
27 import org.slf4j.Logger;
28 import org.slf4j.LoggerFactory;
30 public class BGPReconnectPromise<S extends BGPSession> extends DefaultPromise<Void> {
31 private static final Logger LOG = LoggerFactory.getLogger(BGPReconnectPromise.class);
33 private final InetSocketAddress address;
34 private final ReconnectStrategyFactory strategyFactory;
35 private final Bootstrap bootstrap;
36 private final ChannelPipelineInitializer initializer;
37 private Future<S> pending;
39 public BGPReconnectPromise(final EventExecutor executor, final InetSocketAddress address,
40 final ReconnectStrategyFactory connectStrategyFactory, final Bootstrap bootstrap,
41 final ChannelPipelineInitializer initializer) {
43 this.bootstrap = bootstrap;
44 this.initializer = Preconditions.checkNotNull(initializer);
45 this.address = Preconditions.checkNotNull(address);
46 this.strategyFactory = Preconditions.checkNotNull(connectStrategyFactory);
49 public synchronized void connect() {
50 final ReconnectStrategy reconnectStrategy = this.strategyFactory.createReconnectStrategy();
52 // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
53 this.pending = connectSessionPromise(this.address, reconnectStrategy, this.bootstrap, new ChannelPipelineInitializer<S>() {
55 public void initializeChannel(final SocketChannel channel, final Promise<S> promise) {
56 BGPReconnectPromise.this.initializer.initializeChannel(channel, promise);
57 // add closed channel handler
58 // This handler has to be added as last channel handler and the channel inactive event has to be caught by it
59 // Handlers in front of it can react to channelInactive event, but have to forward the event or the reconnect will not work
60 // 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
61 channel.pipeline().addLast(new ClosedChannelHandler(BGPReconnectPromise.this));
65 this.pending.addListener(new GenericFutureListener<Future<Object>>() {
67 public void operationComplete(final Future<Object> future) throws Exception {
68 if (!future.isSuccess()) {
69 BGPReconnectPromise.this.setFailure(future.cause());
75 public Future<S> connectSessionPromise(final InetSocketAddress address, final ReconnectStrategy strategy, final Bootstrap bootstrap,
76 final ChannelPipelineInitializer initializer) {
77 final BGPProtocolSessionPromise sessionPromise = new BGPProtocolSessionPromise(address, strategy, bootstrap);
78 final ChannelHandler chInit = new ChannelInitializer<SocketChannel>() {
80 protected void initChannel(final SocketChannel channel) {
81 initializer.initializeChannel(channel, sessionPromise);
85 bootstrap.handler(chInit);
86 sessionPromise.connect();
87 LOG.debug("Client created.");
88 return sessionPromise;
92 * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
94 private boolean isInitialConnectFinished() {
95 Preconditions.checkNotNull(this.pending);
96 return this.pending.isDone() && this.pending.isSuccess();
100 public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
101 if (super.cancel(mayInterruptIfRunning)) {
102 Preconditions.checkNotNull(this.pending);
103 this.pending.cancel(mayInterruptIfRunning);
110 * Channel handler that responds to channelInactive event and reconnects the session.
111 * Only if the promise was not canceled.
113 private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
114 private final BGPReconnectPromise promise;
116 public ClosedChannelHandler(final BGPReconnectPromise promise) {
117 this.promise = promise;
121 public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
122 // This is the ultimate channel inactive handler, not forwarding
123 if (this.promise.isCancelled()) {
127 if (!this.promise.isInitialConnectFinished()) {
128 LOG.debug("Connection to {} was dropped during negotiation", this.promise.address);
132 LOG.debug("Reconnecting after connection to {} was dropped", this.promise.address);
133 this.promise.connect();