BUG-7006: Unit-tests sometimes hangs during execution

[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
index f48173a8506d5a34878ed52e7c04edc7b0179258..d375bc0cdb69f5155dbfd44c42084a87c79290b3 100644 (file)
--- 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;
@@ -69,10 +70,10 @@ 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);
@@ -84,14 +85,15 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
         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,16 +114,17 @@ 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() {
         if (Epoll.isAvailable()) {
             this.workerGroup.shutdownGracefully().awaitUninterruptibly();
             this.bossGroup.shutdownGracefully().awaitUninterruptibly();
@@ -131,26 +134,32 @@ public class BGPDispatcherImpl implements BGPDispatcher, AutoCloseable {
     @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, peerRegistry, BGPChannel.createChannelPipelineInitializer(BGPDispatcherImpl.this.handlerFactory, snf));
+            retryTimer, bootstrap, peerRegistry, BGPChannel.createChannelPipelineInitializer(BGPDispatcherImpl.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 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 +170,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 +179,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 +200,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 +209,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) {