improve Sender builder consistency

Author: jerrinot

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

@@ -737,6 +737,9 @@ public LineSenderBuilder asyncMode(boolean enabled) {
          * @return this instance for method chaining
          */
         public LineSenderBuilder autoFlushBytes(int bytes) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("auto flush bytes is only supported for WebSocket transport");
+            }
             if (this.autoFlushBytes != PARAMETER_NOT_SET_EXPLICITLY) {
                 throw new LineSenderException("auto flush bytes was already configured")
                         .put("[bytes=").put(this.autoFlushBytes).put("]");
@@ -1258,6 +1261,9 @@ public LineSenderBuilder maxBackoffMillis(int maxBackoffMillis) {
          * @return this instance for method chaining
          */
         public LineSenderBuilder maxBufferCapacity(int maximumBufferCapacity) {
+            if (protocol == PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("maximum buffer capacity is not supported for WebSocket transport");
+            }
             if (maximumBufferCapacity < DEFAULT_BUFFER_CAPACITY) {
                 throw new LineSenderException("maximum buffer capacity cannot be less than initial buffer capacity ")
                         .put("[maximumBufferCapacity=").put(maximumBufferCapacity)
@@ -1749,13 +1755,14 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     }
                     autoFlushIntervalMillis(autoFlushInterval);
                 } else if (Chars.equals("auto_flush_bytes", sink)) {
-                    if (protocol != PROTOCOL_TCP) {
-                        throw new LineSenderException("auto_flush_bytes is only supported for TCP transport");
+                    if (protocol != PROTOCOL_TCP && protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("auto_flush_bytes is only supported for TCP and WebSocket transport");
                     }
                     pos = getValue(configurationString, pos, sink, "auto_flush_bytes");
-                    if (Chars.equalsIgnoreCase("off", sink)) {
-                        throw new LineSenderException("TCP transport must have auto_flush_bytes enabled");
-                    } else {
+                    if (protocol == PROTOCOL_TCP) {
+                        if (Chars.equalsIgnoreCase("off", sink)) {
+                            throw new LineSenderException("TCP transport must have auto_flush_bytes enabled");
+                        }
                         int autoFlushBytes = parseIntValue(sink, "auto_flush_bytes");
                         if (initBufSizeSet) {
                             if (autoFlushBytes != bufferCapacity) {
@@ -1764,6 +1771,9 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                         } else {
                             bufferCapacity(autoFlushBytes);
                         }
+                    } else {
+                        int autoFlushBytes = parseIntValue(sink, "auto_flush_bytes");
+                        autoFlushBytes(autoFlushBytes);
                     }
                     autoFlushBytesSet = true;
                 } else if (Chars.equals("auto_flush", sink)) {

core/src/test/java/io/questdb/client/test/cutlass/line/LineSenderBuilderTest.java

@@ -176,11 +176,11 @@ public void testConfStringValidation() throws Exception {
             assertConfStrError("TCPS::addr=localhost;", "invalid schema [schema=TCPS, supported-schemas=[http, https, tcp, tcps, ws, wss, udp]]");
             assertConfStrError("http::addr=localhost;auto_flush=off;auto_flush_interval=1;", "cannot set auto flush interval when interval based auto-flush is already disabled");
             assertConfStrError("http::addr=localhost;auto_flush=off;auto_flush_rows=1;", "cannot set auto flush rows when auto-flush is already disabled");
-            assertConfStrError("http::addr=localhost;auto_flush_bytes=1024;", "auto_flush_bytes is only supported for TCP transport");
+            assertConfStrError("http::addr=localhost;auto_flush_bytes=1024;", "auto_flush_bytes is only supported for TCP and WebSocket transport");
             assertConfStrError("http::addr=localhost;protocol_version=10", "current client only supports protocol version 1(text format for all datatypes), 2(binary format for part datatypes), 3(decimal datatype) or explicitly unset");
             assertConfStrError("http::addr=localhost:48884;max_name_len=10;", "max_name_len must be at least 16 bytes [max_name_len=10]");
-            assertConfStrError("ws::addr=localhost;token=foo;", "token is not supported for WebSocket protocol");
-            assertConfStrError("wss::addr=localhost;token=foo;", "token is not supported for WebSocket protocol");
+            assertConfStrError("ws::addr=localhost;max_buf_size=1000000;", "maximum buffer capacity is not supported for WebSocket transport");
+            assertConfStrError("wss::addr=localhost;tls_verify=unsafe_off;max_buf_size=1000000;", "maximum buffer capacity is not supported for WebSocket transport");
 
             assertConfStrOk("addr=localhost:8080", "auto_flush_rows=100", "protocol_version=1");
             assertConfStrOk("addr=localhost:8080", "auto_flush=on", "auto_flush_rows=100", "protocol_version=2");

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

@@ -89,10 +89,10 @@ public void testUdp_authNotSupported() {
     @Test
     public void testUdp_autoFlushBytesNotSupported() {
         assertThrowsAny(
-                Sender.builder(Sender.Transport.UDP)
+                () -> Sender.builder(Sender.Transport.UDP)
                         .address("localhost")
                         .autoFlushBytes(1000),
-                "not supported for UDP");
+                "auto flush bytes is only supported for WebSocket transport");
     }
 
     @Test

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

@@ -120,6 +120,14 @@ public void testAutoFlushBytesZero() {
         Assert.assertNotNull(builder);
     }
 
+    @Test
+    public void testAutoFlushBytesNotSupportedForHttp_fails() {
+        assertThrows("only supported for WebSocket transport",
+                () -> Sender.builder(Sender.Transport.HTTP)
+                        .address("localhost")
+                        .autoFlushBytes(1024));
+    }
+
     @Test
     public void testAutoFlushIntervalMillis() {
         Sender.LineSenderBuilder builder = Sender.builder(Sender.Transport.WEBSOCKET)
@@ -211,6 +219,14 @@ public void testBufferCapacityNegative_fails() {
                         .bufferCapacity(-1));
     }
 
+    @Test
+    public void testMaxBufferCapacityNotSupported_fails() {
+        assertThrows("maximum buffer capacity is not supported for WebSocket transport",
+                () -> Sender.builder(Sender.Transport.WEBSOCKET)
+                        .address(LOCALHOST)
+                        .maxBufferCapacity(128 * 1024));
+    }
+
     @Test
     public void testBuilderWithWebSocketTransport() {
         Sender.LineSenderBuilder builder = Sender.builder(Sender.Transport.WEBSOCKET);
@@ -644,6 +660,27 @@ public void testWsConfigString_withToken() throws Exception {
         });
     }
 
+    @Test
+    public void testWsConfigString_withAutoFlushBytes() {
+        Sender.LineSenderBuilder builder = Sender.builder("ws::addr=localhost:9000;auto_flush_bytes=1024;");
+        Assert.assertNotNull(builder);
+    }
+
+    @Test
+    public void testWsConfigString_withAutoFlushBytesDoubleSet_fails() {
+        assertBadConfig("ws::addr=localhost:9000;auto_flush_bytes=1024;auto_flush_bytes=2048;", "already configured");
+    }
+
+    @Test
+    public void testWsConfigString_withAutoFlushBytesInvalid_fails() {
+        assertBadConfig("ws::addr=localhost:9000;auto_flush_bytes=-1;", "cannot be negative");
+    }
+
+    @Test
+    public void testWsConfigString_withMaxBufSize_fails() {
+        assertBadConfig("ws::addr=localhost:9000;max_buf_size=1000000;", "maximum buffer capacity is not supported for WebSocket transport");
+    }
+
     @Test
     public void testWsConfigString_withUsernamePassword() throws Exception {
         assertMemoryLeak(() -> {
@@ -665,6 +702,17 @@ public void testWssConfigString_withToken() {
         Assert.assertNotNull(builder);
     }
 
+    @Test
+    public void testWssConfigString_withAutoFlushBytes() {
+        Sender.LineSenderBuilder builder = Sender.builder("wss::addr=localhost:9000;tls_verify=unsafe_off;auto_flush_bytes=1024;");
+        Assert.assertNotNull(builder);
+    }
+
+    @Test
+    public void testWssConfigString_withMaxBufSize_fails() {
+        assertBadConfig("wss::addr=localhost:9000;tls_verify=unsafe_off;max_buf_size=1000000;", "maximum buffer capacity is not supported for WebSocket transport");
+    }
+
     @Test
     public void testWssConfigString_uppercaseNotSupported() {
         assertBadConfig("WSS::addr=localhost:9000;", "invalid schema");