Add more logs

Author: Artem Labazin

encon/src/main/java/io/appulse/encon/ModuleClient.java

@@ -18,6 +18,7 @@
 
 import static io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS;
 import static io.netty.channel.ChannelOption.SO_KEEPALIVE;
+import static io.netty.channel.ChannelOption.TCP_NODELAY;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static lombok.AccessLevel.PRIVATE;
 
@@ -88,12 +89,14 @@ private CompletableFuture<Connection> createConnection (@NonNull RemoteNode remo
         .group(moduleConnection.getWorkerGroup())
         .channel(moduleConnection.getClientChannelClass())
         .option(SO_KEEPALIVE, true)
+        .option(TCP_NODELAY, true)
         .option(CONNECT_TIMEOUT_MILLIS, 5000)
         .handler(HandshakeClientInitializer.builder()
             .node(node)
             .future(future)
             .remote(remote)
             .channelCloseAction(remoteNode -> {
+              log.debug("Closing connection to {}", remoteNode);
               node.moduleLookup.remove(remoteNode);
               node.moduleConnection.remove(remoteNode);
             })

encon/src/main/java/io/appulse/encon/ModuleConnection.java

@@ -77,11 +77,11 @@ class ModuleConnection implements Closeable {
 
   Map<RemoteNode, CompletableFuture<Connection>> cache;
 
-  ModuleConnection (int bossThreads, int workerThreads) {
+  ModuleConnection (@NonNull String prefix, int bossThreads, int workerThreads) {
     cache = new ConcurrentHashMap<>();
 
-    val bossThreadFactory = new DefaultThreadFactory("nbg");
-    val workerThreadFactory = new DefaultThreadFactory("nwg");
+    val bossThreadFactory = new DefaultThreadFactory(prefix + "-nbg");
+    val workerThreadFactory = new DefaultThreadFactory(prefix + "-nwg");
 
     if (Epoll.isAvailable()) {
       bossGroup = new EpollEventLoopGroup(bossThreads, bossThreadFactory);
@@ -120,10 +120,14 @@ void add (@NonNull CompletableFuture<Connection> future) {
 
   CompletableFuture<Connection> compute (@NonNull RemoteNode remote,
                                          @NonNull Function<RemoteNode, CompletableFuture<Connection>> function) {
+    if (log.isDebugEnabled()) {
+      log.debug("Remote node {} exists: {}", remote, cache.containsKey(remote));
+    }
     return cache.computeIfAbsent(remote, function);
   }
 
   void remove (@NonNull RemoteNode remote) {
+    log.debug("Removing connection to {}", remote);
     CompletableFuture<Connection> future = cache.remove(remote);
     if (future == null) {
       return;

encon/src/main/java/io/appulse/encon/ModulePing.java

@@ -90,7 +90,7 @@ boolean ping (@NonNull RemoteNode remote) {
 
     try {
       mailbox.receive();
-      mailbox.getNode().remove(mailbox);
+      mailbox.close();
       log.debug("Returning from ping method");
       return true;
     } catch (Exception ex) {

encon/src/main/java/io/appulse/encon/ModuleServer.java

@@ -80,6 +80,7 @@ private void start () {
             .node(node)
             .consumer(moduleConnection::add)
             .channelCloseAction(remote -> {
+              log.debug("Closing connection to {}", remote);
               node.moduleLookup.remove(remote);
               node.moduleConnection.remove(remote);
             })

encon/src/main/java/io/appulse/encon/Node.java

@@ -161,6 +161,7 @@ private Node (@NonNull NodeDescriptor descriptor,
     modulePing = new ModulePing(this);
     moduleLookup = new ModuleLookup(epmd);
     moduleConnection = new ModuleConnection(
+        descriptor.getShortName(),
         config.getServer().getBossThreads(),
         config.getServer().getWorkerThreads()
     );

encon/src/main/java/io/appulse/encon/connection/handshake/AbstractHandshakeChannelInitializer.java

@@ -82,7 +82,7 @@ protected void initChannel (SocketChannel socketChannel) throws Exception {
   protected void initChannel (SocketChannel socketChannel, AbstractHandshakeHandler handler) {
     socketChannel.pipeline()
         .addLast("LOGGING", LOGGING_HANDLER)
-        .addLast("READ_TIMEOUT", new ReadTimeoutHandler(10))
+        .addLast("READ_TIMEOUT", new ReadTimeoutHandler(5))
         .addLast("LENGTH_PREPENDER", LENGTH_FIELD_PREPENDER)
         .addLast("LENGTH_DECODER", new LengthFieldBasedFrameDecoder(MAX_VALUE, 0, 2))
         .addLast("DECODER", decoder)