[plc4x] 01/01: PLC4X-139 close the worker thread on connection abortion to avoid thread and socket leak.

[jf...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jfeinauer pushed a commit to branch PLC4X-139-fix-socket-leak
in repository https://gitbox.apache.org/repos/asf/plc4x.git

commit 6f0826b1d2d7c6b969d4acd3d96e5342ea60ba8e
Author: Julian Feinauer <j....@pragmaticminds.de>
AuthorDate: Sun Aug 4 20:20:25 2019 +0200

    PLC4X-139 close the worker thread on connection abortion to avoid thread and socket leak.
---
 .../base/connection/TcpSocketChannelFactory.java   | 25 +++++++++++++++++++---
 1 file changed, 22 insertions(+), 3 deletions(-)

diff --git a/plc4j/protocols/driver-bases/tcp/src/main/java/org/apache/plc4x/java/base/connection/TcpSocketChannelFactory.java b/plc4j/protocols/driver-bases/tcp/src/main/java/org/apache/plc4x/java/base/connection/TcpSocketChannelFactory.java
index 8f3f101..ca1d5fb 100644
--- a/plc4j/protocols/driver-bases/tcp/src/main/java/org/apache/plc4x/java/base/connection/TcpSocketChannelFactory.java
+++ b/plc4j/protocols/driver-bases/tcp/src/main/java/org/apache/plc4x/java/base/connection/TcpSocketChannelFactory.java
@@ -25,8 +25,12 @@ import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
 import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
 import org.apache.plc4x.java.api.exceptions.PlcException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -34,6 +38,8 @@ import java.net.Socket;
 
 public class TcpSocketChannelFactory implements ChannelFactory {
 
+    private static final Logger logger = LoggerFactory.getLogger(TcpSocketChannelFactory.class);
+
     private static final int PING_TIMEOUT_MS = 1_000;
 
     private final InetAddress address;
@@ -49,14 +55,27 @@ public class TcpSocketChannelFactory implements ChannelFactory {
         throws PlcConnectionException {
         try {
             Bootstrap bootstrap = new Bootstrap();
-            bootstrap.group(new NioEventLoopGroup());
+            final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
+            bootstrap.group(workerGroup);
             bootstrap.channel(NioSocketChannel.class);
             bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
             bootstrap.option(ChannelOption.TCP_NODELAY, true);
+            // TODO we should use an explicit (configurable?) timeout here
+            // bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000);
             bootstrap.handler(channelHandler);
             // Start the client.
-            ChannelFuture f = bootstrap.connect(address, port).sync();
-            f.awaitUninterruptibly();
+            final ChannelFuture f = bootstrap.connect(address, port);
+            f.addListener(new GenericFutureListener<Future<? super Void>>() {
+                @Override public void operationComplete(Future<? super Void> future) throws Exception {
+                    if (!future.isSuccess()) {
+                        logger.info("Unable to connect, shutting down worker thread.");
+                        workerGroup.shutdownGracefully();
+                    }
+                }
+            });
+            // Wait for sync
+            f.sync();
+            f.awaitUninterruptibly(); // jf: unsure if we need that
             // Wait till the session is finished initializing.
             return f.channel();
         } catch (InterruptedException e) {