- Server connection handler fixes.

- Logging tweaks.
- Test case updates.
- Minor version bump.

Author: jjzazuet

build.gradle.kts

@@ -4,7 +4,7 @@ subprojects {
   apply(plugin = "io.vacco.oss.gitflow")
 
   group = "io.vacco.tokoeka"
-  version = "0.4.1"
+  version = "0.5.0"
 
   configure<io.vacco.oss.gitflow.GsPluginProfileExtension> {
     sharedLibrary(true, false)

tk-sdr/src/main/java/io/vacco/tokoeka/TkSocket.java

@@ -71,19 +71,15 @@ public void listen(Supplier<Boolean> go) {
         break;
       }
     }
-    this.close();
+    doClose(this);
   }
 
   @Override public void accept(String s) {
     send(s, this.outputStream);
   }
 
   @Override public void close() {
-    try {
-      socket.close();
-    } catch (IOException e) {
-      log.error("Unable to close ws socket: {} - {}", this, e.getMessage());
-    }
+    doClose(socket);
     if (log.isDebugEnabled()) {
       log.debug("ws connection closed");
     }

tk-sdr/src/main/java/io/vacco/tokoeka/TkSocketServer.java

@@ -54,7 +54,8 @@ private void handleClient(Socket clientSocket) {
           clientSocket, socketState,
           msg -> send(msg, outputStream),
           (code, msg) -> {
-
+            sendClose(outputStream, code, msg);
+            doClose(clientSocket);
           }
         );
         this.socketHdl.onOpen(conn, handshakeResponse);
@@ -72,22 +73,14 @@ private void handleClient(Socket clientSocket) {
         log.debug("Incoming connection handler error - {}", clientSocket.getRemoteSocketAddress(), e);
       }
     } finally {
-      try {
-        clientSocket.close();
-      } catch (IOException e) {
-        log.warn("Incoming connection close error - {}", e.getMessage());
-      }
+      doClose(clientSocket);
     }
   }
 
   @Override public void close() {
     clientThreadPool.shutdown();
     if (serverSocket != null) {
-      try {
-        serverSocket.close();
-      } catch (Exception e) {
-        log.warn("Websocket server close error - {}", e.getMessage());
-      }
+      doClose(serverSocket);
     }
   }
 

tk-sdr/src/main/java/io/vacco/tokoeka/audio/TkAudioPlayer.java

@@ -70,4 +70,5 @@ public void close() {
       line.close();
     }
   }
+
 }

tk-sdr/src/main/java/io/vacco/tokoeka/handler/TkControlHdl.java

@@ -9,7 +9,6 @@
 import java.net.URLDecoder;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.function.Consumer;
 
 import static io.vacco.tokoeka.schema.TkConstants.*;
 import static io.vacco.tokoeka.util.TkCommand.*;
@@ -68,19 +67,15 @@ private void processKeyValue(String key, String value) {
       case down:            this.controlEvent(-1, key, value, true, null); break;
       default:
         if (log.isDebugEnabled()) {
-          var val = value == null ? "" : value.trim();
-          if (val.length() > 64) {
-            val = String.format("%s...", val.substring(0, 64));
-          }
-          log.debug("Unknown message key/value: {} -> {}", key, val);
+          log.debug("Unknown message key/value: {} -> {}", key, shorten(value));
         }
     }
   }
 
   private void processMsg(String body) {
     var params = parseParameters(body);
     if (log.isTraceEnabled()) {
-      log.trace(">> {} {} {}", MSG, body, params);
+      log.trace(">> {} {} {}", MSG, shorten(body), shorten(params.toString()));
     }
     params.forEach(this::processKeyValue);
   }

tk-sdr/src/main/java/io/vacco/tokoeka/spi/TkConn.java

@@ -1,10 +1,9 @@
 package io.vacco.tokoeka.spi;
 
-import java.io.Closeable;
 import java.net.Socket;
 import java.util.function.Consumer;
 
