X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fcommons%2Fprotocol-framework%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fprotocol%2Fframework%2FAbstractDispatcher.java;h=a05d02cd09ad4e1b447a61da687768bc58e3dd94;hp=5e55cddf8e766509336041264dcecd7af06a2932;hb=1d364bf55905f619173d65a4823c6d594c5f29fe;hpb=44bac35e828c5dfee8a9fb7fad4f57bceaa9490b diff --git a/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/AbstractDispatcher.java b/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/AbstractDispatcher.java index 5e55cddf8e..a05d02cd09 100644 --- a/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/AbstractDispatcher.java +++ b/opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/AbstractDispatcher.java @@ -7,12 +7,25 @@ */ package org.opendaylight.protocol.framework; +import java.io.Closeable; +import java.net.InetSocketAddress; +import java.net.SocketAddress; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; + import io.netty.bootstrap.Bootstrap; import io.netty.bootstrap.ServerBootstrap; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; +import io.netty.channel.ServerChannel; +import io.netty.channel.local.LocalServerChannel; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; @@ -22,21 +35,14 @@ import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GlobalEventExecutor; import io.netty.util.concurrent.Promise; -import java.io.Closeable; -import java.net.InetSocketAddress; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.common.base.Preconditions; - /** * Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the * start method that will handle sockets in different thread. */ public abstract class AbstractDispatcher, L extends SessionListener> implements Closeable { - protected interface PipelineInitializer> { + + protected interface ChannelPipelineInitializer> { /** * Initializes channel by specifying the handlers in its pipeline. Handlers are protocol specific, therefore this * method needs to be implemented in protocol specific Dispatchers. @@ -44,7 +50,11 @@ public abstract class AbstractDispatcher, L extends * @param channel whose pipeline should be defined, also to be passed to {@link SessionNegotiatorFactory} * @param promise to be passed to {@link SessionNegotiatorFactory} */ - void initializeChannel(SocketChannel channel, Promise promise); + void initializeChannel(CH channel, Promise promise); + } + + protected interface PipelineInitializer> extends ChannelPipelineInitializer { + } @@ -76,21 +86,47 @@ public abstract class AbstractDispatcher, L extends * @return ChannelFuture representing the binding process */ protected ChannelFuture createServer(final InetSocketAddress address, final PipelineInitializer initializer) { + return createServer(address, NioServerSocketChannel.class, initializer); + } + + /** + * Creates server. Each server needs factories to pass their instances to client sessions. + * + * @param address address to which the server should be bound + * @param channelClass The {@link Class} which is used to create {@link Channel} instances from. + * @param initializer instance of PipelineInitializer used to initialize the channel pipeline + * + * @return ChannelFuture representing the binding process + */ + protected ChannelFuture createServer(final SocketAddress address, final Class channelClass, + final ChannelPipelineInitializer initializer) { final ServerBootstrap b = new ServerBootstrap(); - b.group(this.bossGroup, this.workerGroup); - b.channel(NioServerSocketChannel.class); - b.option(ChannelOption.SO_BACKLOG, 128); - b.childHandler(new ChannelInitializer() { + b.childHandler(new ChannelInitializer() { @Override - protected void initChannel(final SocketChannel ch) { + protected void initChannel(final CH ch) { initializer.initializeChannel(ch, new DefaultPromise(executor)); } }); - b.childOption(ChannelOption.SO_KEEPALIVE, true); + b.option(ChannelOption.SO_BACKLOG, 128); + if (LocalServerChannel.class.equals(channelClass) == false) { + // makes no sense for LocalServer and produces warning + b.childOption(ChannelOption.SO_KEEPALIVE, true); + } + b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); customizeBootstrap(b); + if (b.group() == null) { + b.group(bossGroup, workerGroup); + } + try { + b.channel(channelClass); + } catch (IllegalStateException e) { + // FIXME: if this is ok, document why + LOG.trace("Not overriding channelFactory on bootstrap {}", b, e); + } + // Bind and start to accept incoming connections. final ChannelFuture f = b.bind(address); LOG.debug("Initiated server {} at {}.", f, address); @@ -119,10 +155,9 @@ public abstract class AbstractDispatcher, L extends */ protected Future createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final PipelineInitializer initializer) { final Bootstrap b = new Bootstrap(); - final ProtocolSessionPromise p = new ProtocolSessionPromise(executor, address, strategy, b); - b.group(this.workerGroup).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).handler( + final ProtocolSessionPromise p = new ProtocolSessionPromise<>(executor, address, strategy, b); + b.option(ChannelOption.SO_KEEPALIVE, true).handler( new ChannelInitializer() { - @Override protected void initChannel(final SocketChannel ch) { initializer.initializeChannel(ch, p); @@ -130,6 +165,36 @@ public abstract class AbstractDispatcher, L extends }); customizeBootstrap(b); + setWorkerGroup(b); + setChannelFactory(b); + + p.connect(); + LOG.debug("Client created."); + return p; + } + + private void setWorkerGroup(final Bootstrap b) { + if (b.group() == null) { + b.group(workerGroup); + } + } + + /** + * Create a client but use a pre-configured bootstrap. + * This method however replaces the ChannelInitializer in the bootstrap. All other configuration is preserved. + * + * @param address remote address + */ + protected Future createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final Bootstrap bootstrap, final PipelineInitializer initializer) { + final ProtocolSessionPromise p = new ProtocolSessionPromise<>(executor, address, strategy, bootstrap); + + bootstrap.handler( + new ChannelInitializer() { + @Override + protected void initChannel(final SocketChannel ch) { + initializer.initializeChannel(ch, p); + } + }); p.connect(); LOG.debug("Client created."); @@ -148,6 +213,9 @@ public abstract class AbstractDispatcher, L extends } /** + * + * @deprecated use {@link org.opendaylight.protocol.framework.AbstractDispatcher#createReconnectingClient(java.net.InetSocketAddress, ReconnectStrategyFactory, org.opendaylight.protocol.framework.AbstractDispatcher.PipelineInitializer)} with only one reconnectStrategyFactory instead. + * * Creates a client. * * @param address remote address @@ -157,15 +225,47 @@ public abstract class AbstractDispatcher, L extends * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g. * success if it indicates no further attempts should be made and failure if it reports an error */ + @Deprecated protected Future createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory, final ReconnectStrategy reestablishStrategy, final PipelineInitializer initializer) { + return createReconnectingClient(address, connectStrategyFactory, initializer); + } - final ReconnectPromise p = new ReconnectPromise(GlobalEventExecutor.INSTANCE, this, address, connectStrategyFactory, reestablishStrategy, initializer); - p.connect(); + /** + * Creates a reconnecting client. + * + * @param address remote address + * @param connectStrategyFactory Factory for creating reconnection strategy for every reconnect attempt + * + * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g. + * success if it indicates no further attempts should be made and failure if it reports an error + */ + protected Future createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory, + final PipelineInitializer initializer) { + final Bootstrap b = new Bootstrap(); + final ReconnectPromise p = new ReconnectPromise<>(GlobalEventExecutor.INSTANCE, this, address, connectStrategyFactory, b, initializer); + + b.option(ChannelOption.SO_KEEPALIVE, true); + + customizeBootstrap(b); + setWorkerGroup(b); + setChannelFactory(b); + + p.connect(); return p; } + private void setChannelFactory(final Bootstrap b) { + // There is no way to detect if this was already set by + // customizeBootstrap() + try { + b.channel(NioSocketChannel.class); + } catch (final IllegalStateException e) { + LOG.trace("Not overriding channelFactory on bootstrap {}", b, e); + } + } + /** * @deprecated Should only be used with {@link AbstractDispatcher#AbstractDispatcher()} */ @@ -178,5 +278,4 @@ public abstract class AbstractDispatcher, L extends this.bossGroup.shutdownGracefully(); } } - }