Refactor WebSocketSender API to unify connection methods

Replaced `connectAsync` with an enhanced `connect` to simplify API usage and eliminate redundancy. Updated tests and benchmarks to align with the new unified connection method, ensuring compatibility and maintaining functionality. Adjusted default auto-flush and configuration parameters for better performance.

Author: jerrinot

core/src/main/java/io/questdb/client/Sender.java

@@ -885,7 +885,7 @@ public Sender build() {
 
                 String wsAuthHeader = buildWebSocketAuthHeader();
 
-                return QwpWebSocketSender.connectAsync(
+                return QwpWebSocketSender.connect(
                         hosts.getQuick(0),
                         ports.getQuick(0),
                         tlsEnabled,

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -66,8 +66,8 @@
  * <p>
  * Configuration options:
  * <ul>
- *   <li>{@code autoFlushRows} - Maximum rows per batch (default: 500)</li>
- *   <li>{@code autoFlushBytes} - Maximum bytes per batch (default: 1MB)</li>
+ *   <li>{@code autoFlushRows} - Maximum rows per batch (default: 1000)</li>
+ *   <li>{@code autoFlushBytes} - Maximum bytes per batch (default: 128KB)</li>
  *   <li>{@code autoFlushIntervalNanos} - Maximum age before auto-flush (default: 100ms)</li>
  * </ul>
  * <p>
@@ -189,7 +189,7 @@ private QwpWebSocketSender(
 
     /**
      * Creates a new sender and connects to the specified host and port.
-     * Uses synchronous mode for backward compatibility.
+     * Uses default auto-flush settings and in-flight window size.
      *
      * @param host server host
      * @param port server HTTP port (WebSocket upgrade happens on same port)
@@ -200,8 +200,8 @@ public static QwpWebSocketSender connect(String host, int port) {
     }
 
     /**
-     * Creates a new sender with TLS and connects to the specified host and port.
-     * Uses synchronous mode with default auto-flush settings.
+     * Creates a new sender and connects to the specified host and port.
+     * Uses default auto-flush settings and in-flight window size.
      *
      * @param host       server host
      * @param port       server HTTP port
@@ -210,107 +210,29 @@ public static QwpWebSocketSender connect(String host, int port) {
      */
     public static QwpWebSocketSender connect(String host, int port, boolean tlsEnabled) {
         return connect(
-                host, port, tlsEnabled, DEFAULT_AUTO_FLUSH_ROWS, DEFAULT_AUTO_FLUSH_BYTES, DEFAULT_AUTO_FLUSH_INTERVAL_NANOS
+                host, port, tlsEnabled,
+                DEFAULT_AUTO_FLUSH_ROWS, DEFAULT_AUTO_FLUSH_BYTES, DEFAULT_AUTO_FLUSH_INTERVAL_NANOS,
+                DEFAULT_IN_FLIGHT_WINDOW_SIZE, null
         );
     }
 
     /**
-     * Creates a new sender with TLS and connects to the specified host and port.
-     * Uses synchronous mode with custom auto-flush settings.
+     * Creates a new sender with full configuration and connects.
+     * <p>
+     * In-flight window size controls the flow behavior: 1 means synchronous (each batch
+     * waits for ACK), greater than 1 enables asynchronous pipelining with a background I/O thread.
      *
      * @param host                   server host
      * @param port                   server HTTP port
      * @param tlsEnabled             whether to use TLS
      * @param autoFlushRows          rows per batch (0 = no limit)
      * @param autoFlushBytes         bytes per batch (0 = no limit)
      * @param autoFlushIntervalNanos age before flush in nanos (0 = no limit)
+     * @param inFlightWindowSize     max batches awaiting server ACK (1 = sync, default: 128)
+     * @param authorizationHeader    HTTP Authorization header value, or null
      * @return connected sender
      */
     public static QwpWebSocketSender connect(
-            String host,
-            int port,
-            boolean tlsEnabled,
-            int autoFlushRows,
-            int autoFlushBytes,
-            long autoFlushIntervalNanos
-    ) {
-        return connect(host, port, tlsEnabled, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos, null);
-    }
-
-    public static QwpWebSocketSender connect(
-            String host,
-            int port,
-            boolean tlsEnabled,
-            int autoFlushRows,
-            int autoFlushBytes,
-            long autoFlushIntervalNanos,
-            String authorizationHeader
-    ) {
-        QwpWebSocketSender sender = new QwpWebSocketSender(
-                host, port, tlsEnabled, DEFAULT_BUFFER_SIZE, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos,
-                1,   // window=1 for sync behavior
-                authorizationHeader
-        );
-        try {
-            sender.ensureConnected();
-        } catch (Throwable t) {
-            sender.close();
-            throw t;
-        }
-        return sender;
-    }
-
-    /**
-     * Creates a new sender with async mode and custom configuration.
-     *
-     * @param host                   server host
-     * @param port                   server HTTP port
-     * @param tlsEnabled             whether to use TLS
-     * @param autoFlushRows          rows per batch (0 = no limit)
-     * @param autoFlushBytes         bytes per batch (0 = no limit)
-     * @param autoFlushIntervalNanos age before flush in nanos (0 = no limit)
-     * @return connected sender
-     */
-    public static QwpWebSocketSender connectAsync(
-            String host,
-            int port,
-            boolean tlsEnabled,
-            int autoFlushRows,
-            int autoFlushBytes,
-            long autoFlushIntervalNanos
-    ) {
-        return connectAsync(
-                host, port, tlsEnabled, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos, DEFAULT_IN_FLIGHT_WINDOW_SIZE
-        );
-    }
-
-    /**
-     * Creates a new sender with async mode and full configuration including flow control.
-     *
-     * @param host                   server host
-     * @param port                   server HTTP port
-     * @param tlsEnabled             whether to use TLS
-     * @param autoFlushRows          rows per batch (0 = no limit)
-     * @param autoFlushBytes         bytes per batch (0 = no limit)
-     * @param autoFlushIntervalNanos age before flush in nanos (0 = no limit)
-     * @param inFlightWindowSize     max batches awaiting server ACK (default: 8)
-     * @return connected sender
-     */
-    public static QwpWebSocketSender connectAsync(
-            String host,
-            int port,
-            boolean tlsEnabled,
-            int autoFlushRows,
-            int autoFlushBytes,
-            long autoFlushIntervalNanos,
-            int inFlightWindowSize
-    ) {
-        return connectAsync(
-                host, port, tlsEnabled, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos, inFlightWindowSize, null
-        );
-    }
-
-    public static QwpWebSocketSender connectAsync(
             String host,
             int port,
             boolean tlsEnabled,
@@ -321,7 +243,9 @@ public static QwpWebSocketSender connectAsync(
             String authorizationHeader
     ) {
         QwpWebSocketSender sender = new QwpWebSocketSender(
-                host, port, tlsEnabled, DEFAULT_BUFFER_SIZE, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos, inFlightWindowSize, authorizationHeader
+                host, port, tlsEnabled, DEFAULT_BUFFER_SIZE,
+                autoFlushRows, autoFlushBytes, autoFlushIntervalNanos,
+                inFlightWindowSize, authorizationHeader
         );
         try {
             sender.ensureConnected();
@@ -332,20 +256,6 @@ public static QwpWebSocketSender connectAsync(
         return sender;
     }
 
-    /**
-     * Creates a new sender with async mode and default configuration.
-     *
-     * @param host       server host
-     * @param port       server HTTP port
-     * @param tlsEnabled whether to use TLS
-     * @return connected sender
-     */
-    public static QwpWebSocketSender connectAsync(String host, int port, boolean tlsEnabled) {
-        return connectAsync(
-                host, port, tlsEnabled, DEFAULT_AUTO_FLUSH_ROWS, DEFAULT_AUTO_FLUSH_BYTES, DEFAULT_AUTO_FLUSH_INTERVAL_NANOS
-        );
-    }
-
     /**
      * Creates a sender without connecting. For testing only.
      * <p>
@@ -359,9 +269,10 @@ public static QwpWebSocketSender connectAsync(String host, int port, boolean tls
      */
     public static QwpWebSocketSender createForTesting(String host, int port, int inFlightWindowSize) {
         return new QwpWebSocketSender(
-                host, port, false, DEFAULT_BUFFER_SIZE, DEFAULT_AUTO_FLUSH_ROWS, DEFAULT_AUTO_FLUSH_BYTES, DEFAULT_AUTO_FLUSH_INTERVAL_NANOS, inFlightWindowSize, null
+                host, port, false, DEFAULT_BUFFER_SIZE,
+                DEFAULT_AUTO_FLUSH_ROWS, DEFAULT_AUTO_FLUSH_BYTES, DEFAULT_AUTO_FLUSH_INTERVAL_NANOS,
+                inFlightWindowSize, null
         );
-        // Note: does NOT call ensureConnected()
     }
 
     /**
@@ -384,9 +295,10 @@ public static QwpWebSocketSender createForTesting(
             int inFlightWindowSize
     ) {
         return new QwpWebSocketSender(
-                host, port, false, DEFAULT_BUFFER_SIZE, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos, inFlightWindowSize, null
+                host, port, false, DEFAULT_BUFFER_SIZE,
+                autoFlushRows, autoFlushBytes, autoFlushIntervalNanos,
+                inFlightWindowSize, null
         );
-        // Note: does NOT call ensureConnected()
     }
 
     @Override

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketAckIntegrationTest.java

@@ -55,8 +55,8 @@ public void testAsyncFlushFailsFastOnInvalidAckPayload() throws Exception {
 
             boolean errorCaught = false;
             long start = System.currentTimeMillis();
-            try (QwpWebSocketSender sender = QwpWebSocketSender.connectAsync(
-                    "localhost", port, false, 0, 0, 0)) {
+            try (QwpWebSocketSender sender = QwpWebSocketSender.connect(
+                    "localhost", port, false, 0, 0, 0, QwpWebSocketSender.DEFAULT_IN_FLIGHT_WINDOW_SIZE, null)) {
                 sender.table("test")
                         .longColumn("value", 1)
                         .atNow();
@@ -86,8 +86,8 @@ public void testAsyncFlushFailsFastOnServerClose() throws Exception {
 
             boolean errorCaught = false;
             long start = System.currentTimeMillis();
-            try (QwpWebSocketSender sender = QwpWebSocketSender.connectAsync(
-                    "localhost", port, false, 0, 0, 0)) {
+            try (QwpWebSocketSender sender = QwpWebSocketSender.connect(
+                    "localhost", port, false, 0, 0, 0, QwpWebSocketSender.DEFAULT_IN_FLIGHT_WINDOW_SIZE, null)) {
                 sender.table("test")
                         .longColumn("value", 1)
                         .atNow();
@@ -121,8 +121,8 @@ public void testFlushBlocksUntilAcked() throws Exception {
             server.start();
             Assert.assertTrue("Server failed to start", server.awaitStart(5, TimeUnit.SECONDS));
 
-            try (QwpWebSocketSender sender = QwpWebSocketSender.connectAsync(
-                    "localhost", port, false, 0, 0, 0)) {
+            try (QwpWebSocketSender sender = QwpWebSocketSender.connect(
+                    "localhost", port, false, 0, 0, 0, QwpWebSocketSender.DEFAULT_IN_FLIGHT_WINDOW_SIZE, null)) {
 
                 sender.table("test")
                         .longColumn("value", 42)