Skip to content

Commit 7012fad

Browse files
glasstigerglasstiger
committed
fixing tests, add more tests
1 parent 4bc8c84 commit 7012fad

2 files changed

Lines changed: 243 additions & 0 deletions

File tree

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/QwpWebSocketSenderStateTest.java‎

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,17 +24,27 @@
2424

2525
package io.questdb.client.test.cutlass.qwp.client;
2626

27+
import io.questdb.client.DefaultHttpClientConfiguration;
28+
import io.questdb.client.cutlass.http.client.WebSocketClient;
29+
import io.questdb.client.cutlass.http.client.WebSocketFrameHandler;
2730
import io.questdb.client.cutlass.line.LineSenderException;
2831
import io.questdb.client.cutlass.qwp.client.InFlightWindow;
2932
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
33+
import io.questdb.client.cutlass.qwp.client.WebSocketResponse;
3034
import io.questdb.client.cutlass.qwp.protocol.QwpTableBuffer;
35+
import io.questdb.client.network.PlainSocketFactory;
36+
import io.questdb.client.std.MemoryTag;
37+
import io.questdb.client.std.Unsafe;
3138
import io.questdb.client.test.AbstractTest;
3239
import org.junit.Assert;
3340
import org.junit.Test;
3441

3542
import java.lang.reflect.Field;
3643
import java.lang.reflect.Method;
3744
import java.time.temporal.ChronoUnit;
45+
import java.util.ArrayList;
46+
import java.util.List;
47+
import java.util.function.Consumer;
3848

3949
import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
4050