-public interface TkConn extends Consumer<String>, Closeable {
+public interface TkConn extends Consumer<String> {
   void setAttachment(Object attachment);
   <T> T getAttachment();
   Socket getSocket();

tk-sdr/src/main/java/io/vacco/tokoeka/util/TkConnAdapter.java

@@ -1,8 +1,6 @@
 package io.vacco.tokoeka.util;
 
 import io.vacco.tokoeka.spi.TkConn;
-
-import java.io.IOException;
 import java.net.Socket;
 import java.util.Objects;
 import java.util.function.BiConsumer;
@@ -40,10 +38,6 @@ public TkConnAdapter(Socket socket, TkSocketState socketState,
     return socket;
   }
 
-  @Override public void close() throws IOException {
-    socket.close();
-  }
-
   @Override public void close(int code) {
     onClose.accept(code, null);
   }

tk-sdr/src/main/java/io/vacco/tokoeka/util/TkFormat.java

@@ -12,6 +12,8 @@
 
 public class TkFormat {
 
+  public static int DefaultMessageShorten = 64;
+
   public static String asString(ByteBuffer bytes) {
     return StandardCharsets.UTF_8.decode(bytes).toString();
   }
@@ -50,4 +52,22 @@ public static TkDxConfig loadKiwiDxConfig(String data, TkJsonIn jsonIn) {
     return load(data, TkDxConfig.class, jsonIn);
   }
 
+  public static String shorten(String msg, int maxLength) {
+    if (msg == null) {
+      return "null";
+    }
+    msg = msg.trim();
+    if (msg.length() <= maxLength || maxLength == 1) {
+      return msg;
+    }
+    var ml2 = maxLength / 2;
+    var v0 = msg.substring(0, ml2);
+    var v1 = msg.substring(msg.length() - ml2);
+    return String.format("%s...%s", v0, v1);
+  }
+
+  public static String shorten(String msg) {
+    return shorten(msg, DefaultMessageShorten);
+  }
+
 }

tk-sdr/src/main/java/io/vacco/tokoeka/util/TkSockets.java

@@ -11,6 +11,7 @@
 
 import static java.nio.ByteBuffer.wrap;
 import static java.lang.System.currentTimeMillis;
+import static io.vacco.tokoeka.util.TkFormat.shorten;
 
 public class TkSockets {
 
@@ -218,10 +219,7 @@ public static void send(String message, OutputStream outputStream) {
         log.trace("> TXT: {} ({} bytes)", message, payload.length);
       }
     } catch (Exception e) {
-      var msg = message != null && message.length() > 64
-        ? String.format("%s...", message.substring(0, 64))
-        : message;
-      throw new IllegalStateException(String.format("unable to send text: %s", msg), e);
+      throw new IllegalStateException(String.format("unable to send text: %s", shorten(message)), e);
     }
   }
 
@@ -264,7 +262,7 @@ public static boolean handleMessage(TkSocketHdl socketHdl, TkSocketState socketS
       var pongDiff = nowMs - socketState.lastPongMs;
       var pongExp = pongDiff != nowMs && pongDiff >= socketState.keepAliveMs; // missing subsequent client pong
       if (pongExp) {
-        log.debug("Ping/Pong keep-alive expired {}, pingDiff: {}, pongDiff: {}", conn.getSocket().getRemoteSocketAddress(), pingDiff, pongDiff);
+        log.warn("Ping/Pong keep-alive expired {}, pingDiff: {}, pongDiff: {}", conn.getSocket().getRemoteSocketAddress(), pingDiff, pongDiff);
         sendClose(outputStream, WsCloseGoAway, WsCloseGoAwayRes);
         socketHdl.onClose(conn, WsCloseGoAway, false);
         return true;
@@ -301,7 +299,7 @@ public static boolean handleMessage(TkSocketHdl socketHdl, TkSocketState socketS
       if (opcode == 0x1) {
         var msg = new String(completeMessage);
         if (log.isTraceEnabled()) {
-          log.trace("< TXT: {}", msg);
+          log.trace("< TXT: {}", shorten(msg));
         }
         socketHdl.onMessage(conn, msg);
       } else if (opcode == 0x2) {
@@ -333,4 +331,16 @@ public static boolean handleMessage(TkSocketHdl socketHdl, TkSocketState socketS
     return false;
   }
 
+  public static void doClose(Closeable c) {
+    try {
+      c.close();
+    } catch (Exception e) {
+      if (log.isWarnEnabled()) {
+        log.warn("Error closing {} - {}", c, e.getMessage());
+      } else if (log.isDebugEnabled()) {
+        log.debug("Error closing {}", c);
+      }
+    }
+  }
+
 }

tk-sdr/src/test/java/io/vacco/tokoeka/TkControlHdlTest.java

@@ -8,15 +8,14 @@
 import j8spec.junit.J8SpecRunner;
 import org.junit.runner.RunWith;
 import org.slf4j.*;
-
-import java.io.IOException;
 import java.io.InputStreamReader;
 import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.util.*;
 
 import static j8spec.J8Spec.*;
 import static io.vacco.tokoeka.TkLogging.initLog;
+import static io.vacco.tokoeka.util.TkFormat.shorten;
 
 @DefinedOrder
 @RunWith(J8SpecRunner.class)
@@ -44,8 +43,9 @@ public class TkControlHdlTest {
         @Override public void setAttachment(Object attachment) {}
         @Override public <T> T getAttachment() { return null; }
         @Override public Socket getSocket() { return null; }
-        @Override public void close() {}
-        @Override public void close(int code) {}
+        @Override public void close(int code) {
+          log.info("close - [{}]", code);
+        }
         @Override public void close(int code, String msg) {
           log.info("close - [{}, {}]", code, msg);
         }
@@ -62,7 +62,7 @@ public class TkControlHdlTest {
         .withConfigPin(((kiwiConfig, dxConfig, dxCommConfig) -> {
           for (var o : new Object[]{kiwiConfig, dxConfig, dxCommConfig}) {
             if (o != null) {
-              log.info(g.toJson(o));
+              log.info(shorten(g.toJson(o)));
             }
           }
         }));
@@ -72,7 +72,7 @@ public class TkControlHdlTest {
           var bytes = ByteBuffer.wrap(Base64.getDecoder().decode(msg.data));
           ctlHdl.onMessage(conn, bytes);
         } else {
-          log.info("Ref command ==> {}", msg.data);
+          log.info("Ref command ==> {}", shorten(msg.data));
         }
       }
       for (var msg : wfMessages) {

tk-sdr/src/test/java/io/vacco/tokoeka/TkPairTest.java

@@ -1,5 +1,6 @@
 package io.vacco.tokoeka;
 
+import io.vacco.tokoeka.util.TkFormat;
 import j8spec.annotation.DefinedOrder;
 import j8spec.junit.J8SpecRunner;
 import org.junit.runner.RunWith;
@@ -24,5 +25,19 @@ public class TkPairTest {
       );
       assertEquals("SET single hello=world boolFlag1=1 boolFlag0=0 intArg=999 dblArg=1.235", cmd);
     });
+    it("Formats redacted messages", () -> {
+      assertEquals("null", TkFormat.shorten(null, 0));
+      assertEquals("1", TkFormat.shorten("1", 2));
+      assertEquals("1", TkFormat.shorten("1", 1));
+      assertEquals("1", TkFormat.shorten("1", 2));
+      assertEquals("1...1", TkFormat.shorten("111", 2));
+      assertEquals("1...1", TkFormat.shorten("1111", 2));
+      assertEquals("1...1", TkFormat.shorten("11111", 2));
+      assertEquals("111", TkFormat.shorten("111", 3));
+      assertEquals("1...1", TkFormat.shorten("1111", 3));
+      assertEquals("1...1", TkFormat.shorten("11111", 3));
+      assertEquals("01...34", TkFormat.shorten("01234", 4));
+      assertEquals("0123...CDEF", TkFormat.shorten("0123456789ABCDEF", 8));
+    });
   }
 }