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(),