From 633b3ea9c9dedfec23770ecec622209425dc4d7c Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 23 Jun 2026 17:10:46 +0800 Subject: [PATCH] feat: ack gb32960 after kafka boundary --- .../core/concurrency/DisruptorEventBus.java | 41 ++++-- .../ingest/core/dispatcher/Dispatcher.java | 76 ++++++++-- .../inbound/Gb32960ChannelHandler.java | 119 ++++++++++++---- .../Gb32960ChannelHandlerAckBoundaryTest.java | 133 ++++++++++++++++++ 4 files changed, 322 insertions(+), 47 deletions(-) create mode 100644 modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java diff --git a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java index 6e567e14..bf586cde 100644 --- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java +++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java @@ -14,6 +14,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicLong; @@ -28,10 +29,12 @@ public final class DisruptorEventBus implements AutoCloseable { private static final Logger log = LoggerFactory.getLogger(DisruptorEventBus.class); private final Disruptor disruptor; + private final List sinks; private final AtomicLong published = new AtomicLong(); private final AtomicLong failed = new AtomicLong(); public DisruptorEventBus(int ringBufferSize, String waitStrategyName, List sinks) { + this.sinks = List.copyOf(sinks); ThreadFactory tf = Thread.ofVirtual().name("ingest-bus-", 0).factory(); this.disruptor = new Disruptor<>( VehicleEventSlot::new, @@ -40,14 +43,14 @@ public final class DisruptorEventBus implements AutoCloseable { ProducerType.MULTI, waitStrategy(waitStrategyName)); - EventHandler[] handlers = sinks.stream() + EventHandler[] handlers = this.sinks.stream() .map(this::toHandler) .toArray(EventHandler[]::new); // 每个 sink 一个独立 handler，事件按扇出模式同时写 Kafka、event-file-store 等目标。 disruptor.handleEventsWith(handlers); disruptor.start(); log.info("DisruptorEventBus started ringBuffer={} wait={} sinks={}", - ringBufferSize, waitStrategyName, sinks.size()); + ringBufferSize, waitStrategyName, this.sinks.size()); } public void publish(VehicleEvent event) { @@ -55,6 +58,18 @@ public final class DisruptorEventBus implements AutoCloseable { published.incrementAndGet(); } + public CompletableFuture publishAndAwait(VehicleEvent event) { + published.incrementAndGet(); + List> futures = sinks.stream() + .filter(sink -> sink.accepts(event)) + .map(sink -> publishToSinkAndTrack(sink, event)) + .toList(); + if (futures.isEmpty()) { + return CompletableFuture.completedFuture(null); + } + return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)); + } + @Override public void close() { disruptor.shutdown(); @@ -66,18 +81,28 @@ public final class DisruptorEventBus implements AutoCloseable { VehicleEvent e = slot.event; if (e == null || !sink.accepts(e)) return; try { - sink.publish(e).exceptionally(ex -> { - // sink 异步失败只计数和打日志，不反向阻塞入站连接；生产侧靠 sink 自身重试/DLQ 保证可观测。 - failed.incrementAndGet(); - log.warn("sink {} publish failed", sink.name(), ex); - return null; - }); + publishToSinkAndTrack(sink, e); } finally { if (endOfBatch) slot.clear(); } }; } + private CompletableFuture publishToSinkAndTrack(EventSink sink, VehicleEvent event) { + try { + return sink.publish(event).whenComplete((ignored, ex) -> { + if (ex == null) return; + // sink 异步失败只计数和打日志；await 调用方仍会收到异常并决定是否 ACK。 + failed.incrementAndGet(); + log.warn("sink {} publish failed", sink.name(), ex); + }); + } catch (Throwable t) { + failed.incrementAndGet(); + log.warn("sink {} publish failed", sink.name(), t); + return CompletableFuture.failedFuture(t); + } + } + private static WaitStrategy waitStrategy(String name) { return switch (name == null ? "yielding" : name.toLowerCase()) { case "blocking" -> new BlockingWaitStrategy(); diff --git a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java index 32ab96c2..02b7e78c 100644 --- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java +++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java @@ -10,11 +10,16 @@ import com.lingniu.ingest.core.pipeline.InterceptorChain; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.time.Duration; import java.time.Instant; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; /** @@ -52,13 +57,61 @@ public final class Dispatcher { public void dispatch(RawFrame frame) { IngestContext ctx = new IngestContext(UUID.randomUUID().toString()); try { - // 在 interceptor 之前发 RawArchive，保证原始字节被无条件落盘（dedup/rate-limit - // 不会过滤它），满足"原始可回放"目标。archive 的写盘靠下游 ArchiveEventSink 消费。 - RawArchiveLookup rawArchive = emitRawArchive(frame, ctx); + DispatchPlan plan = mapFrameToEvents(frame, ctx); + for (VehicleEvent e : plan.events()) { + eventBus.publish(e); + } + if (plan.failure() != null) { + throw plan.failure(); + } + } catch (Throwable t) { + log.error("dispatch failure traceId={}", ctx.traceId(), t); + interceptors.onError(t, ctx); + } + } + public CompletableFuture dispatchAndAwait(RawFrame frame) { + IngestContext ctx = new IngestContext(UUID.randomUUID().toString()); + try { + DispatchPlan plan = mapFrameToEvents(frame, ctx); + CompletableFuture[] futures = plan.events().stream() + .map(eventBus::publishAndAwait) + .toArray(CompletableFuture[]::new); + CompletableFuture boundary = CompletableFuture.allOf(futures); + if (plan.failure() == null) { + return boundary; + } + return boundary.handle((ignored, publishFailure) -> { + if (publishFailure != null) { + throw new CompletionException(publishFailure); + } + throw new CompletionException(plan.failure()); + }); + } catch (Throwable t) { + log.error("dispatch failure traceId={}", ctx.traceId(), t); + interceptors.onError(t, ctx); + return CompletableFuture.failedFuture(t); + } + } + + public CompletableFuture dispatchAndAwait(RawFrame frame, Duration timeout) { + CompletableFuture future = dispatchAndAwait(frame); + if (timeout == null || timeout.isZero() || timeout.isNegative()) { + return future; + } + return future.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS); + } + + private DispatchPlan mapFrameToEvents(RawFrame frame, IngestContext ctx) { + List out = new ArrayList<>(); + // 在 interceptor 之前发 RawArchive，保证原始字节被无条件落盘（dedup/rate-limit + // 不会过滤它），满足"原始可回放"目标。archive 的写盘靠下游 Sink 消费。 + RawArchiveLookup rawArchive = appendRawArchive(frame, ctx, out); + + try { if (!interceptors.before(frame, ctx)) { log.debug("frame aborted: {}", ctx.abortReason()); - return; + return new DispatchPlan(out, null); } List handlers = registry.resolve( @@ -67,7 +120,7 @@ public final class Dispatcher { log.debug("no handler for {} cmd=0x{} info=0x{}", frame.protocolId(), Integer.toHexString(frame.command()), Integer.toHexString(frame.infoType())); - return; + return new DispatchPlan(out, null); } for (HandlerDefinition def : handlers) { @@ -79,12 +132,12 @@ public final class Dispatcher { for (VehicleEvent e : events) { e = enrichWithRawArchive(e, rawArchive); interceptors.after(e, ctx); - eventBus.publish(e); + out.add(e); } } + return new DispatchPlan(out, null); } catch (Throwable t) { - log.error("dispatch failure traceId={}", ctx.traceId(), t); - interceptors.onError(t, ctx); + return new DispatchPlan(out, t); } } @@ -95,7 +148,7 @@ public final class Dispatcher { *

VIN 取自 sourceMeta 里的 {@code vin} key（由各入站适配器负责填充）；缺失时 * 留空字符串，archive sink 会用 "unknown-vin" 占位保证 key 可解析。 */ - private RawArchiveLookup emitRawArchive(RawFrame frame, IngestContext ctx) { + private RawArchiveLookup appendRawArchive(RawFrame frame, IngestContext ctx, List out) { byte[] bytes = frame.rawBytes(); if (bytes == null || bytes.length == 0) return RawArchiveLookup.empty(); @@ -117,7 +170,7 @@ public final class Dispatcher { frame.command(), frame.infoType(), bytes); - eventBus.publish(raw); + out.add(raw); return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key)); } @@ -182,4 +235,7 @@ public final class Dispatcher { return key == null || key.isBlank(); } } + + private record DispatchPlan(List events, Throwable failure) { + } } diff --git a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java index 55ff1c3d..cd23252b 100644 --- a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java +++ b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java @@ -122,41 +122,91 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { return; } - // 鉴权通过：按命令类型回 ACK，写成功后统一 dispatch 到通用管线。 + // 鉴权通过：先完成命令侧状态变更/日志，再统一 dispatch 到通用管线。 + // 需要成功 ACK 的已接收帧，等待 dispatch 的持久化边界完成后再回 ACK。 // 平台登入鉴权失败会短路 return，不进 dispatch。 switch (cmd) { - case VEHICLE_LOGIN -> handleVehicleLogin(ctx, msg, rawVin); - case VEHICLE_LOGOUT -> handleVehicleLogout(ctx, msg, rawVin); + case VEHICLE_LOGIN -> logVehicleLogin(ctx, msg); + case VEHICLE_LOGOUT -> logVehicleLogout(ctx, msg); case PLATFORM_LOGIN -> { if (!handlePlatformLogin(ctx, msg, rawVin)) return; } - case PLATFORM_LOGOUT -> handlePlatformLogout(ctx, msg, rawVin); - case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> handleReportOrHeartbeat(ctx, msg, rawVin); - case TIME_CALIBRATION -> handleTimeCalibration(ctx, msg, rawVin); + case PLATFORM_LOGOUT -> logPlatformLogout(ctx); + case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> logReportOrHeartbeat(ctx, msg); + case TIME_CALIBRATION -> { } default -> logOtherFrame(ctx, msg); } + RawFrame rf = rawFrame(ctx, msg, frame); + if (!requiresDurableAck(cmd)) { + dispatcher.dispatch(rf); + return; + } + + dispatcher.dispatchAndAwait(rf).whenComplete((ignored, ex) -> + ctx.executor().execute(() -> { + if (ex != null) { + log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}", + addr(ctx), vin, Integer.toHexString(cmd.code()), ex); + ctx.close(); + return; + } + writeAckForCommand(ctx, msg, rawVin); + })); + } + + private RawFrame rawFrame(ChannelHandlerContext ctx, Gb32960Message msg, byte[] frame) { Map sourceMeta = new HashMap<>(4); - sourceMeta.put("vin", vin); + sourceMeta.put("vin", msg.header().vin()); sourceMeta.put("peer", addr(ctx)); String platformAccount = accessService.platformAccount(ctx.channel()); if (platformAccount != null && !platformAccount.isBlank()) { sourceMeta.put("platformAccount", platformAccount); } // RawFrame.rawBytes 是后续 raw archive 和按 rawArchiveUri 回查的源头；不要在这里裁剪或重编码。 - RawFrame rf = new RawFrame( + return new RawFrame( ProtocolId.GB32960, - cmd.code(), + msg.header().command().code(), 0, msg, frame, sourceMeta, Instant.now()); - dispatcher.dispatch(rf); + } + + private static boolean requiresDurableAck(CommandType cmd) { + return switch (cmd) { + case VEHICLE_LOGIN, + VEHICLE_LOGOUT, + PLATFORM_LOGIN, + PLATFORM_LOGOUT, + REALTIME_REPORT, + RESEND_REPORT, + HEARTBEAT, + TIME_CALIBRATION -> true; + default -> false; + }; + } + + private void writeAckForCommand(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + switch (msg.header().command()) { + case VEHICLE_LOGIN -> writeVehicleLoginAck(ctx, msg, rawVin); + case VEHICLE_LOGOUT -> writeVehicleLogoutAck(ctx, msg, rawVin); + case PLATFORM_LOGIN -> writePlatformLoginAck(ctx, msg, rawVin); + case PLATFORM_LOGOUT -> writePlatformLogoutAck(ctx, msg, rawVin); + case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> writeReportOrHeartbeatAck(ctx, msg, rawVin); + case TIME_CALIBRATION -> writeTimeCalibrationAck(ctx, msg, rawVin); + default -> { } + } } /** 0x01 车辆登入：按 §6.3.2 带原采集时间回成功 ACK + INFO 日志。 */ - private void handleVehicleLogin(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + private void logVehicleLogin(ChannelHandlerContext ctx, Gb32960Message msg) { + log.info("[gb32960] vehicle login peer={} vin={} protocolVersion={}", + addr(ctx), msg.header().vin(), msg.header().protocolVersion()); + } + + private void writeVehicleLoginAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { ackService.writeResponse( ctx, msg.header().protocolVersion(), @@ -166,13 +216,14 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { eventOrNow(msg), null, "vehicle-login-ack"); - log.info("[gb32960] vehicle login peer={} vin={} protocolVersion={}", - addr(ctx), msg.header().vin(), msg.header().protocolVersion()); } /** 0x04 车辆登出：INFO 日志 + 带原采集时间回成功 ACK。 */ - private void handleVehicleLogout(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + private void logVehicleLogout(ChannelHandlerContext ctx, Gb32960Message msg) { log.info("[gb32960] vehicle logout peer={} vin={}", addr(ctx), msg.header().vin()); + } + + private void writeVehicleLogoutAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { ackService.writeResponse(ctx, msg.header().protocolVersion(), CommandType.VEHICLE_LOGOUT, @@ -189,8 +240,8 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { *

    *
  • 消息体解析缺失：关闭连接，返回 false 短路 *
  • 鉴权失败：写 {@code 0x02 OTHER_ERROR} NACK + 关闭连接，返回 false 短路 - *
  • 鉴权通过：把 username 钉到 channel attribute（供后续 vendor profile 路由）， - * 写成功 ACK + INFO 日志，返回 true 继续 dispatch + *
  • 鉴权通过：把 username 钉到 channel attribute（供后续 vendor profile 路由），INFO 日志， + * 返回 true 继续 dispatch；成功 ACK 等 dispatch 持久化边界完成后再写 *
