From ce3f2ffb904adf50b9b283144fe12d34abbf22c2 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 09:15:28 +0800 Subject: [PATCH] refactor: remove production in-memory session store --- .../Gb32960IngestAppCompositionTest.java | 4 +- .../Jt808IngestAppCompositionTest.java | 4 +- modules/core/session-core/pom.xml | 4 - .../ingest/session/InMemorySessionStore.java | 84 ------------------- .../SessionCoreAutoConfigurationTest.java | 9 +- .../protocol/jt808/TestSessionStore.java | 74 ++++++++++++++++ .../downlink/Jt808CommandDispatcherTest.java | 4 +- .../inbound/Jt808ChannelHandlerTest.java | 28 +++---- 8 files changed, 101 insertions(+), 110 deletions(-) delete mode 100644 modules/core/session-core/src/main/java/com/lingniu/ingest/session/InMemorySessionStore.java create mode 100644 modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/TestSessionStore.java diff --git a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java index 8f122c60..aac52908 100644 --- a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java +++ b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java @@ -6,7 +6,7 @@ import com.lingniu.ingest.identity.config.VehicleIdentityAutoConfiguration; import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder; import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration; import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer; -import com.lingniu.ingest.session.InMemorySessionStore; +import com.lingniu.ingest.session.SessionStore; import com.lingniu.ingest.session.config.SessionCoreAutoConfiguration; import com.lingniu.ingest.sink.archive.ArchiveStore; import com.lingniu.ingest.sink.archive.RawArchiveEventSink; @@ -36,7 +36,7 @@ class Gb32960IngestAppCompositionTest { Gb32960AutoConfiguration.class)) .withAllowBeanDefinitionOverriding(true) .withBean("kafkaProducer", KafkaProducer.class, Gb32960IngestAppCompositionTest::kafkaProducer) - .withBean(InMemorySessionStore.class, InMemorySessionStore::new) + .withBean(SessionStore.class, () -> mock(SessionStore.class)) .withPropertyValues( "lingniu.ingest.gb32960.enabled=true", "lingniu.ingest.gb32960.server.enabled=true", diff --git a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java b/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java index 4e5935b4..849fdbd9 100644 --- a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java +++ b/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java @@ -6,7 +6,7 @@ import com.lingniu.ingest.identity.config.VehicleIdentityAutoConfiguration; import com.lingniu.ingest.protocol.jt808.codec.Jt808MessageDecoder; import com.lingniu.ingest.protocol.jt808.config.Jt808AutoConfiguration; import com.lingniu.ingest.protocol.jt808.inbound.Jt808NettyServer; -import com.lingniu.ingest.session.InMemorySessionStore; +import com.lingniu.ingest.session.SessionStore; import com.lingniu.ingest.session.config.SessionCoreAutoConfiguration; import com.lingniu.ingest.sink.archive.ArchiveStore; import com.lingniu.ingest.sink.archive.RawArchiveEventSink; @@ -36,7 +36,7 @@ class Jt808IngestAppCompositionTest { Jt808AutoConfiguration.class)) .withAllowBeanDefinitionOverriding(true) .withBean("kafkaProducer", KafkaProducer.class, Jt808IngestAppCompositionTest::kafkaProducer) - .withBean(InMemorySessionStore.class, InMemorySessionStore::new) + .withBean(SessionStore.class, () -> mock(SessionStore.class)) .withPropertyValues( "lingniu.ingest.jt808.enabled=true", "lingniu.ingest.jt808.port=0", diff --git a/modules/core/session-core/pom.xml b/modules/core/session-core/pom.xml index 076d077f..7669ffa2 100644 --- a/modules/core/session-core/pom.xml +++ b/modules/core/session-core/pom.xml @@ -28,10 +28,6 @@ org.springframework.boot spring-boot-starter-data-redis - - com.github.ben-manes.caffeine - caffeine - org.junit.jupiter junit-jupiter diff --git a/modules/core/session-core/src/main/java/com/lingniu/ingest/session/InMemorySessionStore.java b/modules/core/session-core/src/main/java/com/lingniu/ingest/session/InMemorySessionStore.java deleted file mode 100644 index ccdb80d1..00000000 --- a/modules/core/session-core/src/main/java/com/lingniu/ingest/session/InMemorySessionStore.java +++ /dev/null @@ -1,84 +0,0 @@ -package com.lingniu.ingest.session; - -import com.github.benmanes.caffeine.cache.Cache; -import com.github.benmanes.caffeine.cache.Caffeine; - -import java.time.Duration; -import java.util.Optional; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.function.UnaryOperator; - -/** - * 测试/本地用纯内存会话存储。支持按 sessionId / vin / phone 三种 key 查询。 - * - *

失活清理：基于 {@link Caffeine} 的 {@code expireAfterAccess 30 分钟}。 - * 生产自动配置拒绝 memory 模式，应使用 Redis 会话索引。 - */ -public final class InMemorySessionStore implements SessionStore { - - private final Cache byId; - private final ConcurrentMap vinToId = new ConcurrentHashMap<>(); - private final ConcurrentMap phoneToId = new ConcurrentHashMap<>(); - - public InMemorySessionStore() { - this.byId = Caffeine.newBuilder() - .expireAfterAccess(Duration.ofMinutes(30)) - .removalListener((String sid, DeviceSession s, com.github.benmanes.caffeine.cache.RemovalCause c) -> { - if (s == null) return; - if (s.vin() != null) vinToId.remove(s.vin(), sid); - if (s.phone() != null) phoneToId.remove(s.phone(), sid); - }) - .build(); - } - - @Override - public void put(DeviceSession session) { - byId.put(session.sessionId(), session); - // 三索引用同一个 sessionId 汇聚；HTTP 下行可以拿 sessionId/VIN/phone 任意一种查在线连接。 - if (session.vin() != null) vinToId.put(session.vin(), session.sessionId()); - if (session.phone() != null) phoneToId.put(session.phone(), session.sessionId()); - } - - @Override - public Optional findBySessionId(String sessionId) { - return Optional.ofNullable(byId.getIfPresent(sessionId)); - } - - @Override - public Optional findByVin(String vin) { - String sid = vinToId.get(vin); - return sid == null ? Optional.empty() : findBySessionId(sid); - } - - @Override - public Optional findByPhone(String phone) { - String sid = phoneToId.get(phone); - return sid == null ? Optional.empty() : findBySessionId(sid); - } - - @Override - public synchronized Optional update(String sessionId, UnaryOperator updater) { - DeviceSession old = byId.getIfPresent(sessionId); - if (old == null) return Optional.empty(); - DeviceSession next = updater.apply(old); - // update 复用 put，保证 VIN/phone 二级索引和主会话对象一起刷新。 - put(next); - return Optional.of(next); - } - - @Override - public void remove(String sessionId) { - DeviceSession s = byId.getIfPresent(sessionId); - if (s != null) { - byId.invalidate(sessionId); - if (s.vin() != null) vinToId.remove(s.vin(), sessionId); - if (s.phone() != null) phoneToId.remove(s.phone(), sessionId); - } - } - - @Override - public int size() { - return (int) byId.estimatedSize(); - } -} diff --git a/modules/core/session-core/src/test/java/com/lingniu/ingest/session/config/SessionCoreAutoConfigurationTest.java b/modules/core/session-core/src/test/java/com/lingniu/ingest/session/config/SessionCoreAutoConfigurationTest.java index c94d6893..7b42483e 100644 --- a/modules/core/session-core/src/test/java/com/lingniu/ingest/session/config/SessionCoreAutoConfigurationTest.java +++ b/modules/core/session-core/src/test/java/com/lingniu/ingest/session/config/SessionCoreAutoConfigurationTest.java @@ -1,6 +1,5 @@ package com.lingniu.ingest.session.config; -import com.lingniu.ingest.session.InMemorySessionStore; import com.lingniu.ingest.session.RedisSessionStore; import com.lingniu.ingest.session.SessionStore; import org.junit.jupiter.api.Test; @@ -9,6 +8,7 @@ import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.data.redis.core.StringRedisTemplate; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; class SessionCoreAutoConfigurationTest { @@ -20,10 +20,15 @@ class SessionCoreAutoConfigurationTest { void doesNotCreateImplicitInMemorySessionStoreByDefault() { contextRunner.run(context -> { assertThat(context).doesNotHaveBean(SessionStore.class); - assertThat(context).doesNotHaveBean(InMemorySessionStore.class); }); } + @Test + void doesNotPublishInMemorySessionStoreImplementation() { + assertThatThrownBy(() -> Class.forName("com.lingniu.ingest.session.InMemorySessionStore")) + .isInstanceOf(ClassNotFoundException.class); + } + @Test void usesRedisSessionStoreWhenEnabledAndRedisTemplateExists() { contextRunner diff --git a/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/TestSessionStore.java b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/TestSessionStore.java new file mode 100644 index 00000000..dc1be186 --- /dev/null +++ b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/TestSessionStore.java @@ -0,0 +1,74 @@ +package com.lingniu.ingest.protocol.jt808; + +import com.lingniu.ingest.session.DeviceSession; +import com.lingniu.ingest.session.SessionStore; + +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.function.UnaryOperator; + +public final class TestSessionStore implements SessionStore { + + private final ConcurrentMap byId = new ConcurrentHashMap<>(); + private final ConcurrentMap vinToId = new ConcurrentHashMap<>(); + private final ConcurrentMap phoneToId = new ConcurrentHashMap<>(); + + @Override + public void put(DeviceSession session) { + byId.put(session.sessionId(), session); + if (session.vin() != null) { + vinToId.put(session.vin(), session.sessionId()); + } + if (session.phone() != null) { + phoneToId.put(session.phone(), session.sessionId()); + } + } + + @Override + public Optional findBySessionId(String sessionId) { + return Optional.ofNullable(byId.get(sessionId)); + } + + @Override + public Optional findByVin(String vin) { + String sessionId = vinToId.get(vin); + return sessionId == null ? Optional.empty() : findBySessionId(sessionId); + } + + @Override + public Optional findByPhone(String phone) { + String sessionId = phoneToId.get(phone); + return sessionId == null ? Optional.empty() : findBySessionId(sessionId); + } + + @Override + public synchronized Optional update(String sessionId, UnaryOperator updater) { + DeviceSession current = byId.get(sessionId); + if (current == null) { + return Optional.empty(); + } + DeviceSession next = updater.apply(current); + put(next); + return Optional.of(next); + } + + @Override + public void remove(String sessionId) { + DeviceSession removed = byId.remove(sessionId); + if (removed == null) { + return; + } + if (removed.vin() != null) { + vinToId.remove(removed.vin(), sessionId); + } + if (removed.phone() != null) { + phoneToId.remove(removed.phone(), sessionId); + } + } + + @Override + public int size() { + return byId.size(); + } +} diff --git a/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/downlink/Jt808CommandDispatcherTest.java b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/downlink/Jt808CommandDispatcherTest.java index f8a3413c..94a92435 100644 --- a/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/downlink/Jt808CommandDispatcherTest.java +++ b/modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/downlink/Jt808CommandDispatcherTest.java @@ -10,7 +10,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.DeviceSession; -import com.lingniu.ingest.session.InMemorySessionStore; +import com.lingniu.ingest.protocol.jt808.TestSessionStore; import io.netty.buffer.ByteBuf; import io.netty.channel.embedded.EmbeddedChannel; import org.junit.jupiter.api.Test; @@ -26,7 +26,7 @@ class Jt808CommandDispatcherTest { @Test void notifyWritesSubpackageFramesForLargeBody() { - InMemorySessionStore sessions = new InMemorySessionStore(); + TestSessionStore sessions = new TestSessionStore(); sessions.put(new DeviceSession("sid-1", ProtocolId.JT808, "LNVIN000000000303", "123456789012", "", "127.0.0.1:10000", Instant.now(), Instant.now(), Map.of())); Jt808ChannelRegistry channels = new Jt808ChannelRegistry(); 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 b1471dd9..7f29f78b 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 @@ -28,7 +28,7 @@ import com.lingniu.ingest.protocol.jt808.model.Jt808Message; 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 com.lingniu.ingest.protocol.jt808.TestSessionStore; import io.netty.buffer.ByteBuf; import io.netty.channel.embedded.EmbeddedChannel; import org.junit.jupiter.api.Test; @@ -47,7 +47,7 @@ class Jt808ChannelHandlerTest { @Test void registerSessionUsesResolvedInternalVinInsteadOfDeviceIdPlaceholder() { - InMemorySessionStore sessions = new InMemorySessionStore(); + TestSessionStore sessions = new TestSessionStore(); InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService(); identity.bind(new VehicleIdentityBinding( ProtocolId.JT808, "LNVIN000000000808", "123456789012", "DEV808", "B80808")); @@ -88,7 +88,7 @@ class Jt808ChannelHandlerTest { @Test void unresolvedRegisterBindsGeneratedInternalVinToExternalIdentifiers() { - InMemorySessionStore sessions = new InMemorySessionStore(); + TestSessionStore sessions = new TestSessionStore(); InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService(); DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink())); AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); @@ -127,7 +127,7 @@ class Jt808ChannelHandlerTest { @Test void authSessionUsesResolvedInternalVinFromImei() { - InMemorySessionStore sessions = new InMemorySessionStore(); + TestSessionStore sessions = new TestSessionStore(); InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService(); identity.bind(new VehicleIdentityBinding( ProtocolId.JT808, "LNVIN000000AUTH01", "123456789012", "123456789012345", "")); @@ -177,7 +177,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser()))), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), new InMemoryVehicleIdentityService(), new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -203,7 +203,7 @@ class Jt808ChannelHandlerTest { @Test void inactiveChannelRemovesSessionStoreEntry() { - InMemorySessionStore sessions = new InMemorySessionStore(); + TestSessionStore sessions = new TestSessionStore(); DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink())); AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish); Dispatcher dispatcher = new Dispatcher( @@ -253,7 +253,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of(new LocationBodyParser()))), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), identity, new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -315,7 +315,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser()))), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), identity, new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -364,7 +364,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser(), new LocationBodyParser()))), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), identity, new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -422,7 +422,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of(new LocationBodyParser()))), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), identity, new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -465,7 +465,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser()))), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), identity, new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -511,7 +511,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of())), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), new InMemoryVehicleIdentityService(), new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -562,7 +562,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of())), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), new InMemoryVehicleIdentityService(), new Jt808ChannelRegistry(), new Jt808PendingRequests()); @@ -614,7 +614,7 @@ class Jt808ChannelHandlerTest { Jt808ChannelHandler handler = new Jt808ChannelHandler( new Jt808MessageDecoder(new BodyParserRegistry(List.of(new LocationBodyParser()))), dispatcher, - new InMemorySessionStore(), + new TestSessionStore(), new InMemoryVehicleIdentityService(), failingIdentity, new Jt808ChannelRegistry(),