wx44wx/live-call-asr-monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add call session store with close API and TTL cleanup.

Browse Source 
Support immediate in-memory release via POST /api/v1/calls/{callId}/close while keeping scheduled TTL as a fallback for unclosed sessions.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Xin Wang
2026-07-15 21:43:58 +08:00
parent 31845f7796
commit ed703d6f39
12 changed files with 567 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
asr_event_monitor_confirmed_technical_solution.md
View File

@@ -51,6 +51,7 @@
- 保存成功后重建 AC 自动机并立即生效
- 提供单页面配置与测试
- JVM 内存保存通话状态
- 提供通话主动关闭接口
- TTL 自动清理过期会话
- 单实例部署
@@ -64,7 +65,6 @@
- Dify 实时确认
- WebSocket 或 SSE
- 通话状态查询接口
- 通话关闭接口
- 独立 `eventId`
- 多实例部署
- 复杂草稿、审核和发布流程
@@ -75,7 +75,7 @@
### 3.1 实时检测接口
只保留一个实时业务接口：
实时检测业务接口：
```http
POST /api/v1/asr-events
@@ -104,18 +104,21 @@ GET /api/v1/calls/{callId}/state
- 本次新增提醒：`newAlerts`
- 当前通话累计结果：`currentResults`
### 3.3 不提供通话关闭接口
### 3.3 提供通话关闭接口
不提供：
通话结束时，前端应主动关闭会话，立即释放内存中的通话状态：
```http
POST /api/v1/calls/{callId}/close
```
同时保留 TTL 定时清理，作为兜底机制，处理前端未调用 close、进程异常退出等遗漏场景。
约束条件：
- 每通电话的 `callId` 必须唯一
- 通话状态通过 TTL 自动清理
- 正常结束路径优先调用 close
- 未关闭或异常遗留的会话由 TTL 自动清理
### 3.4 保留每通电话独立锁
@@ -261,15 +264,18 @@ demo/
│ │ │
│ │ ├── api/
│ │ │ ├── AsrEventController.java
│ │ │ ├── CallController.java
│ │ │ └── RuleAdminController.java
│ │ │
│ │ ├── application/
│ │ │ ├── AsrEventMonitorService.java
│ │ │ ├── CallSessionService.java
│ │ │ └── RuleManagementService.java
│ │ │
│ │ ├── domain/
│ │ │ ├── AsrFinalEventRequest.java
│ │ │ ├── MonitorResponse.java
│ │ │ ├── CloseCallResponse.java
│ │ │ ├── MatchResult.java
│ │ │ ├── AlertResult.java
│ │ │ ├── RuleDocument.java
@@ -758,9 +764,76 @@ try {
---
## 15. 会话 TTL 清理
## 15. 会话关闭与 TTL 清理
由于不提供 close 接口，会话状态依靠 TTL 清理。
会话释放采用 **主动关闭 + TTL 兜底** 双路径：
```text
通话结束
→ 优先：POST /api/v1/calls/{callId}/close（立即释放）
→ 兜底：TTL 定时任务清理遗漏会话
```
### 15.1 通话关闭接口
```http
POST /api/v1/calls/{callId}/close
```
响应示例：
```json
{
"callId": "call-1001",
"closed": true,
"existed": true
}
```
| 字段 | 说明 |
|---|---|
| `callId` | 被关闭的通话 ID |
| `closed` | 本次调用是否完成关闭处理（接口幂等，恒为 `true`） |
| `existed` | 关闭前会话是否存在；重复关闭或不存在时为 `false` |
处理流程：
```text
获取 callId 独立锁
→ 删除 CallSession
→ 删除对应 ReentrantLock
→ 返回 closed=true
```
语义约定：
- 接口幂等：同一 `callId` 多次 close 不报错
- 通话结束后应尽快调用，避免会话长期占用内存
- 由于 `callId` 必须全局唯一，关闭后不应再复用同一 `callId` 提交 ASR Final
服务层示意：
```java
public CloseCallResponse close(String callId) {
ReentrantLock lock = sessionStore.getLock(callId);
lock.lock();
try {
boolean existed = sessionStore.remove(callId);
sessionStore.removeLock(callId);
return new CloseCallResponse(callId, true, existed);
} finally {
lock.unlock();
}
}
```
### 15.2 TTL 兜底清理
即使提供 close，仍保留 TTL，用于清理以下遗留会话：
- 前端未调用 close
- 页面异常关闭、网络故障导致 close 未送达
- 进程重启前未及时释放的会话（单实例内存方案下重启后本就为空，TTL 主要覆盖运行期遗漏）
```yaml
monitor:
@@ -784,11 +857,13 @@ public void cleanupExpiredSessions() {
}
```
清理时同时删除：
TTL 清理时同样删除：
- `CallSession`
- 对应的 `ReentrantLock`
close 与 TTL 互不冲突：已 close 的会话不存在，TTL 跳过；未 close 的会话到期后由 TTL 删除。
---
## 16. 规则数据模型
@@ -1306,11 +1381,12 @@ java \
8. `EMS`、`ems`、`E M S` 均可命中
9. 相邻 Final 拼接后可命中被切分的品牌
10. 不同通话状态相互隔离
11. TTL 能清理过期会话
12. 保存新规则后 Demo 接口立即使用新版本
13. 保存失败时旧规则继续有效
14. 两个管理页面同时编辑时能够检测版本冲突
15. Actuator 健康检查正常
11. close 能立即释放通话会话，重复 close 幂等
12. TTL 能清理未 close 的过期会话
13. 保存新规则后 Demo 接口立即使用新版本
14. 保存失败时旧规则继续有效
15. 两个管理页面同时编辑时能够检测版本冲突
16. Actuator 健康检查正常
---
@@ -1414,11 +1490,19 @@ Demo 页面立即验证
```text
POST /api/v1/asr-events
POST /api/v1/calls/{callId}/close
GET /api/v1/admin/rules
PUT /api/v1/admin/rules
POST /api/v1/admin/rules/test
```
会话清理：
```text
通话结束 → close 立即释放
遗漏会话 → TTL 定时兜底清理
```
最终采用：
```text