* * @return true 表示鉴权通过应继续 dispatch；false 表示调用方应 short-circuit return @@ -221,6 +272,10 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { addr(ctx), pl.username(), pl.encryptRule(), pl.serialNo(), pl.loginTime(), result); // 把 username 钉到 channel attribute，供后续 0x02/0x03 帧的 vendor profile 路由 accessService.bindPlatformAccount(ctx.channel(), pl.username()); + return true; + } + + private void writePlatformLoginAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { ackService.writeResponse( ctx, msg.header().protocolVersion(), @@ -230,12 +285,14 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { eventOrNow(msg), null, "platform-login-ack"); - return true; } /** 0x06 平台登出：INFO 日志 + 带原采集时间回成功 ACK。 */ - private void handlePlatformLogout(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + private void logPlatformLogout(ChannelHandlerContext ctx) { log.info("[gb32960] platform logout peer={}", addr(ctx)); + } + + private void writePlatformLogoutAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { ackService.writeResponse(ctx, msg.header().protocolVersion(), CommandType.PLATFORM_LOGOUT, @@ -247,7 +304,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { } /** - * 0x02/0x03/0x07 实时 / 补发 / 心跳：按 §6.3.2 回应答帧，保留原帧采集时间。 + * 0x02/0x03/0x07 实时 / 补发 / 心跳：记录帧诊断；成功 ACK 在 dispatch 持久化边界后写回。 * 部分车端实现严格按规范，没收到应答会停止后续上报或重连。ack-tag 按 cmd.code 区分便于日志筛选。 * *

注意：应答成功不代表下游已完成持久化，只表示平台已接收并通过 Dispatcher 投递。 @@ -264,17 +321,9 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { * * 帧内字段全量 JSON 仍保留 DEBUG。 */ - private void handleReportOrHeartbeat(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + private void logReportOrHeartbeat(ChannelHandlerContext ctx, Gb32960Message msg) { CommandType cmd = msg.header().command(); String platformAccount = accessService.platformAccount(ctx.channel()); - ackService.writeResponse(ctx, - msg.header().protocolVersion(), - cmd, - ResponseFlag.SUCCESS, - rawVin, - eventOrNow(msg), - null, - "report-ack-0x" + Integer.toHexString(cmd.code())); if (cmd == CommandType.HEARTBEAT) { if (log.isDebugEnabled()) { log.debug("[gb32960] HEARTBEAT peer={} vin={} platformAccount={}", @@ -304,8 +353,20 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { } } + private void writeReportOrHeartbeatAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + CommandType cmd = msg.header().command(); + ackService.writeResponse(ctx, + msg.header().protocolVersion(), + cmd, + ResponseFlag.SUCCESS, + rawVin, + eventOrNow(msg), + null, + "report-ack-0x" + Integer.toHexString(cmd.code())); + } + /** 0x08 终端校时：平台应答 data 段为平台当前时间 6B（用 Instant.now()）。 */ - private void handleTimeCalibration(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { + private void writeTimeCalibrationAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) { ackService.writeResponse(ctx, msg.header().protocolVersion(), CommandType.TIME_CALIBRATION, diff --git a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java new file mode 100644 index 00000000..2c58b2ad --- /dev/null +++ b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java @@ -0,0 +1,133 @@ +package com.lingniu.ingest.protocol.gb32960.inbound; + +import com.lingniu.ingest.api.event.VehicleEvent; +import com.lingniu.ingest.api.sink.EventSink; +import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor; +import com.lingniu.ingest.core.concurrency.DisruptorEventBus; +import com.lingniu.ingest.core.dispatcher.Dispatcher; +import com.lingniu.ingest.core.dispatcher.HandlerInvoker; +import com.lingniu.ingest.core.dispatcher.HandlerRegistry; +import com.lingniu.ingest.core.pipeline.InterceptorChain; +import com.lingniu.ingest.protocol.gb32960.auth.Gb32960PlatformAuthorizer; +import com.lingniu.ingest.protocol.gb32960.auth.Gb32960VinAuthorizer; +import com.lingniu.ingest.protocol.gb32960.codec.Gb32960BodyParser; +import com.lingniu.ingest.protocol.gb32960.codec.Gb32960DecoderTest; +import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder; +import com.lingniu.ingest.protocol.gb32960.codec.InfoBlockParserRegistry; +import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.PositionV2016BlockParser; +import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.VehicleV2016BlockParser; +import com.lingniu.ingest.protocol.gb32960.config.Gb32960Properties; +import io.netty.buffer.ByteBuf; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; + +class Gb32960ChannelHandlerAckBoundaryTest { + + @Test + void realtimeReportAckWaitsForDispatchFutureCompletion() { + try (DispatchHarness dispatch = new DispatchHarness()) { + EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher())); + + assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse(); + assertThat((Object) channel.readOutbound()).isNull(); + + dispatch.completeDurability(); + channel.runPendingTasks(); + + ByteBuf ack = channel.readOutbound(); + assertThat(ack).isNotNull(); + assertThat(ack.getUnsignedByte(2)).isEqualTo((short) 0x02); + assertThat(ack.getUnsignedByte(3)).isEqualTo((short) 0x01); + ack.release(); + channel.finishAndReleaseAll(); + } + } + + @Test + void realtimeReportAckIsNotWrittenWhenDispatchFutureFails() { + try (DispatchHarness dispatch = new DispatchHarness()) { + EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher())); + + assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse(); + + dispatch.failDurability(); + channel.runPendingTasks(); + + assertThat((Object) channel.readOutbound()).isNull(); + assertThat(channel.isActive()).isFalse(); + channel.finishAndReleaseAll(); + } + } + + private static Gb32960ChannelHandler handlerWith(Dispatcher dispatcher) { + return new Gb32960ChannelHandler( + decoder(), + dispatcher, + accessService(), + new Gb32960AckService(), + new Gb32960FrameDiagnostics(16)); + } + + private static Gb32960MessageDecoder decoder() { + return new Gb32960MessageDecoder(new Gb32960BodyParser(new InfoBlockParserRegistry(List.of( + new VehicleV2016BlockParser(), + new PositionV2016BlockParser())))); + } + + private static Gb32960AccessService accessService() { + Gb32960Properties.Auth auth = new Gb32960Properties.Auth(); + auth.setEnabled(false); + return new Gb32960AccessService( + new Gb32960VinAuthorizer(auth), + new Gb32960PlatformAuthorizer(auth.getPlatforms())); + } + + private static final class DispatchHarness implements AutoCloseable { + private final ControlledSink sink = new ControlledSink(); + private final DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink)); + private final AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); + private final Dispatcher dispatcher = new Dispatcher( + new HandlerRegistry(), + new InterceptorChain(List.of()), + new HandlerInvoker(), + eventBus, + batchExecutor); + + Dispatcher dispatcher() { + return dispatcher; + } + + void completeDurability() { + sink.future.complete(null); + } + + void failDurability() { + sink.future.completeExceptionally(new IllegalStateException("durability failed")); + } + + @Override + public void close() { + batchExecutor.close(); + eventBus.close(); + } + } + + private static final class ControlledSink implements EventSink { + private final CompletableFuture future = new CompletableFuture<>(); + + @Override + public String name() { + return "controlled"; + } + + @Override + public CompletableFuture publish(VehicleEvent event) { + return future; + } + } +}