Merge "BUG-190 Simplify reconnect logic in protocol-framework."
[controller.git] / opendaylight / commons / protocol-framework / src / main / java / org / opendaylight / protocol / framework / AbstractDispatcher.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 java.io.Closeable;
11 import java.net.InetSocketAddress;
12 import java.net.SocketAddress;
13
14 import org.slf4j.Logger;
15 import org.slf4j.LoggerFactory;
16
17 import com.google.common.base.Preconditions;
18
19 import io.netty.bootstrap.Bootstrap;
20 import io.netty.bootstrap.ServerBootstrap;
21 import io.netty.buffer.PooledByteBufAllocator;
22 import io.netty.channel.Channel;
23 import io.netty.channel.ChannelFuture;
24 import io.netty.channel.ChannelInitializer;
25 import io.netty.channel.ChannelOption;
26 import io.netty.channel.EventLoopGroup;
27 import io.netty.channel.ServerChannel;
28 import io.netty.channel.local.LocalServerChannel;
29 import io.netty.channel.socket.SocketChannel;
30 import io.netty.channel.socket.nio.NioServerSocketChannel;
31 import io.netty.channel.socket.nio.NioSocketChannel;
32 import io.netty.util.concurrent.DefaultPromise;
33 import io.netty.util.concurrent.EventExecutor;
34 import io.netty.util.concurrent.Future;
35 import io.netty.util.concurrent.GlobalEventExecutor;
36 import io.netty.util.concurrent.Promise;
37
38 /**
39  * Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the
40  * start method that will handle sockets in different thread.
41  */
42 public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> implements Closeable {
43
44
45     protected interface ChannelPipelineInitializer<CH extends Channel, S extends ProtocolSession<?>> {
46         /**
47          * Initializes channel by specifying the handlers in its pipeline. Handlers are protocol specific, therefore this
48          * method needs to be implemented in protocol specific Dispatchers.
49          *
50          * @param channel whose pipeline should be defined, also to be passed to {@link SessionNegotiatorFactory}
51          * @param promise to be passed to {@link SessionNegotiatorFactory}
52          */
53         void initializeChannel(CH channel, Promise<S> promise);
54     }
55
56     protected interface PipelineInitializer<S extends ProtocolSession<?>> extends ChannelPipelineInitializer<SocketChannel, S> {
57
58     }
59
60
61     private static final Logger LOG = LoggerFactory.getLogger(AbstractDispatcher.class);
62
63     private final EventLoopGroup bossGroup;
64
65     private final EventLoopGroup workerGroup;
66
67     private final EventExecutor executor;
68
69     protected AbstractDispatcher(final EventLoopGroup bossGroup, final EventLoopGroup workerGroup) {
70         this(GlobalEventExecutor.INSTANCE, bossGroup, workerGroup);
71     }
72
73     protected AbstractDispatcher(final EventExecutor executor, final EventLoopGroup bossGroup, final EventLoopGroup workerGroup) {
74         this.bossGroup = Preconditions.checkNotNull(bossGroup);
75         this.workerGroup = Preconditions.checkNotNull(workerGroup);
76         this.executor = Preconditions.checkNotNull(executor);
77     }
78
79
80     /**
81      * Creates server. Each server needs factories to pass their instances to client sessions.
82      *
83      * @param address address to which the server should be bound
84      * @param initializer instance of PipelineInitializer used to initialize the channel pipeline
85      *
86      * @return ChannelFuture representing the binding process
87      */
88     protected ChannelFuture createServer(final InetSocketAddress address, final PipelineInitializer<S> initializer) {
89         return createServer(address, NioServerSocketChannel.class, initializer);
90     }
91
92     /**
93      * Creates server. Each server needs factories to pass their instances to client sessions.
94      *
95      * @param address address to which the server should be bound
96      * @param channelClass The {@link Class} which is used to create {@link Channel} instances from.
97      * @param initializer instance of PipelineInitializer used to initialize the channel pipeline
98      *
99      * @return ChannelFuture representing the binding process
100      */
101     protected <CH extends Channel> ChannelFuture createServer(final SocketAddress address, final Class<? extends ServerChannel> channelClass,
102             final ChannelPipelineInitializer<CH, S> initializer) {
103         final ServerBootstrap b = new ServerBootstrap();
104         b.childHandler(new ChannelInitializer<CH>() {
105
106             @Override
107             protected void initChannel(final CH ch) {
108                 initializer.initializeChannel(ch, new DefaultPromise<S>(executor));
109             }
110         });
111
112         b.option(ChannelOption.SO_BACKLOG, 128);
113         if (LocalServerChannel.class.equals(channelClass) == false) {
114             // makes no sense for LocalServer and produces warning
115             b.childOption(ChannelOption.SO_KEEPALIVE, true);
116         }
117         b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
118         customizeBootstrap(b);
119
120         if (b.group() == null) {
121             b.group(bossGroup, workerGroup);
122         }
123         try {
124             b.channel(channelClass);
125         } catch (IllegalStateException e) {
126             // FIXME: if this is ok, document why
127             LOG.trace("Not overriding channelFactory on bootstrap {}", b, e);
128         }
129
130         // Bind and start to accept incoming connections.
131         final ChannelFuture f = b.bind(address);
132         LOG.debug("Initiated server {} at {}.", f, address);
133         return f;
134     }
135
136     /**
137      * Customize a server bootstrap before the server is created. This allows
138      * subclasses to assign non-default server options before the server is
139      * created.
140      *
141      * @param b Server bootstrap
142      */
143     protected void customizeBootstrap(final ServerBootstrap b) {
144         // The default is a no-op
145     }
146
147     /**
148      * Creates a client.
149      *
150      * @param address remote address
151      * @param connectStrategy Reconnection strategy to be used when initial connection fails
152      *
153      * @return Future representing the connection process. Its result represents the combined success of TCP connection
154      *         as well as session negotiation.
155      */
156     protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final PipelineInitializer<S> initializer) {
157         final Bootstrap b = new Bootstrap();
158         final ProtocolSessionPromise<S> p = new ProtocolSessionPromise<>(executor, address, strategy, b);
159         b.option(ChannelOption.SO_KEEPALIVE, true).handler(
160                 new ChannelInitializer<SocketChannel>() {
161                     @Override
162                     protected void initChannel(final SocketChannel ch) {
163                         initializer.initializeChannel(ch, p);
164                     }
165                 });
166
167         customizeBootstrap(b);
168         setWorkerGroup(b);
169         setChannelFactory(b);
170
171         p.connect();
172         LOG.debug("Client created.");
173         return p;
174     }
175
176     private void setWorkerGroup(final Bootstrap b) {
177         if (b.group() == null) {
178             b.group(workerGroup);
179         }
180     }
181
182     /**
183      * Create a client but use a pre-configured bootstrap.
184      * This method however replaces the ChannelInitializer in the bootstrap. All other configuration is preserved.
185      *
186      * @param address remote address
187      */
188     protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final Bootstrap bootstrap, final PipelineInitializer<S> initializer) {
189         final ProtocolSessionPromise<S> p = new ProtocolSessionPromise<>(executor, address, strategy, bootstrap);
190
191         bootstrap.handler(
192                 new ChannelInitializer<SocketChannel>() {
193                     @Override
194                     protected void initChannel(final SocketChannel ch) {
195                         initializer.initializeChannel(ch, p);
196                     }
197                 });
198
199         p.connect();
200         LOG.debug("Client created.");
201         return p;
202     }
203
204     /**
205      * Customize a client bootstrap before the connection is attempted. This
206      * allows subclasses to assign non-default options before the client is
207      * created.
208      *
209      * @param b Client bootstrap
210      */
211     protected void customizeBootstrap(final Bootstrap b) {
212         // The default is a no-op
213     }
214
215     /**
216      *
217      * @deprecated use {@link org.opendaylight.protocol.framework.AbstractDispatcher#createReconnectingClient(java.net.InetSocketAddress, ReconnectStrategyFactory, org.opendaylight.protocol.framework.AbstractDispatcher.PipelineInitializer)} with only one reconnectStrategyFactory instead.
218      *
219      * Creates a client.
220      *
221      * @param address remote address
222      * @param connectStrategyFactory Factory for creating reconnection strategy to be used when initial connection fails
223      * @param reestablishStrategy Reconnection strategy to be used when the already-established session fails
224      *
225      * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
226      *         success if it indicates no further attempts should be made and failure if it reports an error
227      */
228     @Deprecated
229     protected Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
230             final ReconnectStrategy reestablishStrategy, final PipelineInitializer<S> initializer) {
231         return createReconnectingClient(address, connectStrategyFactory, initializer);
232     }
233
234     /**
235      * Creates a reconnecting client.
236      *
237      * @param address remote address
238      * @param connectStrategyFactory Factory for creating reconnection strategy for every reconnect attempt
239      *
240      * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
241      *         success if it indicates no further attempts should be made and failure if it reports an error
242      */
243     protected Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
244             final PipelineInitializer<S> initializer) {
245         final Bootstrap b = new Bootstrap();
246
247         final ReconnectPromise<S, L> p = new ReconnectPromise<>(GlobalEventExecutor.INSTANCE, this, address, connectStrategyFactory, b, initializer);
248
249         b.option(ChannelOption.SO_KEEPALIVE, true);
250
251         customizeBootstrap(b);
252         setWorkerGroup(b);
253         setChannelFactory(b);
254
255         p.connect();
256         return p;
257     }
258
259     private void setChannelFactory(final Bootstrap b) {
260         // There is no way to detect if this was already set by
261         // customizeBootstrap()
262         try {
263             b.channel(NioSocketChannel.class);
264         } catch (final IllegalStateException e) {
265             LOG.trace("Not overriding channelFactory on bootstrap {}", b, e);
266         }
267     }
268
269     /**
270      * @deprecated Should only be used with {@link AbstractDispatcher#AbstractDispatcher()}
271      */
272     @Deprecated
273     @Override
274     public void close() {
275         try {
276             this.workerGroup.shutdownGracefully();
277         } finally {
278             this.bossGroup.shutdownGracefully();
279         }
280     }
281 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.