- * Pass {@code 0} or negative to disable keepalive PINGs entirely - * — the caller then accepts that durable-ack frames may not - * arrive on idle connections (e.g. they are sending data - * continuously, or have their own ping driver). + * Pass {@code 0} or negative to disable it. The idle PING + * ({@link #keepaliveIntervalMillis(long)}) is separate. *
* Default * {@value io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS} @@ -2599,6 +2599,27 @@ public LineSenderBuilder initialConnectRetry(boolean enabled) { return this; } + /** + * Cadence (in millis) of the keepalive PING the SF cursor I/O loop + * sends once nothing has gone out on the connection for this long. + * It stops load balancers, proxies, firewalls and NAT gateways (an + * AWS ALB closes idle connections after 60 s by default), and the + * server's own idle timeout, from closing a quiet connection, which + * would otherwise cost a reconnect each time. + *
+ * Pass {@code 0} or negative to disable it. Default
+ * {@value io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop#DEFAULT_KEEPALIVE_INTERVAL_MILLIS}
+ * ms. Background orphan drainers ({@link #drainOrphans(boolean)})
+ * always use the default. WebSocket transport only.
+ */
+ public LineSenderBuilder keepaliveIntervalMillis(long millis) {
+ if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+ throw new LineSenderException("keepalive_interval_millis is only supported for WebSocket transport");
+ }
+ this.keepaliveIntervalMillis = millis;
+ return this;
+ }
+
/**
* Cap on concurrent background drainer threads when
* {@link #drainOrphans(boolean)} is on. Default {@code 4}. Each
@@ -4157,6 +4178,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
}
pos = getValue(configurationString, pos, sink, "durable_ack_keepalive_interval_millis");
durableAckKeepaliveIntervalMillis(parseLongValue(sink, "durable_ack_keepalive_interval_millis"));
+ } else if (Chars.equals("keepalive_interval_millis", sink)) {
+ if (protocol != PROTOCOL_WEBSOCKET) {
+ throw new LineSenderException("keepalive_interval_millis is only supported for WebSocket transport");
+ }
+ pos = getValue(configurationString, pos, sink, "keepalive_interval_millis");
+ keepaliveIntervalMillis(parseLongValue(sink, "keepalive_interval_millis"));
} else if (Chars.equals("reconnect_max_duration_millis", sink)) {
if (protocol != PROTOCOL_WEBSOCKET) {
throw new LineSenderException("reconnect_max_duration_millis is only supported for WebSocket transport");
@@ -4468,6 +4495,9 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
if (view.has("durable_ack_keepalive_interval_millis")) {
durableAckKeepaliveIntervalMillis(wsLong(view, v, "durable_ack_keepalive_interval_millis"));
}
+ if (view.has("keepalive_interval_millis")) {
+ keepaliveIntervalMillis(wsLong(view, v, "keepalive_interval_millis"));
+ }
if (view.has("reconnect_max_duration_millis")) {
reconnectMaxDurationMillis(wsLong(view, v, "reconnect_max_duration_millis"));
}
@@ -4677,6 +4707,7 @@ public java.util.Map
* Recycling requires a non-null engine rebuild factory installed via
* {@link #setEngineRebuildFactory(EngineRebuildFactory)};
@@ -1219,7 +1223,8 @@ public static QwpWebSocketSender connectWithCredentialSupplier(
catchUpCapGapMinEscalationWindowMillis,
DEFAULT_SYMBOL_DICT_RESET_ENABLED,
DEFAULT_SYMBOL_DICT_RESET_THRESHOLD_SYMBOLS,
- DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS);
+ DEFAULT_SYMBOL_DICT_RESET_MAX_WAIT_MILLIS,
+ CursorWebSocketSendLoop.DEFAULT_KEEPALIVE_INTERVAL_MILLIS);
}
/**
@@ -1228,7 +1233,8 @@ public static QwpWebSocketSender connectWithCredentialSupplier(
* recycle knobs ({@code symbol_dict_reset}, {@code symbol_dict_reset_threshold},
* {@code symbol_dict_reset_max_wait_millis}): consecutive server-active
* rejections of the same head-of-line frame, with no ack progress in
- * between, before the loop escalates to a typed terminal.
+ * between, before the loop escalates to a typed terminal. Also takes the
+ * idle keepalive PING cadence ({@code keepalive_interval_millis}).
*
* Named apart from {@code connect} for the reason given on
* {@link #connectWithCredentialSupplier(List, ClientTlsConfiguration, int, int, long, Supplier,
@@ -1272,7 +1278,8 @@ public static QwpWebSocketSender connectWithCredentialSupplier(
long catchUpCapGapMinEscalationWindowMillis,
boolean symbolDictResetEnabled,
int symbolDictResetThresholdSymbols,
- long symbolDictResetMaxWaitMillis
+ long symbolDictResetMaxWaitMillis,
+ long keepaliveIntervalMillis
) {
if (symbolDictResetThresholdSymbols <= 0
|| symbolDictResetThresholdSymbols > QwpConstants.MAX_SYMBOL_DICTIONARY_SIZE / 2) {
@@ -1301,6 +1308,7 @@ public static QwpWebSocketSender connectWithCredentialSupplier(
sender.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
sender.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
sender.durableAckKeepaliveIntervalMillis = durableAckKeepaliveIntervalMillis;
+ sender.keepaliveIntervalMillis = keepaliveIntervalMillis;
sender.maxFrameRejections = maxFrameRejections;
sender.poisonMinEscalationWindowMillis = poisonMinEscalationWindowMillis;
sender.catchUpCapGapMinEscalationWindowMillis = catchUpCapGapMinEscalationWindowMillis;
@@ -4896,6 +4904,7 @@ private void ensureConnected() {
errorDispatcher = new SenderErrorDispatcher(errorHandler, errorInboxCapacity);
}
cursorSendLoop.setErrorDispatcher(errorDispatcher);
+ cursorSendLoop.setKeepaliveIntervalMillis(keepaliveIntervalMillis);
// Symmetric progress dispatcher: lazy-allocated mirror of the
// error path. Wired before start() for the same reason -- the
// I/O thread should never observe a null dispatcher between
diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
index 469d2a0d..c58f918f 100644
--- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
+++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java
@@ -129,14 +129,25 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
/**
* Default cadence for the keepalive PING the I/O loop emits while
* waiting on STATUS_DURABLE_ACK frames. See
- * {@link #sendDurableAckKeepaliveIfDue()} for the rationale: the OSS
+ * {@link #sendPingIfDue(long)} for the rationale: the OSS
* server only flushes pending durable-ack frames on inbound recv
* events, so an opted-in idle client has to prod it. {@code 200} ms
* trades one PING per 200 ms per idle opted-in connection for
* sub-second confirmation latency once the upload completes
- * server-side. {@code 0} or negative disables the keepalive entirely.
+ * server-side. {@code 0} or negative disables this PING, not the idle keepalive.
*/
public static final long DEFAULT_DURABLE_ACK_KEEPALIVE_INTERVAL_MILLIS = 200L;
+ /**
+ * Default cadence for the idle keepalive PING, sent once nothing has gone
+ * out on the wire for this long. Without it a quiet connection is closed
+ * by whatever sits on the path with an idle timeout -- L7 load balancers
+ * and proxies (AWS ALB: 60 s), firewalls, NAT gateways -- and by the
+ * server's own idle reaper ({@code http.net.connection.timeout}, 5 min),
+ * and the loop pays a reconnect each time. TCP keepalive does not help
+ * through an L7 proxy, which terminates TCP. {@code 0} or negative
+ * disables it.
+ */
+ public static final long DEFAULT_KEEPALIVE_INTERVAL_MILLIS = 30_000L;
public static final long DEFAULT_PARK_NANOS = 50_000L; // 50us idle backoff
/**
* Default initial reconnect backoff (100 ms).
@@ -271,9 +282,11 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// visible to the loop under test.
@TestOnly
public static volatile boolean forceMirrorSeedFailureForTest;
- // Pre-converted to nanos for the comparison in sendDurableAckKeepaliveIfDue.
- // Zero or negative disables the keepalive entirely.
+ // Pre-converted to nanos for the comparison in sendPingIfDue.
+ // Zero or negative disables the durable-ack PING only.
private final long durableAckKeepaliveIntervalNanos;
+ // Idle keepalive PING cadence in nanos; 0 or negative disables. Set before start().
+ private long keepaliveIntervalNanos = TimeUnit.MILLISECONDS.toNanos(DEFAULT_KEEPALIVE_INTERVAL_MILLIS);
// When true, OK frames do NOT advance engine.acknowledge -- only
// STATUS_DURABLE_ACK frames do. The OK frame's wireSeq is stashed in
// pendingDurable along with its per-table seqTxns, and trim only advances
@@ -510,10 +523,10 @@ public final class CursorWebSocketSendLoop implements QuietCloseable {
// rethrows always deliver the same instance.
private volatile LineSenderException terminalError;
// Wall clock of the last outbound activity on the wire -- a sent frame
- // (trySendOne) or a keepalive PING (sendDurableAckKeepaliveIfDue).
- // Throttles the durable-ack keepalive PING so it fires only when the
+ // (trySendOne) or a keepalive PING (sendPingIfDue).
+ // Throttles the keepalive PINGs so they fire only when the
// configured interval has elapsed since the most recent outbound event.
- // Zero until the first send; reset to zero on reconnect.
+ // Set when the I/O loop starts.
private long lastFrameOrPingNanos;
// Throttles the catch-up oversize WARN. Only consulted when the server
// advertises NO batch cap, where the loop has no terminal to fall back on
@@ -663,7 +676,7 @@ public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
/**
* Master constructor — also accepts the cadence at which the I/O loop
* sends keepalive PINGs while waiting on STATUS_DURABLE_ACK frames.
- * Pass {@code 0} or negative to disable keepalive PINGs entirely (the
+ * Pass {@code 0} or negative to disable this durable-ack PING (the
* caller takes responsibility for prodding the server, e.g. by sending
* data, or by accepting indefinite waits on idle connections).
*/
@@ -969,7 +982,7 @@ private CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
* The policy-aware constructor as released in 1.3.x: equivalent to the overload
* below with {@code externalFsnBase = 0L}, the value every caller outside a
* symbol-dictionary recycle passes. Kept so callers compiled against a released
- * jar keep linking; {@code ExportedApiCompatibilityTest} pins it.
+ * jar keep linking.
*/
public CursorWebSocketSendLoop(WebSocketClient client, CursorSendEngine engine,
long fsnAtZero, long parkNanos,
@@ -1716,6 +1729,14 @@ public void setErrorDispatcher(SenderErrorDispatcher dispatcher) {
this.errorDispatcher = dispatcher;
}
+ /**
+ * Idle keepalive PING cadence, see {@link #DEFAULT_KEEPALIVE_INTERVAL_MILLIS}.
+ * {@code 0} or negative disables it. Set once before {@link #start()}.
+ */
+ public void setKeepaliveIntervalMillis(long millis) {
+ this.keepaliveIntervalNanos = TimeUnit.MILLISECONDS.toNanos(millis);
+ }
+
/**
* Test seam: resize the {@link #close()} graceful-stop window (default
* {@link #DEFAULT_CLOSE_GRACEFUL_STOP_MILLIS}). {@code 0} disables it, so
@@ -1896,10 +1917,6 @@ private void clearDurableAckTracking() {
releasePendingEntry(pendingDurable.pollFirst());
}
durableTableWatermarks.clear();
- // Reset the keepalive throttle so the new connection can prod the
- // server immediately rather than waiting out the leftover interval
- // from before the reconnect.
- lastFrameOrPingNanos = 0L;
}
private static boolean isCatchUpCapGap(Throwable t) {
@@ -2033,12 +2050,14 @@ private void connectWalk(Throwable initial, String phase, long paceFirstAttemptM
}
break;
}
- swapClient(newClient);
+ long resumeFsn = swapClient(newClient);
counters.reconnects.incrementAndGet();
long elapsedMs = (System.nanoTime() - outageStartNanos) / 1_000_000L;
+ // swapClient cleared replayTargetFsn when nothing below it is unacked
+ long replayFrames = replayTargetFsn >= 0L ? replayTargetFsn - resumeFsn + 1L : 0L;
LOG.info("cursor I/O loop {} succeeded after {}ms, {} attempts; "
- + "replaying from FSN {}",
- phase, elapsedMs, attempts, externalFsnBase + fsnAtZero);
+ + "resuming at FSN {}, {} unacked frames to replay",
+ phase, elapsedMs, attempts, externalFsnBase + resumeFsn, replayFrames);
return;
}
// A null factory result is an unsuccessful connect state, not a cap-gap
@@ -2601,6 +2620,9 @@ private void ioLoop() {
if (client == null && running) {
attemptInitialConnect();
}
+ // The upgrade handshake was outbound traffic: start the idle
+ // keepalive clock here rather than pinging on the first idle pass.
+ lastFrameOrPingNanos = System.nanoTime();
while (running) {
boolean didWork = trySendOne();
// 1. Try to send next frame(s).
@@ -2618,7 +2640,11 @@ private void ioLoop() {
if (!didWork && running && durableAckMode
&& durableAckKeepaliveIntervalNanos > 0L
&& !pendingDurable.isEmpty()) {
- sendDurableAckKeepaliveIfDue();
+ sendPingIfDue(durableAckKeepaliveIntervalNanos);
+ }
+ // 4. Idle keepalive PING, see DEFAULT_KEEPALIVE_INTERVAL_MILLIS.
+ if (!didWork && running && keepaliveIntervalNanos > 0L) {
+ sendPingIfDue(keepaliveIntervalNanos);
}
if (!didWork && running) {
LockSupport.parkNanos(parkNanos);
@@ -2822,27 +2848,29 @@ private void releaseSendingSegment() {
}
/**
- * Send a WebSocket PING to prod the server into flushing pending
- * STATUS_DURABLE_ACK frames, but only when the throttle interval has
- * elapsed since the last outbound activity -- a sent frame or a prior
- * keepalive PING. The server's egress code only runs flushPendingAck on
- * inbound recv events; without this prod, an idle connection waiting on
- * durable-ack confirmation can sit forever. The "last sent frame" half
- * of the gate avoids a redundant PING shortly after a producer batch
- * goes idle: the recent frame already triggered a server-side flush.
+ * Send a WebSocket PING, but only when {@code intervalNanos} has elapsed
+ * since the last outbound activity -- a sent frame or a prior PING.
+ * Serves two callers: the idle keepalive, and the durable-ack keepalive
+ * that prods the server into flushing pending STATUS_DURABLE_ACK frames
+ * (its egress code only runs flushPendingAck on inbound recv events, so
+ * an idle connection waiting on durable-ack confirmation can sit
+ * forever). The "last sent frame" half of the gate avoids a redundant
+ * PING shortly after a producer batch goes idle.
*
* Best-effort: any send failure routes through the standard fail() path
* so the reconnect loop can take over. Caller is responsible for the
- * "do we even need to send" gate (durableAckMode + non-empty pending).
+ * "do we even need to send" gate.
*/
- private void sendDurableAckKeepaliveIfDue() {
+ private void sendPingIfDue(long intervalNanos) {
long now = System.nanoTime();
- if (now - lastFrameOrPingNanos < durableAckKeepaliveIntervalNanos) {
+ if (now - lastFrameOrPingNanos < intervalNanos) {
return;
}
lastFrameOrPingNanos = now;
try {
- client.sendPing(1000);
+ // A timeout here reconnects and replays unacked frames, so it must be as
+ // tolerant as a data send: a slow-draining server or a GC pause is not a dead peer.
+ client.sendPing();
} catch (Throwable t) {
fail(t);
}
@@ -2870,8 +2898,10 @@ private void snapshotReplayTarget() {
* realign {@code fsnAtZero} to the next unacked FSN, restart wire
* sequencing from 0, and reposition the cursor so the next
* {@link #trySendOne} call replays the first unacked frame.
+ *
+ * @return the first unacked FSN, where sending resumes on the new connection
*/
- private void swapClient(WebSocketClient newClient) {
+ private long swapClient(WebSocketClient newClient) {
WebSocketClient old = this.client;
this.client = newClient;
this.lastReconnectError = null;
@@ -2915,6 +2945,7 @@ private void swapClient(WebSocketClient newClient) {
// established, so the sender stays INITIALIZING and a subsequent
// endpoint-policy rejection still latches the startup terminal.
this.hasEverConnected = true;
+ return replayStart;
}
/**
diff --git a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java
index b9725608..bc2d0a6a 100644
--- a/core/src/main/java/io/questdb/client/impl/ConfigSchema.java
+++ b/core/src/main/java/io/questdb/client/impl/ConfigSchema.java
@@ -71,6 +71,7 @@ public final class ConfigSchema {
str("durable_ack_keepalive_interval_millis", Side.INGRESS);
str("error_inbox_capacity", Side.INGRESS);
str("initial_connect_retry", Side.INGRESS);
+ str("keepalive_interval_millis", Side.INGRESS);
str("max_background_drainers", Side.INGRESS);
// Poison-frame detector threshold (see CursorWebSocketSendLoop): the
// Sender applies it; the QwpQueryClient accepts it as a syntactic no-op
diff --git a/core/src/test/java/io/questdb/client/test/compat/ExportedApiCompatibilityTest.java b/core/src/test/java/io/questdb/client/test/compat/ExportedApiCompatibilityTest.java
deleted file mode 100644
index 48e7e864..00000000
--- a/core/src/test/java/io/questdb/client/test/compat/ExportedApiCompatibilityTest.java
+++ /dev/null
@@ -1,270 +0,0 @@
-/*+*****************************************************************************
- * ___ _ ____ ____
- * / _ \ _ _ ___ ___| |_| _ \| __ )
- * | | | | | | |/ _ \/ __| __| | | | _ \
- * | |_| | |_| | __/\__ \ |_| |_| | |_) |
- * \__\_\\__,_|\___||___/\__|____/|____/
- *
- * Copyright (c) 2014-2019 Appsicle
- * Copyright (c) 2019-2026 QuestDB
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- ******************************************************************************/
-
-package io.questdb.client.test.compat;
-
-import io.questdb.client.cutlass.http.client.Fragment;
-import io.questdb.client.cutlass.http.client.Response;
-import io.questdb.client.cutlass.line.http.AbstractLineHttpSender;
-import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
-import io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.lang.reflect.Constructor;
-import java.lang.reflect.Method;
-import java.lang.reflect.Modifier;
-import java.util.LinkedHashSet;
-import java.util.Set;
-import java.util.TreeSet;
-
-/**
- * Pins the public signatures this branch had to restore after replacing them in place.
- *
- * Four exported methods were changed rather than added to - {@code Response.recv(int)} arrived as an
- * abstract interface method, two {@code QwpWebSocketSender.connect(..., String, ...)} overloads were retyped
- * to {@code Supplier
- * There is no japicmp or revapi gate on this build, so this test is the gate. The expected signatures below
- * are the ones present at this branch's merge base ({@code 981bdb02}); they are written out literally rather
- * than derived from the current classes, because a pin computed from the thing it pins proves nothing.
- * Adding an overload is fine and this test stays green; retyping or removing one turns it red.
- *
- * Not every pin guards a signature that shipped. The {@code connect}, {@code createLineSender} and
- * {@code Response.recv} pins, and the seven {@link #RELEASED_CONSTRUCTOR_SIGNATURES} constructor pins, all
- * guard signatures present in a published jar - 1.3.9 has every one of them. The two
- * {@code connectWithCredentialSupplier} pins are different: that signature is on {@code main} since the OIDC
- * device-flow change ({@code 0b9b5766}), after 1.3.9, and has not shipped in a release, so the pin protects
- * a caller building against {@code main}, not against a published jar.
- *
- * {@link #RELEASED_CONSTRUCTOR_SIGNATURES} pins every public {@code CursorWebSocketSendLoop} constructor
- * present in the published {@code org.questdb:questdb-client:1.3.9} jar, copied from {@code javap} of that
- * jar for the same reason: the policy-aware constructor was widened in place on this branch and the
- * released 13-argument form had to be restored as an overload.
- */
-public class ExportedApiCompatibilityTest {
-
- /**
- * Every {@code QwpWebSocketSender.connect}, {@code QwpWebSocketSender.connectWithCredentialSupplier} and
- * {@code AbstractLineHttpSender.createLineSender} signature that existed at the merge base, as
- * {@code name(paramType,...)returnType} over erased type names.
- */
- private static final String[] PRE_BRANCH_SIGNATURES = {
- // ---- QwpWebSocketSender.connect ----
- "connect(java.lang.String,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.lang.String,int,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connect(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.lang.String,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int,int,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- // ---- QwpWebSocketSender.connectWithCredentialSupplier ----
- "connectWithCredentialSupplier(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.util.function.Supplier,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- "connectWithCredentialSupplier(java.util.List,io.questdb.client.ClientTlsConfiguration,int,int,long,java.util.function.Supplier,boolean,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,long,long,io.questdb.client.Sender$InitialConnectMode,io.questdb.client.SenderErrorHandler,int,long,long,int,io.questdb.client.SenderConnectionListener,int,int,long,long)io.questdb.client.cutlass.qwp.client.QwpWebSocketSender",
- // ---- AbstractLineHttpSender.createLineSender ----
- "createLineSender(java.lang.String,int,java.lang.String,io.questdb.client.HttpClientConfiguration,io.questdb.client.ClientTlsConfiguration,int,java.lang.String,java.lang.String,java.lang.String,int,long,int,long,long,int)io.questdb.client.cutlass.line.http.AbstractLineHttpSender",
- "createLineSender(io.questdb.client.std.ObjList,io.questdb.client.std.IntList,java.lang.String,io.questdb.client.HttpClientConfiguration,io.questdb.client.ClientTlsConfiguration,int,java.lang.String,java.lang.String,java.lang.String,int,long,int,long,long,int)io.questdb.client.cutlass.line.http.AbstractLineHttpSender",
- };
-
- /**
- * Every public {@code CursorWebSocketSendLoop} constructor in the 1.3.9 jar, as {@code (paramType,...)}
- * over erased type names, in the order {@code javap} printed them.
- */
- private static final String[] RELEASED_CONSTRUCTOR_SIGNATURES = {
- "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long)",
- "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean)",
- "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long)",
- "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int)",
- "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int,long)",
- "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int,long,long)",
- "(io.questdb.client.cutlass.http.client.WebSocketClient,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorSendEngine,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectFactory,long,long,boolean,long,int,long,long,io.questdb.client.cutlass.qwp.client.sf.cursor.CursorWebSocketSendLoop$ReconnectPolicy)",
- };
-
- @Test
- public void testPreBranchCreateLineSenderOverloadsStillLink() {
- assertSignaturesPresent(AbstractLineHttpSender.class, "createLineSender");
- }
-
- @Test
- public void testPreBranchQwpWebSocketSenderConnectOverloadsStillLink() {
- assertSignaturesPresent(QwpWebSocketSender.class, "connect");
- }
-
- @Test
- public void testPreBranchQwpWebSocketSenderConnectWithCredentialSupplierOverloadsStillLink() {
- assertSignaturesPresent(QwpWebSocketSender.class, "connectWithCredentialSupplier");
- }
-
- @Test
- public void testReleasedCursorWebSocketSendLoopConstructorsStillLink() {
- assertConstructorsPresent(CursorWebSocketSendLoop.class, RELEASED_CONSTRUCTOR_SIGNATURES);
- }
-
- @Test
- public void testResponseRecvIntIsDefaultNotAbstract() throws Exception {
- // The defect was recv(int) arriving as an abstract interface method: an external implementation
- // written against the earlier Response compiles fine and then fails at run time with
- // AbstractMethodError, which is exactly the failure a unit test of this library never sees.
- Method recvWithTimeout = Response.class.getMethod("recv", int.class);
- Assert.assertFalse("Response.recv(int) must stay a default method - an implementation written "
- + "before the overload existed has no override for it",
- Modifier.isAbstract(recvWithTimeout.getModifiers()));
- Assert.assertEquals(Fragment.class, recvWithTimeout.getReturnType());
-
- Method recv = Response.class.getMethod("recv");
- Assert.assertTrue("recv() is the one method an implementation must supply",
- Modifier.isAbstract(recv.getModifiers()));
- Assert.assertEquals(Fragment.class, recv.getReturnType());
- }
-
- @Test
- public void testResponseImplementorOverridingOnlyRecvStillWorks() {
- // LegacyResponse below is the compile-time half: it implements Response and overrides recv() ONLY,
- // exactly as an implementation predating the overload does. If recv(int) went back to being
- // abstract, this class stops compiling and the whole test module goes with it - which is the point.
- LegacyResponse legacy = new LegacyResponse();
- Fragment first = legacy.recv();
- Assert.assertNotNull(first);
- Assert.assertEquals(1, legacy.calls);
-
- // and the default must keep the PREVIOUS behaviour, not merely link: it ignores the bound and
- // defers to recv(), which is what such an implementation did before the overload existed
- Fragment bounded = legacy.recv(5_000);
- Assert.assertSame("the default must delegate to recv()", legacy.fragment, bounded);
- Assert.assertEquals("and must not read anything of its own", 2, legacy.calls);
-
- Fragment unbounded = legacy.recv(0);
- Assert.assertSame("a non-positive timeout is the legacy unbounded path, same delegation",
- legacy.fragment, unbounded);
- Assert.assertEquals(3, legacy.calls);
- }
-
- private static void assertConstructorsPresent(Class type, String[] expectedSignatures) {
- final Set