Merge "Introduce message bus models"
[controller.git] / opendaylight / commons / protocol-framework / src / main / java / org / opendaylight / protocol / framework / ReconnectPromise.java
1 /*
2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.protocol.framework;
9
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.channel.socket.SocketChannel;
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.GenericFutureListener;
19 import io.netty.util.concurrent.Promise;
20 import java.net.InetSocketAddress;
21 import org.slf4j.Logger;
22 import org.slf4j.LoggerFactory;
23
24 @Deprecated
25 final class ReconnectPromise<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> extends DefaultPromise<Void> {
26     private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
27
28     private final AbstractDispatcher<S, L> dispatcher;
29     private final InetSocketAddress address;
30     private final ReconnectStrategyFactory strategyFactory;
31     private final Bootstrap b;
32     private final AbstractDispatcher.PipelineInitializer<S> initializer;
33     private Future<?> pending;
34
35     public ReconnectPromise(final EventExecutor executor, final AbstractDispatcher<S, L> dispatcher, final InetSocketAddress address,
36                             final ReconnectStrategyFactory connectStrategyFactory, final Bootstrap b, final AbstractDispatcher.PipelineInitializer<S> initializer) {
37         super(executor);
38         this.b = b;
39         this.initializer = Preconditions.checkNotNull(initializer);
40         this.dispatcher = Preconditions.checkNotNull(dispatcher);
41         this.address = Preconditions.checkNotNull(address);
42         this.strategyFactory = Preconditions.checkNotNull(connectStrategyFactory);
43     }
44
45     synchronized void connect() {
46         final ReconnectStrategy cs = this.strategyFactory.createReconnectStrategy();
47
48         // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
49         pending = this.dispatcher.createClient(this.address, cs, b, new AbstractDispatcher.PipelineInitializer<S>() {
50             @Override
51             public void initializeChannel(final SocketChannel channel, final Promise<S> promise) {
52                 initializer.initializeChannel(channel, promise);
53                 // add closed channel handler
54                 // This handler has to be added as last channel handler and the channel inactive event has to be caught by it
55                 // Handlers in front of it can react to channelInactive event, but have to forward the event or the reconnect will not work
56                 // 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
57                 channel.pipeline().addLast(new ClosedChannelHandler(ReconnectPromise.this));
58             }
59         });
60
61         pending.addListener(new GenericFutureListener<Future<Object>>() {
62             @Override
63             public void operationComplete(Future<Object> future) throws Exception {
64                 if (!future.isSuccess()) {
65                     ReconnectPromise.this.setFailure(future.cause());
66                 }
67             }
68         });
69     }
70
71     /**
72      *
73      * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
74      */
75     private boolean isInitialConnectFinished() {
76         Preconditions.checkNotNull(pending);
77         return pending.isDone() && pending.isSuccess();
78     }
79
80     @Override
81     public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
82         if (super.cancel(mayInterruptIfRunning)) {
83             Preconditions.checkNotNull(pending);
84             this.pending.cancel(mayInterruptIfRunning);
85             return true;
86         }
87
88         return false;
89     }
90
91     /**
92      * Channel handler that responds to channelInactive event and reconnects the session.
93      * Only if the promise was not canceled.
94      */
95     private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
96         private final ReconnectPromise<?, ?> promise;
97
98         public ClosedChannelHandler(final ReconnectPromise<?, ?> promise) {
99             this.promise = promise;
100         }
101
102         @Override
103         public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
104             // This is the ultimate channel inactive handler, not forwarding
105             if (promise.isCancelled()) {
106                 return;
107             }
108
109             if (promise.isInitialConnectFinished() == false) {
110                 LOG.debug("Connection to {} was dropped during negotiation, reattempting", promise.address);
111             }
112
113             LOG.debug("Reconnecting after connection to {} was dropped", promise.address);
114             promise.connect();
115         }
116     }
117
118 }