Merge "Add XSQL into default compilation"
[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 com.google.common.base.Preconditions;
11
12 import io.netty.bootstrap.Bootstrap;
13 import io.netty.bootstrap.ServerBootstrap;
14 import io.netty.channel.Channel;
15 import io.netty.buffer.PooledByteBufAllocator;
16 import io.netty.channel.ChannelFuture;
17 import io.netty.channel.ChannelInitializer;
18 import io.netty.channel.ChannelOption;
19 import io.netty.channel.EventLoopGroup;
20 import io.netty.channel.ServerChannel;
21 import io.netty.channel.local.LocalServerChannel;
22 import io.netty.channel.socket.SocketChannel;
23 import io.netty.channel.socket.nio.NioServerSocketChannel;
24 import io.netty.channel.socket.nio.NioSocketChannel;
25 import io.netty.util.concurrent.DefaultPromise;
26 import io.netty.util.concurrent.EventExecutor;
27 import io.netty.util.concurrent.Future;
28 import io.netty.util.concurrent.GlobalEventExecutor;
29 import io.netty.util.concurrent.Promise;
30
31 import java.io.Closeable;
32 import java.net.InetSocketAddress;
33 import java.net.SocketAddress;
34
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
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<S>(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
169         if (b.group() == null) {
170             b.group(workerGroup);
171         }
172
173         // There is no way to detect if this was already set by
174         // customizeBootstrap()
175         try {
176             b.channel(NioSocketChannel.class);
177         } catch (IllegalStateException e) {
178             LOG.trace("Not overriding channelFactory on bootstrap {}", b, e);
179         }
180
181         p.connect();
182         LOG.debug("Client created.");
183         return p;
184     }
185
186     /**
187      * Customize a client bootstrap before the connection is attempted. This
188      * allows subclasses to assign non-default options before the client is
189      * created.
190      *
191      * @param b Client bootstrap
192      */
193     protected void customizeBootstrap(final Bootstrap b) {
194         // The default is a no-op
195     }
196
197     /**
198      * Creates a client.
199      *
200      * @param address remote address
201      * @param connectStrategyFactory Factory for creating reconnection strategy to be used when initial connection fails
202      * @param reestablishStrategy Reconnection strategy to be used when the already-established session fails
203      *
204      * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
205      *         success if it indicates no further attempts should be made and failure if it reports an error
206      */
207     protected Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
208             final ReconnectStrategy reestablishStrategy, final PipelineInitializer<S> initializer) {
209
210         final ReconnectPromise<S, L> p = new ReconnectPromise<S, L>(GlobalEventExecutor.INSTANCE, this, address, connectStrategyFactory, reestablishStrategy, initializer);
211         p.connect();
212
213         return p;
214     }
215
216     /**
217      * @deprecated Should only be used with {@link AbstractDispatcher#AbstractDispatcher()}
218      */
219     @Deprecated
220     @Override
221     public void close() {
222         try {
223             this.workerGroup.shutdownGracefully();
224         } finally {
225             this.bossGroup.shutdownGracefully();
226         }
227     }
228
229 }

©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.