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-06-29 13:55:18 +08:00
parent 3904dcf869
commit 84de9b2dfa
1 changed files with 16 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -42,9 +42,8 @@ class Gb32960ChannelHandlerAckBoundaryTest {
dispatch.awaitPublishCount(1); dispatch.awaitPublishCount(1);
dispatch.completeDurability(); dispatch.completeDurability();
channel.runPendingTasks();
ByteBuf ack = channel.readOutbound(); ByteBuf ack = readOutboundAfterRunningTasks(channel);
assertThat(ack).isNotNull(); assertThat(ack).isNotNull();
assertThat(ack.getUnsignedByte(2)).isEqualTo((short) 0x02); assertThat(ack.getUnsignedByte(2)).isEqualTo((short) 0x02);
assertThat(ack.getUnsignedByte(3)).isEqualTo((short) 0x01); assertThat(ack.getUnsignedByte(3)).isEqualTo((short) 0x01);
@@ -85,10 +84,9 @@ class Gb32960ChannelHandlerAckBoundaryTest {
assertThat((Object) channel.readOutbound()).isNull(); assertThat((Object) channel.readOutbound()).isNull();
dispatch.completeDurability(0); dispatch.completeDurability(0);
channel.runPendingTasks();
ByteBuf firstAck = channel.readOutbound(); ByteBuf firstAck = readOutboundAfterRunningTasks(channel);
ByteBuf secondAck = channel.readOutbound(); ByteBuf secondAck = readOutboundAfterRunningTasks(channel);
assertThat(firstAck).isNotNull(); assertThat(firstAck).isNotNull();
assertThat(secondAck).isNotNull(); assertThat(secondAck).isNotNull();
assertThat(firstAck.getUnsignedByte(2)).isEqualTo((short) 0x02); assertThat(firstAck.getUnsignedByte(2)).isEqualTo((short) 0x02);
@@ -99,6 +97,19 @@ class Gb32960ChannelHandlerAckBoundaryTest {
} }
} }
private static ByteBuf readOutboundAfterRunningTasks(EmbeddedChannel channel) {
long deadline = System.currentTimeMillis() + 3000;
while (System.currentTimeMillis() < deadline) {
channel.runPendingTasks();
ByteBuf outbound = channel.readOutbound();
if (outbound != null) {
return outbound;
}
Thread.onSpinWait();
}
return null;
}
private static Gb32960ChannelHandler handlerWith(Dispatcher dispatcher) { private static Gb32960ChannelHandler handlerWith(Dispatcher dispatcher) {
return new Gb32960ChannelHandler( return new Gb32960ChannelHandler(
decoder(), decoder(),