Bump versions by x.y.(z+1)

[bgpcep.git] / bgp / rib-impl / src / main / java / org / opendaylight / protocol / bgp / rib / impl / BGPDispatcherImpl.java

diff --git a/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BGPDispatcherImpl.java b/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BGPDispatcherImpl.java
old mode 100755 (executable)
new mode 100644 (file)
index 074c31b..5160d56
--- a/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BGPDispatcherImpl.java
+++ b/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BGPDispatcherImpl.java
@@ -7,6 +7,7 @@
  */
 package org.opendaylight.protocol.bgp.rib.impl;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import io.netty.bootstrap.Bootstrap;
@@ -31,6 +32,7 @@ import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GlobalEventExecutor;
 import io.netty.util.concurrent.Promise;
 import java.net.InetSocketAddress;
+import java.util.concurrent.TimeUnit;
 import org.opendaylight.protocol.bgp.parser.spi.MessageRegistry;
 import org.opendaylight.protocol.bgp.rib.impl.protocol.BGPProtocolSessionPromise;
 import org.opendaylight.protocol.bgp.rib.impl.protocol.BGPReconnectPromise;
@@ -51,6 +53,7 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
     private static final int SOCKET_BACKLOG_SIZE = 128;
     private static final int HIGH_WATER_MARK = 256 * 1024;
     private static final int LOW_WATER_MARK = 128 * 1024;
+    private static final long TIMEOUT = 10;
 
     private final BGPHandlerFactory handlerFactory;
     private final EventLoopGroup bossGroup;
@@ -69,29 +72,30 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
 
     @Override
     public synchronized Future<BGPSessionImpl> createClient(final InetSocketAddress remoteAddress, final BGPPeerRegistry listener, final int retryTimer) {
-        return createClient(remoteAddress, listener, retryTimer, createClientBootStrap(Optional.<KeyMapping>absent(), this.workerGroup));
+        return createClient(remoteAddress, listener, retryTimer, createClientBootStrap(Optional.absent(), false));
     }
 
-    private Future<BGPSessionImpl> createClient(final InetSocketAddress remoteAddress, final BGPPeerRegistry listener, final int retryTimer,
+    private synchronized Future<BGPSessionImpl> createClient(final InetSocketAddress remoteAddress, final BGPPeerRegistry listener, final int retryTimer,
             final Bootstrap clientBootStrap) {
         final BGPClientSessionNegotiatorFactory snf = new BGPClientSessionNegotiatorFactory(listener);
-        final ChannelPipelineInitializer initializer = BGPChannel.createChannelPipelineInitializer(BGPDispatcherImpl.this.handlerFactory, snf);
+        final ChannelPipelineInitializer initializer = BGPChannel.createChannelPipelineInitializer(this.handlerFactory, snf);
 
-        final BGPProtocolSessionPromise sessionPromise = new BGPProtocolSessionPromise(remoteAddress, retryTimer, clientBootStrap);
+        final BGPProtocolSessionPromise sessionPromise = new BGPProtocolSessionPromise(remoteAddress, retryTimer, clientBootStrap, listener);
         clientBootStrap.handler(BGPChannel.createClientChannelHandler(initializer, sessionPromise));
         sessionPromise.connect();
         LOG.debug("Client created.");
         return sessionPromise;
     }
 
-    public Future<BGPSessionImpl> createClient(final InetSocketAddress localAddress, final InetSocketAddress remoteAddress,
-            final BGPPeerRegistry strictBGPPeerRegistry, final int retryTimer) {
-        final Bootstrap clientBootStrap = createClientBootStrap(Optional.<KeyMapping>absent(), this.workerGroup);
+    @VisibleForTesting
+    public synchronized Future<BGPSessionImpl> createClient(final InetSocketAddress localAddress, final InetSocketAddress remoteAddress,
+            final BGPPeerRegistry strictBGPPeerRegistry, final int retryTimer, final boolean reuseAddress) {
+        final Bootstrap clientBootStrap = createClientBootStrap(Optional.absent(), reuseAddress);
         clientBootStrap.localAddress(localAddress);
         return createClient(remoteAddress, strictBGPPeerRegistry, retryTimer, clientBootStrap);
     }
 
-    protected Bootstrap createClientBootStrap(final Optional<KeyMapping> keys, final EventLoopGroup workerGroup) {
+    private synchronized Bootstrap createClientBootStrap(final Optional<KeyMapping> keys, final boolean reuseAddress) {
         final Bootstrap bootstrap = new Bootstrap();
         if (Epoll.isAvailable()) {
             bootstrap.channel(EpollSocketChannel.class);
@@ -112,45 +116,53 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
         bootstrap.option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
         bootstrap.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, HIGH_WATER_MARK);
         bootstrap.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, LOW_WATER_MARK);
+        bootstrap.option(ChannelOption.SO_REUSEADDR, reuseAddress);
 
         if (bootstrap.group() == null) {
-            bootstrap.group(workerGroup);
+            bootstrap.group(this.workerGroup);
         }
 
         return bootstrap;
     }
 
     @Override
-    public void close() {
+    public synchronized void close() throws InterruptedException {
         if (Epoll.isAvailable()) {
-            this.workerGroup.shutdownGracefully().awaitUninterruptibly();
-            this.bossGroup.shutdownGracefully().awaitUninterruptibly();
+            LOG.debug("Closing Dispatcher");
+            this.workerGroup.shutdownGracefully(0, TIMEOUT, TimeUnit.SECONDS);
+            this.bossGroup.shutdownGracefully(0, TIMEOUT, TimeUnit.SECONDS);
         }
     }
 
     @Override
     public synchronized Future<Void> createReconnectingClient(final InetSocketAddress remoteAddress, final BGPPeerRegistry peerRegistry,
             final int retryTimer, final Optional<KeyMapping> keys) {
+        return createReconnectingClient(remoteAddress, peerRegistry, retryTimer, keys, null, false);
+    }
+
+    @VisibleForTesting
+    protected synchronized Future<Void> createReconnectingClient(final InetSocketAddress remoteAddress, final BGPPeerRegistry peerRegistry,
+        final int retryTimer, final Optional<KeyMapping> keys, final InetSocketAddress localAddress, final boolean reuseAddress) {
         final BGPClientSessionNegotiatorFactory snf = new BGPClientSessionNegotiatorFactory(peerRegistry);
-        final Bootstrap bootstrap = createClientBootStrap(keys, this.workerGroup);
+        final Bootstrap bootstrap = createClientBootStrap(keys, reuseAddress);
+        bootstrap.localAddress(localAddress);
         final BGPReconnectPromise reconnectPromise = new BGPReconnectPromise(GlobalEventExecutor.INSTANCE, remoteAddress,
-                retryTimer, bootstrap, BGPChannel.createChannelPipelineInitializer(BGPDispatcherImpl.this.handlerFactory, snf));
+            retryTimer, bootstrap, peerRegistry, BGPChannel.createChannelPipelineInitializer(this.handlerFactory, snf));
         reconnectPromise.connect();
         return reconnectPromise;
     }
 
     @Override
-    public ChannelFuture createServer(final BGPPeerRegistry registry, final InetSocketAddress localAddress) {
+    public synchronized ChannelFuture createServer(final BGPPeerRegistry registry, final InetSocketAddress serverAddress) {
         final BGPServerSessionNegotiatorFactory snf = new BGPServerSessionNegotiatorFactory(registry);
-        final ChannelPipelineInitializer initializer = BGPChannel.createChannelPipelineInitializer(BGPDispatcherImpl.this.handlerFactory, snf);
-        final ServerBootstrap serverBootstrap = createServerBootstrap(initializer, this.bossGroup, this.workerGroup);
-        final ChannelFuture channelFuture = serverBootstrap.bind(localAddress);
-        LOG.debug("Initiated server {} at {}.", channelFuture, localAddress);
+        final ChannelPipelineInitializer initializer = BGPChannel.createChannelPipelineInitializer(this.handlerFactory, snf);
+        final ServerBootstrap serverBootstrap = createServerBootstrap(initializer);
+        final ChannelFuture channelFuture = serverBootstrap.bind(serverAddress);
+        LOG.debug("Initiated server {} at {}.", channelFuture, serverAddress);
         return channelFuture;
     }
 
-    public static ServerBootstrap createServerBootstrap(final ChannelPipelineInitializer initializer,
-            final EventLoopGroup bossGroup, final EventLoopGroup workerGroup) {
+    private synchronized ServerBootstrap createServerBootstrap(final ChannelPipelineInitializer initializer) {
         final ServerBootstrap serverBootstrap = new ServerBootstrap();
         if (Epoll.isAvailable()) {
             serverBootstrap.channel(EpollServerSocketChannel.class);
@@ -161,7 +173,7 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
         final ChannelHandler serverChannelHandler = BGPChannel.createServerChannelHandler(initializer);
         serverBootstrap.childHandler(serverChannelHandler);
 
-        serverBootstrap.option(ChannelOption.SO_BACKLOG, Integer.valueOf(SOCKET_BACKLOG_SIZE));
+        serverBootstrap.option(ChannelOption.SO_BACKLOG, SOCKET_BACKLOG_SIZE);
         serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
         serverBootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, HIGH_WATER_MARK);
         serverBootstrap.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, LOW_WATER_MARK);
@@ -170,19 +182,19 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
         serverBootstrap.option(ChannelOption.MAX_MESSAGES_PER_READ, 1);
 
         if (serverBootstrap.group() == null) {
-            serverBootstrap.group(bossGroup, workerGroup);
+            serverBootstrap.group(this.bossGroup, this.workerGroup);
         }
         return serverBootstrap;
     }
 
-    public static final class BGPChannel {
+    private static final class BGPChannel {
         private static final String NEGOTIATOR = "negotiator";
 
         private BGPChannel() {
 
         }
 
-        public static <S extends BGPSession, T extends BGPSessionNegotiatorFactory> ChannelPipelineInitializer
+        static <T extends BGPSessionNegotiatorFactory> ChannelPipelineInitializer
         createChannelPipelineInitializer(final BGPHandlerFactory hf, final T snf) {
             return (channel, promise) -> {
                 channel.pipeline().addLast(hf.getDecoders());
@@ -191,7 +203,7 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
             };
         }
 
-        public static <S extends BGPSession> ChannelHandler createClientChannelHandler(final ChannelPipelineInitializer initializer, final Promise<S> promise) {
+        static <S extends BGPSession> ChannelHandler createClientChannelHandler(final ChannelPipelineInitializer initializer, final Promise<S> promise) {
             return new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(final SocketChannel channel) {
@@ -200,7 +212,7 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
             };
         }
 
-        public static ChannelHandler createServerChannelHandler(final ChannelPipelineInitializer initializer) {
+        static ChannelHandler createServerChannelHandler(final ChannelPipelineInitializer initializer) {
             return new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(final SocketChannel channel) {