shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: stabilize gb32960 durable ack boundary

Browse Source
This commit is contained in:
lingniu
2026-07-01 04:18:33 +08:00
parent d5ef061cca
commit 2e54607bd6
1 changed files with 13 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java
View File

@@ -61,10 +61,9 @@ class Gb32960ChannelHandlerAckBoundaryTest {
dispatch.awaitPublishCount(1); dispatch.awaitPublishCount(1);
dispatch.failDurability(); dispatch.failDurability();
channel.runPendingTasks();
assertThat((Object) channel.readOutbound()).isNull(); assertThat((Object) channel.readOutbound()).isNull();
assertThat(channel.isActive()).isFalse(); assertThat(awaitInactive(channel)).isTrue();
channel.finishAndReleaseAll(); channel.finishAndReleaseAll();
} }
} }
@@ -151,6 +150,18 @@ class Gb32960ChannelHandlerAckBoundaryTest {
return null; return null;
} }
private static boolean awaitInactive(EmbeddedChannel channel) {
long deadline = System.currentTimeMillis() + 3000;
while (System.currentTimeMillis() < deadline) {
channel.runPendingTasks();
if (!channel.isActive()) {
return true;
}
Thread.onSpinWait();
}
return false;
}
private static final class DispatchHarness implements AutoCloseable { private static final class DispatchHarness implements AutoCloseable {
private final ControlledSink sink = new ControlledSink(); private final ControlledSink sink = new ControlledSink();
private final DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink)); private final DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink));