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

feat: add gb32960 session diagnostics
Some checks failed
ci/woodpecker/push/woodpecker Pipeline was canceled
Details

Browse Source
This commit is contained in:
lingniu
2026-06-29 21:17:34 +08:00
parent b1bd67d4d2
commit 8b7a1ae76b
10 changed files with 285 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java
View File

@@ -42,6 +42,7 @@ import com.lingniu.ingest.protocol.gb32960.codec.profile.VendorExtensionSelector
import com.lingniu.ingest.protocol.gb32960.handler.Gb32960RealtimeHandler;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960AccessService;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960AckService;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960ConnectionDiagnostics;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960FrameDiagnostics;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
import com.lingniu.ingest.protocol.gb32960.mapper.Gb32960EventMapper;
@@ -220,6 +221,12 @@ public class Gb32960AutoConfiguration {
props.getDiagnostics().getMaxLoggedFrameKeysPerChannel());
}
@Bean
@ConditionalOnMissingBean
public Gb32960ConnectionDiagnostics gb32960ConnectionDiagnostics() {
return new Gb32960ConnectionDiagnostics();
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "lingniu.ingest.gb32960.server", name = "enabled",
@@ -229,7 +236,8 @@ public class Gb32960AutoConfiguration {
Dispatcher dispatcher,
Gb32960AccessService accessService,
Gb32960AckService ackService,
Gb32960FrameDiagnostics diagnostics) {
Gb32960FrameDiagnostics diagnostics,
Gb32960ConnectionDiagnostics connectionDiagnostics) {
return new Gb32960NettyServer(
props.getPort(),
props.getWorkerThreads(),
@@ -239,6 +247,7 @@ public class Gb32960AutoConfiguration {
accessService,
ackService,
diagnostics,
connectionDiagnostics,
props.getTls());
}
}

14
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java
View File

@@ -64,26 +64,31 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
private final Gb32960AccessService accessService;
private final Gb32960AckService ackService;
private final Gb32960FrameDiagnostics diagnostics;
private final Gb32960ConnectionDiagnostics connectionDiagnostics;
public Gb32960ChannelHandler(Gb32960MessageDecoder decoder,
Dispatcher dispatcher,
Gb32960AccessService accessService,
Gb32960AckService ackService,
Gb32960FrameDiagnostics diagnostics) {
Gb32960FrameDiagnostics diagnostics,
Gb32960ConnectionDiagnostics connectionDiagnostics) {
this.decoder = decoder;
this.dispatcher = dispatcher;
this.accessService = accessService;
this.ackService = ackService;
this.diagnostics = diagnostics;
this.connectionDiagnostics = connectionDiagnostics;
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
connectionDiagnostics.opened(sessionId(ctx), addr(ctx));
log.info("[gb32960] connection opened peer={}", addr(ctx));
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
connectionDiagnostics.closed(sessionId(ctx));
log.info("[gb32960] connection closed peer={}", addr(ctx));
}
@@ -94,6 +99,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent idle && idle.state() == IdleState.READER_IDLE) {
connectionDiagnostics.readIdle(sessionId(ctx));
log.warn("[gb32960] read idle peer={} noInboundFrameFor>=thresholdSeconds (see idleReadSeconds)",
addr(ctx));
}
@@ -115,6 +121,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
String vin = msg.header().vin();
CommandType cmd = msg.header().command();
connectionDiagnostics.received(sessionId(ctx), cmd, vin);
// 原样回显请求帧的 17 字节 VIN（frame[4..20]），避免平台登入 VIN 全 0
// 被 vinPadded 重写为空格导致对端判定 VIN 不匹配。
byte[] rawVin = new byte[17];
@@ -339,6 +346,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
addr(ctx), pl.username(), pl.encryptRule(), pl.serialNo(), pl.loginTime(), result);
// 把 username 钉到 channel attribute，供后续 0x02/0x03 帧的 vendor profile 路由
accessService.bindPlatformAccount(ctx.channel(), pl.username());
connectionDiagnostics.platformLoginAccepted(sessionId(ctx), pl.username());
return true;
}
@@ -494,4 +502,8 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
return "unknown";
}
private static String sessionId(ChannelHandlerContext ctx) {
return ctx.channel().id().asLongText();
}
}

113
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnostics.java Normal file
View File

@@ -0,0 +1,113 @@
package com.lingniu.ingest.protocol.gb32960.inbound;
import com.lingniu.ingest.protocol.gb32960.model.CommandType;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class Gb32960ConnectionDiagnostics {
private final Clock clock;
private final ConcurrentMap<String, MutableSession> activeSessions = new ConcurrentHashMap<>();
public Gb32960ConnectionDiagnostics() {
this(Clock.systemUTC());
}
Gb32960ConnectionDiagnostics(Clock clock) {
this.clock = clock;
}
public void opened(String sessionId, String peer) {
Instant now = clock.instant();
activeSessions.put(sessionId, new MutableSession(sessionId, peer, now));
}
public void received(String sessionId, CommandType command, String vin) {
activeSessions.computeIfPresent(sessionId, (ignored, session) -> {
session.lastInboundAt = clock.instant();
session.lastCommandCode = command.code();
session.lastCommand = command.name();
session.vin = vin == null ? "" : vin;
return session;
});
}
public void platformLoginAccepted(String sessionId, String platformAccount) {
activeSessions.computeIfPresent(sessionId, (ignored, session) -> {
session.platformAccount = platformAccount == null ? "" : platformAccount;
return session;
});
}
public void readIdle(String sessionId) {
activeSessions.computeIfPresent(sessionId, (ignored, session) -> {
session.readIdleCount++;
return session;
});
}
public void closed(String sessionId) {
activeSessions.remove(sessionId);
}
public List<SessionSnapshot> activeSessions() {
Instant now = clock.instant();
return activeSessions.values().stream()
.map(session -> session.snapshot(now))
.sorted(Comparator.comparing(SessionSnapshot::connectedAt))
.toList();
}
public record SessionSnapshot(
String sessionId,
String peer,
Instant connectedAt,
Instant lastInboundAt,
long idleSeconds,
int lastCommandCode,
String lastCommand,
String vin,
String platformAccount,
long readIdleCount) {
}
private static final class MutableSession {
private final String sessionId;
private final String peer;
private final Instant connectedAt;
private Instant lastInboundAt;
private int lastCommandCode;
private String lastCommand = "";
private String vin = "";
private String platformAccount = "";
private long readIdleCount;
private MutableSession(String sessionId, String peer, Instant connectedAt) {
this.sessionId = sessionId;
this.peer = peer;
this.connectedAt = connectedAt;
this.lastInboundAt = connectedAt;
}
private SessionSnapshot snapshot(Instant now) {
long idleSeconds = Math.max(0, Duration.between(lastInboundAt, now).toSeconds());
return new SessionSnapshot(
sessionId,
peer,
connectedAt,
lastInboundAt,
idleSeconds,
lastCommandCode,
lastCommand,
vin,
platformAccount,
readIdleCount);
}
}
}

6
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960NettyServer.java
View File

@@ -50,6 +50,7 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean {
private final Gb32960AccessService accessService;
private final Gb32960AckService ackService;
private final Gb32960FrameDiagnostics diagnostics;
private final Gb32960ConnectionDiagnostics connectionDiagnostics;
private final Gb32960Properties.Tls tlsConfig;
private EventLoopGroup boss;
@@ -63,6 +64,7 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean {
Gb32960AccessService accessService,
Gb32960AckService ackService,
Gb32960FrameDiagnostics diagnostics,
Gb32960ConnectionDiagnostics connectionDiagnostics,
Gb32960Properties.Tls tlsConfig) {
this.port = port;
this.workerThreads = workerThreads;
@@ -72,6 +74,7 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean {
this.accessService = accessService;
this.ackService = ackService;
this.diagnostics = diagnostics;
this.connectionDiagnostics = connectionDiagnostics;
this.tlsConfig = tlsConfig;
}
@@ -113,7 +116,8 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean {
.addLast("frame", new Gb32960FrameDecoder())
.addLast("dispatch",
new Gb32960ChannelHandler(messageDecoder, dispatcher,
accessService, ackService, diagnostics));
accessService, ackService, diagnostics,
connectionDiagnostics));
}
});
this.serverChannel = b.bind(port).sync().channel();

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

@@ -139,7 +139,8 @@ class Gb32960ChannelHandlerAckBoundaryTest {
dispatcher,
accessService(),
new Gb32960AckService(),
new Gb32960FrameDiagnostics(16));
new Gb32960FrameDiagnostics(16),
new Gb32960ConnectionDiagnostics());
}
private static Gb32960MessageDecoder decoder() {

78
modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnosticsTest.java Normal file
View File

@@ -0,0 +1,78 @@
package com.lingniu.ingest.protocol.gb32960.inbound;
import com.lingniu.ingest.protocol.gb32960.model.CommandType;
import org.junit.jupiter.api.Test;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import static org.assertj.core.api.Assertions.assertThat;
class Gb32960ConnectionDiagnosticsTest {
private static final Instant NOW = Instant.parse("2026-06-29T10:00:00Z");
@Test
void tracksActiveSessionCommandPlatformAccountAndReadIdleCount() {
MutableClock clock = new MutableClock(NOW);
Gb32960ConnectionDiagnostics diagnostics = new Gb32960ConnectionDiagnostics(clock);
diagnostics.opened("session-1", "127.0.0.1:65469");
clock.advanceSeconds(5);
diagnostics.received("session-1", CommandType.PLATFORM_LOGIN, "");
diagnostics.platformLoginAccepted("session-1", "Hyundai");
clock.advanceSeconds(30);
diagnostics.readIdle("session-1");
assertThat(diagnostics.activeSessions()).singleElement().satisfies(session -> {
assertThat(session.sessionId()).isEqualTo("session-1");
assertThat(session.peer()).isEqualTo("127.0.0.1:65469");
assertThat(session.connectedAt()).isEqualTo(NOW);
assertThat(session.lastInboundAt()).isEqualTo(NOW.plusSeconds(5));
assertThat(session.idleSeconds()).isEqualTo(30);
assertThat(session.lastCommandCode()).isEqualTo(0x05);
assertThat(session.lastCommand()).isEqualTo("PLATFORM_LOGIN");
assertThat(session.vin()).isEmpty();
assertThat(session.platformAccount()).isEqualTo("Hyundai");
assertThat(session.readIdleCount()).isEqualTo(1);
});
}
@Test
void removesSessionWhenChannelCloses() {
Gb32960ConnectionDiagnostics diagnostics = new Gb32960ConnectionDiagnostics(Clock.fixed(NOW, ZoneOffset.UTC));
diagnostics.opened("session-1", "127.0.0.1:65469");
diagnostics.closed("session-1");
assertThat(diagnostics.activeSessions()).isEmpty();
}
private static final class MutableClock extends Clock {
private Instant instant;
private MutableClock(Instant instant) {
this.instant = instant;
}
void advanceSeconds(long seconds) {
instant = instant.plusSeconds(seconds);
}
@Override
public ZoneOffset getZone() {
return ZoneOffset.UTC;
}
@Override
public Clock withZone(java.time.ZoneId zone) {
return this;
}
@Override
public Instant instant() {
return instant;
}
}
}