2
src/main/java/com/example/demo/DemoApplication.java
View File

@@ -4,9 +4,11 @@ import com.example.demo.config.MonitorProperties;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication
@EnableConfigurationProperties(MonitorProperties.class)
@EnableScheduling
public class DemoApplication {
public static void main(String[] args) {

25
src/main/java/com/example/demo/api/CallController.java Normal file
View File

@@ -0,0 +1,25 @@
package com.example.demo.api;
import com.example.demo.application.CallSessionService;
import com.example.demo.domain.CloseCallResponse;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/api/v1/calls")
public class CallController {
private final CallSessionService callSessionService;
public CallController(CallSessionService callSessionService) {
this.callSessionService = callSessionService;
}
@PostMapping("/{callId}/close")
public ResponseEntity<CloseCallResponse> close(@PathVariable String callId) {
return ResponseEntity.ok(callSessionService.close(callId));
}
}

31
src/main/java/com/example/demo/application/CallSessionService.java Normal file
View File

@@ -0,0 +1,31 @@
package com.example.demo.application;
import com.example.demo.domain.CloseCallResponse;
import com.example.demo.repository.SessionStore;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.stereotype.Service;
@Service
public class CallSessionService {
private final SessionStore sessionStore;
public CallSessionService(SessionStore sessionStore) {
this.sessionStore = sessionStore;
}
/**
* Idempotently close a call session and release its lock mapping.
*/
public CloseCallResponse close(String callId) {
ReentrantLock lock = sessionStore.getLock(callId);
lock.lock();
try {
boolean existed = sessionStore.remove(callId);
sessionStore.removeLock(callId);
return new CloseCallResponse(callId, true, existed);
} finally {
lock.unlock();
}
}
}

8
src/main/java/com/example/demo/domain/CloseCallResponse.java Normal file
View File

@@ -0,0 +1,8 @@
package com.example.demo.domain;
public record CloseCallResponse(
String callId,
boolean closed,
boolean existed
) {
}

39
src/main/java/com/example/demo/job/SessionCleanupJob.java Normal file
View File

@@ -0,0 +1,39 @@
package com.example.demo.job;
import com.example.demo.config.MonitorProperties;
import com.example.demo.repository.SessionStore;
import java.time.Clock;
import java.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Component
public class SessionCleanupJob {
private static final Logger log = LoggerFactory.getLogger(SessionCleanupJob.class);
private final SessionStore sessionStore;
private final MonitorProperties properties;
private final Clock clock;
public SessionCleanupJob(
SessionStore sessionStore,
MonitorProperties properties,
Clock clock
) {
this.sessionStore = sessionStore;
this.properties = properties;
this.clock = clock;
}
@Scheduled(fixedDelayString = "${monitor.session-cleanup-interval:PT10M}")
public void cleanupExpiredSessions() {
Instant threshold = Instant.now(clock).minus(properties.sessionTtl());
int removed = sessionStore.removeExpired(threshold);
if (removed > 0) {
log.info("Removed {} expired call sessions older than {}", removed, threshold);
}
}
}

88
src/main/java/com/example/demo/repository/InMemorySessionStore.java Normal file
View File

@@ -0,0 +1,88 @@
package com.example.demo.repository;
import com.example.demo.domain.CallSession;
import java.time.Clock;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.stereotype.Component;
@Component
public class InMemorySessionStore implements SessionStore {
private final ConcurrentHashMap<String, CallSession> sessions = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, ReentrantLock> callLocks = new ConcurrentHashMap<>();
private final Clock clock;
public InMemorySessionStore(Clock clock) {
this.clock = clock;
}
@Override
public ReentrantLock getLock(String callId) {
return callLocks.computeIfAbsent(callId, ignored -> new ReentrantLock());
}
@Override
public CallSession getOrCreate(String callId) {
return sessions.computeIfAbsent(callId, id -> new CallSession(id, clock));
}
@Override
public Optional<CallSession> find(String callId) {
return Optional.ofNullable(sessions.get(callId));
}
@Override
public boolean remove(String callId) {
return sessions.remove(callId) != null;
}
@Override
public void removeLock(String callId) {
callLocks.compute(callId, (id, existing) -> {
if (existing == null) {
return null;
}
// Drop only when no session remains and no waiters remain after current holder unlocks.
if (!sessions.containsKey(id)
&& !existing.hasQueuedThreads()
&& (existing.isHeldByCurrentThread() || !existing.isLocked())) {
return null;
}
return existing;
});
}
@Override
public int removeExpired(Instant threshold) {
int removed = 0;
List<String> callIds = new ArrayList<>(sessions.keySet());
for (String callId : callIds) {
ReentrantLock lock = getLock(callId);
lock.lock();
try {
CallSession session = sessions.get(callId);
if (session != null && session.lastAccessed().isBefore(threshold)) {
if (sessions.remove(callId, session)) {
removed++;
}
}
} finally {
lock.unlock();
}
removeLock(callId);
}
return removed;
}
@Override
public int size() {
return sessions.size();
}
}

38
src/main/java/com/example/demo/repository/SessionStore.java Normal file
View File

@@ -0,0 +1,38 @@
package com.example.demo.repository;
import com.example.demo.domain.CallSession;
import java.time.Instant;
import java.util.Optional;
import java.util.concurrent.locks.ReentrantLock;
public interface SessionStore {
ReentrantLock getLock(String callId);
CallSession getOrCreate(String callId);
Optional<CallSession> find(String callId);
/**
* Remove the session for callId if present.
*
* @return true if a session existed and was removed
*/
boolean remove(String callId);
/**
* Remove the per-call lock mapping when idle and no session remains.
* Safe to call while holding that same lock instance.
*/
void removeLock(String callId);
/**
* Remove sessions whose lastAccessed is strictly before threshold,
* along with their per-call locks.
*
* @return number of sessions removed
*/
int removeExpired(Instant threshold);
int size();
}

40
src/test/java/com/example/demo/api/CallControllerTest.java Normal file
View File

@@ -0,0 +1,40 @@
package com.example.demo.api;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
import com.example.demo.repository.SessionStore;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import org.springframework.web.context.WebApplicationContext;
@SpringBootTest
class CallControllerTest {
@Autowired
private WebApplicationContext context;
@Autowired
private SessionStore sessionStore;
@Test
void closeEndpointIsIdempotent() throws Exception {
MockMvc mockMvc = MockMvcBuilders.webAppContextSetup(context).build();
sessionStore.getOrCreate("call-1001");
mockMvc.perform(post("/api/v1/calls/call-1001/close"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.callId").value("call-1001"))
.andExpect(jsonPath("$.closed").value(true))
.andExpect(jsonPath("$.existed").value(true));
mockMvc.perform(post("/api/v1/calls/call-1001/close"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.closed").value(true))
.andExpect(jsonPath("$.existed").value(false));
}
}

47
src/test/java/com/example/demo/application/CallSessionServiceTest.java Normal file
View File

@@ -0,0 +1,47 @@
package com.example.demo.application;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import com.example.demo.domain.CloseCallResponse;
import com.example.demo.repository.InMemorySessionStore;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
class CallSessionServiceTest {
private InMemorySessionStore store;
private CallSessionService service;
@BeforeEach
void setUp() {
store = new InMemorySessionStore(Clock.fixed(Instant.parse("2026-07-15T00:00:00Z"), ZoneOffset.UTC));
service = new CallSessionService(store);
}
@Test
void closeReleasesExistingSessionIdempotently() {
store.getOrCreate("call-1001");
CloseCallResponse first = service.close("call-1001");
assertEquals("call-1001", first.callId());
assertTrue(first.closed());
assertTrue(first.existed());
assertEquals(0, store.size());
CloseCallResponse second = service.close("call-1001");
assertTrue(second.closed());
assertFalse(second.existed());
}
@Test
void closeUnknownCallStillSucceeds() {
CloseCallResponse response = service.close("missing");
assertTrue(response.closed());
assertFalse(response.existed());
}
}

39
src/test/java/com/example/demo/job/SessionCleanupJobTest.java Normal file
View File

@@ -0,0 +1,39 @@
package com.example.demo.job;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.example.demo.config.MonitorProperties;
import com.example.demo.repository.SessionStore;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
class SessionCleanupJobTest {
@Test
void cleanupUsesTtlThreshold() {
SessionStore sessionStore = mock(SessionStore.class);
MonitorProperties properties = new MonitorProperties(
2,
500,
Duration.ofHours(2),
Duration.ofMinutes(10)
);
Instant fixed = Instant.parse("2026-07-15T12:00:00Z");
Clock clock = Clock.fixed(fixed, ZoneOffset.UTC);
when(sessionStore.removeExpired(fixed.minus(Duration.ofHours(2)))).thenReturn(3);
SessionCleanupJob job = new SessionCleanupJob(sessionStore, properties, clock);
job.cleanupExpiredSessions();
ArgumentCaptor<Instant> captor = ArgumentCaptor.forClass(Instant.class);
verify(sessionStore).removeExpired(captor.capture());
assertEquals(fixed.minus(Duration.ofHours(2)), captor.getValue());
}
}

113
src/test/java/com/example/demo/repository/InMemorySessionStoreTest.java Normal file
View File

@@ -0,0 +1,113 @@
package com.example.demo.repository;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import com.example.demo.domain.CallSession;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
class InMemorySessionStoreTest {
private AtomicReference<Instant> now;
private InMemorySessionStore store;
@BeforeEach
void setUp() {
now = new AtomicReference<>(Instant.parse("2026-07-15T00:00:00Z"));
Clock clock = new Clock() {
@Override
public ZoneOffset getZone() {
return ZoneOffset.UTC;
}
@Override
public Clock withZone(java.time.ZoneId zone) {
return this;
}
@Override
public Instant instant() {
return now.get();
}
};
store = new InMemorySessionStore(clock);
}
@Test
void getOrCreateReturnsSameSessionPerCallId() {
CallSession a = store.getOrCreate("call-1");
CallSession b = store.getOrCreate("call-1");
CallSession c = store.getOrCreate("call-2");
assertSame(a, b);
assertNotSame(a, c);
assertEquals(2, store.size());
}
@Test
void getLockReturnsSameLockPerCallId() {
ReentrantLock a = store.getLock("call-1");
ReentrantLock b = store.getLock("call-1");
assertSame(a, b);
assertNotSame(a, store.getLock("call-2"));
}
@Test
void removeExpiredDeletesSessionAndAllowsNewLock() {
CallSession session = store.getOrCreate("call-1");
ReentrantLock oldLock = store.getLock("call-1");
session.touch(Clock.fixed(now.get(), ZoneOffset.UTC));
now.set(now.get().plus(Duration.ofHours(3)));
Instant threshold = now.get().minus(Duration.ofHours(2));
assertEquals(1, store.removeExpired(threshold));
assertEquals(0, store.size());
assertTrue(store.find("call-1").isEmpty());
ReentrantLock newLock = store.getLock("call-1");
assertNotSame(oldLock, newLock);
}
@Test
void removeDeletesSessionImmediately() {
store.getOrCreate("call-1");
ReentrantLock oldLock = store.getLock("call-1");
assertTrue(store.remove("call-1"));
store.removeLock("call-1");
assertEquals(0, store.size());
assertFalse(store.remove("call-1"));
assertNotSame(oldLock, store.getLock("call-1"));
}
@Test
void removeExpiredKeepsFreshSessions() {
store.getOrCreate("fresh");
now.set(now.get().plus(Duration.ofMinutes(30)));
Instant threshold = now.get().minus(Duration.ofHours(2));
assertEquals(0, store.removeExpired(threshold));
assertEquals(1, store.size());
}
@Test
void differentCallsAreIsolated() {
store.getOrCreate("call-a").appendCitizenFinal(1, "顺丰", 2);
store.getOrCreate("call-b").appendCitizenFinal(1, "拼多多", 2);
assertEquals("顺丰", store.find("call-a").orElseThrow().buildMatchText());
assertEquals("拼多多", store.find("call-b").orElseThrow().buildMatchText());
}
}