@@ -50,6 +60,26 @@
5060
*/
5161
public class QwpWebSocketSenderStateTest extends AbstractTest {
5262

63+
@Test
64+
public void testConnectWithDurableAckToClosedPort() throws Exception {
65+
assertMemoryLeak(() -> {
66+
try {
67+
QwpWebSocketSender.connect(
68+
"127.0.0.1", 1, null,
69+
QwpWebSocketSender.DEFAULT_AUTO_FLUSH_ROWS,
70+
QwpWebSocketSender.DEFAULT_AUTO_FLUSH_BYTES,
71+
QwpWebSocketSender.DEFAULT_AUTO_FLUSH_INTERVAL_NANOS,
72+
1, null,
73+
QwpWebSocketSender.DEFAULT_MAX_SCHEMAS_PER_CONNECTION,
74+
true
75+
).close();
76+
Assert.fail("Expected LineSenderException");
77+
} catch (LineSenderException e) {
78+
Assert.assertTrue(e.getMessage().contains("Failed to connect"));
79+
}
80+
});
81+
}
82+
5383
@Test
5484
public void testGetHighestDurableSequenceDefaultsToMinusOne() throws Exception {
5585
assertMemoryLeak(() -> {
@@ -112,6 +142,101 @@ public void testSetRequestDurableAckOnClosedSenderThrows() throws Exception {
112142
});
113143
}
114144

145+
@Test
146+
public void testPingAfterCloseThrows() throws Exception {
147+
assertMemoryLeak(() -> {
148+
QwpWebSocketSender sender = QwpWebSocketSender.createForTesting("localhost", 0, 1);
149+
sender.close();
150+
try {
151+
sender.ping();
152+
Assert.fail("Expected exception");
153+
} catch (LineSenderException e) {
154+
Assert.assertTrue(e.getMessage().contains("closed"));
155+
}
156+
});
157+
}
158+
159+
@Test
160+
public void testSyncPingProcessesDurableAck() throws Exception {
161+
assertMemoryLeak(() -> {
162+
QwpWebSocketSender sender = QwpWebSocketSender.createForTesting("localhost", 0, 1);
163+
PingTestClient client = new PingTestClient();
164+
try {
165+
client.frameSequence.add(handler -> emitBinaryResponse(handler, WebSocketResponse.durableAck(5)));
166+
client.frameSequence.add(handler -> handler.onPong(0, 0));
167+
168+
setField(sender, "client", client);
169+
setField(sender, "connected", true);
170+
setField(sender, "inFlightWindow", new InFlightWindow(1, InFlightWindow.DEFAULT_TIMEOUT_MS));
171+
172+
sender.ping();
173+
174+
Assert.assertTrue(client.pingSent);
175+
Assert.assertEquals(5L, sender.getHighestDurableSequence());
176+
} finally {
177+
setField(sender, "client", null);
178+
setField(sender, "connected", false);
179+
sender.close();
180+
client.close();
181+
}
182+
});
183+
}
184+
185+
@Test
186+
public void testSyncPingProcessesStatusOk() throws Exception {
187+
assertMemoryLeak(() -> {
188+
QwpWebSocketSender sender = QwpWebSocketSender.createForTesting("localhost", 0, 1);
189+
PingTestClient client = new PingTestClient();
190+
try {
191+
client.frameSequence.add(handler -> emitBinaryResponse(handler, WebSocketResponse.success(3)));
192+
client.frameSequence.add(handler -> handler.onPong(0, 0));
193+
194+
setField(sender, "client", client);
195+
setField(sender, "connected", true);
196+
InFlightWindow window = new InFlightWindow(8, InFlightWindow.DEFAULT_TIMEOUT_MS);
197+
window.addInFlight(0);
198+
window.addInFlight(1);
199+
window.addInFlight(2);
200+
window.addInFlight(3);
201+
setField(sender, "inFlightWindow", window);
202+
203+
sender.ping();
204+
205+
Assert.assertTrue(client.pingSent);
206+
Assert.assertEquals(3L, sender.getHighestAckedSequence());
207+
} finally {
208+
setField(sender, "client", null);
209+
setField(sender, "connected", false);
210+
sender.close();
211+
client.close();
212+
}
213+
});
214+
}
215+
216+
@Test
217+
public void testSyncPingReturnsOnPong() throws Exception {
218+
assertMemoryLeak(() -> {
219+
QwpWebSocketSender sender = QwpWebSocketSender.createForTesting("localhost", 0, 1);
220+
PingTestClient client = new PingTestClient();
221+
try {
222+
client.frameSequence.add(handler -> handler.onPong(0, 0));
223+
224+
setField(sender, "client", client);
225+
setField(sender, "connected", true);
226+
setField(sender, "inFlightWindow", new InFlightWindow(1, InFlightWindow.DEFAULT_TIMEOUT_MS));
227+
228+
sender.ping();
229+
230+
Assert.assertTrue(client.pingSent);
231+
} finally {
232+
setField(sender, "client", null);
233+
setField(sender, "connected", false);
234+
sender.close();
235+
client.close();
236+
}
237+
});
238+
}
239+
115240
@Test
116241
public void testAutoFlushAccumulatesRowsAcrossAllTables() throws Exception {
117242
assertMemoryLeak(() -> {
@@ -387,4 +512,56 @@ private static void setField(Object target, String fieldName, Object value) thro
387512
f.setAccessible(true);
388513
f.set(target, value);
389514
}
515+
516+
private static void emitBinaryResponse(WebSocketFrameHandler handler, WebSocketResponse response) {
517+
int size = response.serializedSize();
518+
long ptr = Unsafe.malloc(size, MemoryTag.NATIVE_DEFAULT);
519+
try {
520+
response.writeTo(ptr);
521+
handler.onBinaryMessage(ptr, size);
522+
} finally {
523+
Unsafe.free(ptr, size, MemoryTag.NATIVE_DEFAULT);
524+
}
525+
}
526+
527+
private static class PingTestClient extends WebSocketClient {
528+
final List<Consumer<WebSocketFrameHandler>> frameSequence = new ArrayList<>();
529+
boolean pingSent = false;
530+
private int nextFrame = 0;
531+
532+
PingTestClient() {
533+
super(DefaultHttpClientConfiguration.INSTANCE, PlainSocketFactory.INSTANCE);
534+
}
535+
536+
@Override
537+
public boolean isConnected() {
538+
return true;
539+
}
540+
541+
@Override
542+
public boolean receiveFrame(WebSocketFrameHandler handler, int timeout) {
543+
if (nextFrame < frameSequence.size()) {
544+
frameSequence.get(nextFrame++).accept(handler);
545+
return true;
546+
}
547+
return false;
548+
}
549+
550+
@Override
551+
public void sendBinary(long dataPtr, int length) {
552+
}
553+
554+
@Override
555+
public void sendPing(int timeout) {
556+
pingSent = true;
557+
}
558+
559+
@Override
560+
protected void ioWait(int timeout, int op) {
561+
}
562+
563+
@Override
564+
protected void setupIoWait() {
565+
}
566+
}
390567
}

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/WebSocketSendQueueTest.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -458,6 +458,68 @@ public void testDurableAckInterleavedWithStatusOk() throws Exception {
458458
});
459459
}
460460

461+
@Test
462+
public void testPingAndDrainCompletesSuccessfully() throws Exception {
463+
assertMemoryLeak(() -> {
464+
InFlightWindow window = new InFlightWindow(8, 5_000);
465+
WebSocketSendQueue queue = null;
466+
try (FakeWebSocketClient client = new FakeWebSocketClient()) {
467+
AtomicBoolean ackDelivered = new AtomicBoolean(false);
468+
client.setTryReceiveBehavior(handler -> {
469+
if (ackDelivered.compareAndSet(false, true)) {
470+
emitDurableAck(handler, 7);
471+
return true;
472+
}
473+
return false;
474+
});
475+
476+
queue = new WebSocketSendQueue(client, window, 1_000, 500);
477+
478+
queue.pingAndDrain();
479+
480+
assertEquals(7, queue.getHighestDurableSequence());
481+
} finally {
482+
closeQuietly(queue);
483+
}
484+
});
485+
}
486+
487+
@Test
488+
public void testPingAndDrainWithInFlightBatches() throws Exception {
489+
assertMemoryLeak(() -> {
490+
InFlightWindow window = new InFlightWindow(8, 5_000);
491+
WebSocketSendQueue queue = null;
492+
try (FakeWebSocketClient client = new FakeWebSocketClient()) {
493+
window.addInFlight(0);
494+
window.addInFlight(1);
495+
496+
AtomicInteger callCount = new AtomicInteger();
497+
client.setTryReceiveBehavior(handler -> {
498+
int n = callCount.getAndIncrement();
499+
switch (n) {
500+
case 0:
501+
emitAck(handler, 1);
502+
return true;
503+
case 1:
504+
emitDurableAck(handler, 0);
505+
return true;
506+
default:
507+
return false;
508+
}
509+
});
510+
511+
queue = new WebSocketSendQueue(client, window, 1_000, 500);
512+
513+
queue.pingAndDrain();
514+
515+
assertEquals(0, window.getInFlightCount());
516+
assertEquals(0, queue.getHighestDurableSequence());
517+
} finally {
518+
closeQuietly(queue);
519+
}
520+
});
521+
}
522+
461523
@Test
462524
public void testHighestDurableSequenceInitiallyMinusOne() throws Exception {
463525
assertMemoryLeak(() -> {
@@ -573,6 +635,10 @@ public void sendBinary(long dataPtr, int length) {
573635
sendBehavior.send(dataPtr, length);
574636
}
575637

638+
@Override
639+
public void sendPing(int timeout) {
640+
}
641+
576642
public void setSendBehavior(SendBehavior sendBehavior) {
577643
this.sendBehavior = sendBehavior;
578644
}

0 commit comments

Comments
 (0)