diff --git a/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java b/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java index a94aa5d1..43a15f68 100644 --- a/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java +++ b/modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java @@ -25,6 +25,7 @@ import com.lingniu.ingest.session.SessionStore; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.util.AttributeKey; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,6 +35,7 @@ import java.time.Instant; import java.util.HashMap; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; /** * JT808 Netty 入站处理器。职责： @@ -49,6 +51,8 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler { private static final Logger log = LoggerFactory.getLogger(Jt808ChannelHandler.class); private static final String PROCESSING_ERROR_PREFIX = "processing failed: "; + private static final AttributeKey> DURABLE_ACK_CHAIN_ATTR = + AttributeKey.valueOf("jt808.durableAckChain"); private final Jt808MessageDecoder decoder; private final Dispatcher dispatcher; @@ -119,25 +123,21 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler { // ====== 会话维护 & 同步应答匹配 ====== String phone = msg.header().phone(); IdentityResolution identity = resolveIdentity(msg); + byte[] durableAck = null; switch (msg.header().messageId()) { case Jt808MessageId.TERMINAL_REGISTER -> { DeviceSession session = upsertSession(ctx, msg, identity.identity()); channelRegistry.bind(phone, ctx.channel()); - byte[] registerAck = Jt808FrameEncoder.encode( - Jt808Commands.registerAck(msg.header().serialNo(), 0, session.token()).messageId(), - phone, - channelRegistry.nextSerial(phone), - Jt808Commands.registerAck(msg.header().serialNo(), 0, session.token()).body()); - ctx.writeAndFlush(Unpooled.wrappedBuffer(registerAck)); + durableAck = registerAckFrame(phone, msg.header().serialNo(), session.token()); } case Jt808MessageId.TERMINAL_AUTH -> { upsertSession(ctx, msg, identity.identity()); channelRegistry.bind(phone, ctx.channel()); - ackTerminal(ctx, phone, msg.header().serialNo(), msg.header().messageId(), 0); + durableAck = terminalAckFrame(phone, msg.header().serialNo(), msg.header().messageId(), 0); } case Jt808MessageId.TERMINAL_HEARTBEAT -> { channelRegistry.bind(phone, ctx.channel()); - ackTerminal(ctx, phone, msg.header().serialNo(), msg.header().messageId(), 0); + durableAck = terminalAckFrame(phone, msg.header().serialNo(), msg.header().messageId(), 0); } case Jt808MessageId.TERMINAL_GENERAL_RESPONSE -> { // body: ackSerial(2) + ackMsgId(2) + result(1) @@ -173,7 +173,63 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler { frame, meta, Instant.now()); - dispatcher.dispatch(rf); + if (durableAck == null) { + dispatcher.dispatch(rf); + return; + } + enqueueDurableAck(ctx, phone, msg.header().messageId(), dispatcher.dispatchAndAwait(rf), durableAck); + } + + private void enqueueDurableAck(ChannelHandlerContext ctx, + String phone, + int messageId, + CompletableFuture dispatchFuture, + byte[] ackFrame) { + CompletableFuture previous = ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).get(); + if (previous == null) { + previous = CompletableFuture.completedFuture(null); + } + CompletableFuture next = previous + .handle((ignored, previousFailure) -> null) + .thenCompose(ignored -> dispatchFuture.handle((ok, failure) -> failure)) + .thenCompose(failure -> runOnEventLoop(ctx, + () -> handleDurableAckResult(ctx, phone, messageId, failure, ackFrame))); + ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).set(next); + } + + private CompletableFuture runOnEventLoop(ChannelHandlerContext ctx, Runnable task) { + CompletableFuture done = new CompletableFuture<>(); + Runnable wrapped = () -> { + try { + task.run(); + done.complete(null); + } catch (Throwable t) { + done.completeExceptionally(t); + } + }; + if (ctx.executor().inEventLoop()) { + wrapped.run(); + } else { + ctx.executor().execute(wrapped); + } + return done; + } + + private void handleDurableAckResult(ChannelHandlerContext ctx, + String phone, + int messageId, + Throwable failure, + byte[] ackFrame) { + if (!ctx.channel().isActive()) { + return; + } + if (failure != null) { + log.warn("[jt808] dispatch failed before ack peer={} phone={} msgId=0x{}", + addr(ctx), phone, Integer.toHexString(messageId), failure); + ctx.close(); + return; + } + ctx.writeAndFlush(Unpooled.wrappedBuffer(ackFrame)); } private IdentityResolution resolveIdentity(Jt808Message msg) { @@ -341,10 +397,14 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler { return value == null ? "" : value; } - private void ackTerminal(ChannelHandlerContext ctx, String phone, int ackSerial, int ackMsgId, int result) { + private byte[] registerAckFrame(String phone, int ackSerial, String token) { + var cmd = Jt808Commands.registerAck(ackSerial, 0, token); + return Jt808FrameEncoder.encode(cmd.messageId(), phone, channelRegistry.nextSerial(phone), cmd.body()); + } + + private byte[] terminalAckFrame(String phone, int ackSerial, int ackMsgId, int result) { var cmd = Jt808Commands.platformAck(ackSerial, ackMsgId, result); - byte[] frame = Jt808FrameEncoder.encode(cmd.messageId(), phone, channelRegistry.nextSerial(phone), cmd.body()); - ctx.writeAndFlush(Unpooled.wrappedBuffer(frame)); + return Jt808FrameEncoder.encode(cmd.messageId(), phone, channelRegistry.nextSerial(phone), cmd.body()); } private static String generateToken(String phone) { diff --git a/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java index cb774bc4..d089c6fc 100644 --- a/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java +++ b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java @@ -29,6 +29,7 @@ import com.lingniu.ingest.protocol.jt808.model.Jt808MessageId; import com.lingniu.ingest.protocol.jt808.session.Jt808ChannelRegistry; import com.lingniu.ingest.protocol.jt808.session.Jt808PendingRequests; import com.lingniu.ingest.session.InMemorySessionStore; +import io.netty.buffer.ByteBuf; import io.netty.channel.embedded.EmbeddedChannel; import org.junit.jupiter.api.Test; @@ -49,7 +50,7 @@ class Jt808ChannelHandlerTest { InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService(); identity.bind(new VehicleIdentityBinding( ProtocolId.JT808, "LNVIN000000000808", "123456789012", "DEV808", "B80808")); - DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of()); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink())); AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); Dispatcher dispatcher = new Dispatcher( new HandlerRegistry(), @@ -88,7 +89,7 @@ class Jt808ChannelHandlerTest { void unresolvedRegisterDoesNotBindUnknownVinToExternalIdentifiers() { InMemorySessionStore sessions = new InMemorySessionStore(); InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService(); - DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of()); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink())); AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); Dispatcher dispatcher = new Dispatcher( new HandlerRegistry(), @@ -129,7 +130,7 @@ class Jt808ChannelHandlerTest { InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService(); identity.bind(new VehicleIdentityBinding( ProtocolId.JT808, "LNVIN000000AUTH01", "123456789012", "123456789012345", "")); - DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of()); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink())); AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); Dispatcher dispatcher = new Dispatcher( new HandlerRegistry(), @@ -161,10 +162,48 @@ class Jt808ChannelHandlerTest { eventBus.close(); } + @Test + void registerAckWaitsForDispatchDurability() { + ControlledSink sink = new ControlledSink(); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink)); + AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); + Dispatcher dispatcher = new Dispatcher( + new HandlerRegistry(), + new InterceptorChain(List.of()), + new HandlerInvoker(), + eventBus, + batchExecutor); + Jt808ChannelHandler handler = new Jt808ChannelHandler( + new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser()))), + dispatcher, + new InMemorySessionStore(), + new InMemoryVehicleIdentityService(), + new Jt808ChannelRegistry(), + new Jt808PendingRequests()); + EmbeddedChannel channel = new EmbeddedChannel(handler); + + channel.writeInbound(buildFrame( + Jt808MessageId.TERMINAL_REGISTER, + "123456789012", + 1, + buildRegisterBody("DEV808", "B80808"))); + + assertThat((Object) channel.readOutbound()).isNull(); + sink.awaitPublishCount(1); + sink.future(0).complete(null); + + ByteBuf ack = readOutboundAfterRunningTasks(channel); + assertThat(ack).isNotNull(); + ack.release(); + + batchExecutor.close(); + eventBus.close(); + } + @Test void inactiveChannelRemovesSessionStoreEntry() { InMemorySessionStore sessions = new InMemorySessionStore(); - DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of()); + DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink())); AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); Dispatcher dispatcher = new Dispatcher( new HandlerRegistry(), @@ -596,6 +635,62 @@ class Jt808ChannelHandlerTest { os.write((int) (v & 0xFF)); } + 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 final class ControlledSink implements EventSink { + private final List> futures = new CopyOnWriteArrayList<>(); + + @Override + public String name() { + return "kafka"; + } + + @Override + public CompletableFuture publish(VehicleEvent event) { + CompletableFuture future = new CompletableFuture<>(); + futures.add(future); + return future; + } + + private CompletableFuture future(int index) { + return futures.get(index); + } + + private void awaitPublishCount(int expected) { + long deadline = System.currentTimeMillis() + 3000; + while (System.currentTimeMillis() < deadline) { + if (futures.size() >= expected) { + return; + } + Thread.onSpinWait(); + } + throw new AssertionError("expected " + expected + " sink publishes, actual=" + futures.size()); + } + } + + private static final class ImmediateKafkaSink implements EventSink { + @Override + public String name() { + return "kafka"; + } + + @Override + public CompletableFuture publish(VehicleEvent event) { + return CompletableFuture.completedFuture(null); + } + } + private static final class RecordingSink implements EventSink { private final List events = new CopyOnWriteArrayList<>(); private final CountDownLatch latch; @@ -606,7 +701,7 @@ class Jt808ChannelHandlerTest { @Override public String name() { - return "recording"; + return "kafka"; } @Override