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

refactor: remove production in-memory session store

Browse Source
This commit is contained in:
lingniu
2026-07-01 09:15:28 +08:00
parent 42e8742b10
commit ce3f2ffb90
8 changed files with 101 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java
View File

@@ -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",

4
modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java
View File

@@ -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",

4
modules/core/session-core/pom.xml
View File

@@ -28,10 +28,6 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>

84
modules/core/session-core/src/main/java/com/lingniu/ingest/session/InMemorySessionStore.java
View File

@@ -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 查询。
*
* <p>失活清理：基于 {@link Caffeine} 的 {@code expireAfterAccess 30 分钟}。
* 生产自动配置拒绝 memory 模式，应使用 Redis 会话索引。
*/
public final class InMemorySessionStore implements SessionStore {
private final Cache<String, DeviceSession> byId;
private final ConcurrentMap<String, String> vinToId = new ConcurrentHashMap<>();
private final ConcurrentMap<String, String> 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<DeviceSession> findBySessionId(String sessionId) {
return Optional.ofNullable(byId.getIfPresent(sessionId));
}
@Override
public Optional<DeviceSession> findByVin(String vin) {
String sid = vinToId.get(vin);
return sid == null ? Optional.empty() : findBySessionId(sid);
}
@Override
public Optional<DeviceSession> findByPhone(String phone) {
String sid = phoneToId.get(phone);
return sid == null ? Optional.empty() : findBySessionId(sid);
}
@Override
public synchronized Optional<DeviceSession> update(String sessionId, UnaryOperator<DeviceSession> 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();
}
}

9
modules/core/session-core/src/test/java/com/lingniu/ingest/session/config/SessionCoreAutoConfigurationTest.java
View File

@@ -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

74
modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/TestSessionStore.java Normal file
View File

@@ -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<String, DeviceSession> byId = new ConcurrentHashMap<>();
private final ConcurrentMap<String, String> vinToId = new ConcurrentHashMap<>();
private final ConcurrentMap<String, String> 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<DeviceSession> findBySessionId(String sessionId) {
return Optional.ofNullable(byId.get(sessionId));
}
@Override
public Optional<DeviceSession> findByVin(String vin) {
String sessionId = vinToId.get(vin);
return sessionId == null ? Optional.empty() : findBySessionId(sessionId);
}
@Override
public Optional<DeviceSession> findByPhone(String phone) {
String sessionId = phoneToId.get(phone);
return sessionId == null ? Optional.empty() : findBySessionId(sessionId);
}
@Override
public synchronized Optional<DeviceSession> update(String sessionId, UnaryOperator<DeviceSession> 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();
}
}

4
modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/downlink/Jt808CommandDispatcherTest.java
View File

@@ -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();

28
modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java
View File

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