package com.zy.asrs.service.impl;
|
|
import com.alibaba.fastjson.JSON;
|
import com.zy.asrs.domain.replay.DeviceReplayChunkMeta;
|
import com.zy.asrs.domain.replay.DeviceReplayManifest;
|
import com.zy.asrs.domain.replay.DeviceReplayState;
|
import com.zy.asrs.domain.replay.DeviceReplayStreamKey;
|
import com.zy.asrs.domain.replay.ReplayChunk;
|
import com.zy.asrs.domain.replay.ReplayFrameSummary;
|
import com.zy.asrs.domain.replay.ReplaySeekResult;
|
import com.zy.asrs.domain.replay.ReplaySessionContext;
|
import com.zy.asrs.entity.DeviceDataLog;
|
import com.zy.asrs.service.DeviceLogReplayManifestService;
|
import com.zy.asrs.service.DeviceLogReplayNormalizer;
|
import com.zy.asrs.service.DeviceLogReplayService;
|
import jakarta.annotation.PreDestroy;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.stereotype.Service;
|
|
import java.io.BufferedReader;
|
import java.nio.charset.StandardCharsets;
|
import java.nio.file.Files;
|
import java.nio.file.Path;
|
import java.nio.file.Paths;
|
import java.time.LocalDate;
|
import java.time.ZoneId;
|
import java.time.format.DateTimeFormatter;
|
import java.time.format.DateTimeParseException;
|
import java.util.ArrayList;
|
import java.util.Collections;
|
import java.util.Comparator;
|
import java.util.LinkedHashSet;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.UUID;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.Executors;
|
|
@Slf4j
|
@Service
|
public class DeviceLogReplayServiceImpl implements DeviceLogReplayService {
|
|
private static final long CURRENT_DAY_MANIFEST_REFRESH_INTERVAL_MS = 2_000L;
|
private static final long DEFAULT_REPLAY_WINDOW_HALF_MS = 15L * 60L * 1000L;
|
private static final long DAY_END_OFFSET_MS = 24L * 60L * 60L * 1000L - 1L;
|
private static final String REPLAY_WINDOW_MISSING = "REPLAY_WINDOW_MISSING";
|
private static final String REPLAY_WINDOW_INVALID = "REPLAY_WINDOW_INVALID";
|
private static final String REPLAY_WINDOW_CROSS_DAY = "REPLAY_WINDOW_CROSS_DAY";
|
private static final String REPLAY_WINDOW_TOO_LARGE = "REPLAY_WINDOW_TOO_LARGE";
|
private static final String REPLAY_SEEK_OUT_OF_WINDOW = "REPLAY_SEEK_OUT_OF_WINDOW";
|
|
private final DeviceLogReplayManifestService manifestService;
|
private final DeviceLogReplayNormalizer replayNormalizer;
|
|
private final ConcurrentHashMap<String, ReplaySessionContext> sessions = new ConcurrentHashMap<>();
|
private final ExecutorService replayWarmupExecutor = Executors.newSingleThreadExecutor(runnable -> {
|
Thread thread = new Thread(runnable, "replay-tail-checkpoint-warmup");
|
thread.setDaemon(true);
|
return thread;
|
});
|
|
@Value("${deviceLogStorage.type}")
|
private String storageType;
|
@Value("${deviceLogStorage.loggingPath}")
|
private String loggingPath;
|
@Value("${deviceReplay.chunkSampleSize:400}")
|
private Integer chunkSampleSize;
|
@Value("${deviceReplay.maxChunkFrames:32}")
|
private Integer maxChunkFrames;
|
@Value("${deviceReplay.timelineBucketSeconds:20}")
|
private Integer timelineBucketSeconds;
|
@Value("${deviceReplay.maxWindowMinutes:240}")
|
private Integer maxWindowMinutes;
|
@Value("${deviceReplay.maxConcurrentSessions:4}")
|
private Integer maxConcurrentSessions;
|
@Value("${deviceReplay.maxPrefetchChunks:3}")
|
private Integer maxPrefetchChunks;
|
@Value("${deviceReplay.maxChunkCacheEntries:128}")
|
private Integer maxChunkCacheEntries;
|
@Value("${deviceReplay.sessionTtlSeconds:1800}")
|
private Integer sessionTtlSeconds;
|
|
public DeviceLogReplayServiceImpl(DeviceLogReplayManifestService manifestService,
|
DeviceLogReplayNormalizer replayNormalizer) {
|
this.manifestService = manifestService;
|
this.replayNormalizer = replayNormalizer;
|
}
|
|
@Override
|
public Map<String, Object> createSession(String day,
|
String type,
|
Integer deviceNo,
|
Integer stationId,
|
Long targetTimestamp,
|
Long windowStartTimeMs,
|
Long windowEndTimeMs) {
|
ensureFileMode();
|
ReplayWindow replayWindow = resolveReplayWindow(day, targetTimestamp, windowStartTimeMs, windowEndTimeMs);
|
cleanupExpiredSessions();
|
if (sessions.size() >= safePositive(maxConcurrentSessions, 4)) {
|
throw new IllegalStateException("历史回放会话已满，请先关闭未使用的回放页面");
|
}
|
boolean currentReplayDay = isCurrentReplayDay(day);
|
if (type != null && !type.isBlank() && deviceNo != null) {
|
warmupEntryManifest(day, type, deviceNo, stationId);
|
}
|
List<DeviceReplayManifest> manifests = loadSessionManifests(day, currentReplayDay, true);
|
if (manifests.isEmpty()) {
|
throw new IllegalArgumentException("当天没有可回放清单，请确认该日期日志已生成 replay manifest");
|
}
|
TimelineChunkSnapshotData timelineSnapshot = buildTimelineChunkSnapshot(manifests, replayWindow.startTimeMs, replayWindow.endTimeMs);
|
ReplaySessionContext session = new ReplaySessionContext();
|
session.setSessionId(UUID.randomUUID().toString().replace("-", ""));
|
session.setDay(day);
|
session.setCreatedAtMs(System.currentTimeMillis());
|
session.setLastAccessAtMs(session.getCreatedAtMs());
|
session.setWindowStartTimeMs(replayWindow.startTimeMs);
|
session.setWindowEndTimeMs(replayWindow.endTimeMs);
|
session.setStartTimeMs(replayWindow.startTimeMs);
|
session.setEndTimeMs(replayWindow.endTimeMs);
|
session.setTimelineChunks(timelineSnapshot.timelineChunks);
|
session.setChunkStreamIdsByChunk(timelineSnapshot.chunkStreamIdsByChunk);
|
session.setLastManifestRefreshAtMs(System.currentTimeMillis());
|
session.setManifestSnapshotKey(buildManifestSnapshotKey(manifests));
|
session.setManifestByStream(buildManifestSnapshot(manifests));
|
sessions.put(session.getSessionId(), session);
|
if (currentReplayDay) {
|
manifestService.scheduleDayManifestRefresh(day);
|
scheduleTailCheckpointWarmup(session.getSessionId());
|
}
|
|
Map<String, Object> data = new LinkedHashMap<>();
|
data.put("sessionId", session.getSessionId());
|
data.put("day", day);
|
data.put("startTimeMs", session.getStartTimeMs());
|
data.put("endTimeMs", session.getEndTimeMs());
|
data.put("windowStartTimeMs", session.getWindowStartTimeMs());
|
data.put("windowEndTimeMs", session.getWindowEndTimeMs());
|
data.put("chunkCount", session.getTimelineChunks().size());
|
data.put("streamCount", session.getManifestByStream().size());
|
data.put("entryType", type);
|
data.put("entryDeviceNo", deviceNo);
|
data.put("entryStationId", stationId);
|
data.put("maxPrefetchChunks", safePositive(maxPrefetchChunks, 3));
|
return data;
|
}
|
|
@Override
|
public Map<String, Object> loadTimeline(String sessionId) {
|
ReplaySessionContext session = requireSession(sessionId);
|
Map<String, Object> data = new LinkedHashMap<>();
|
data.put("sessionId", session.getSessionId());
|
data.put("day", session.getDay());
|
data.put("startTimeMs", session.getStartTimeMs());
|
data.put("endTimeMs", session.getEndTimeMs());
|
data.put("windowStartTimeMs", session.getWindowStartTimeMs());
|
data.put("windowEndTimeMs", session.getWindowEndTimeMs());
|
data.put("chunks", session.getTimelineChunks());
|
data.put("maxPrefetchChunks", safePositive(maxPrefetchChunks, 3));
|
return data;
|
}
|
|
@Override
|
public Map<String, Object> loadTimelineSummary(String sessionId, Integer bucketCount) {
|
ReplaySessionContext session = requireSession(sessionId);
|
int effectiveBucketCount = resolveSummaryBucketCount(session, bucketCount);
|
synchronized (session.getTimelineSummaryCache()) {
|
Map<String, Object> cachedSummary = session.getTimelineSummaryCache().get(effectiveBucketCount);
|
if (cachedSummary != null) {
|
return cachedSummary;
|
}
|
}
|
TimelineSummarySnapshot summarySnapshot = buildTimelineSummarySnapshot(session);
|
Map<String, Object> summary = buildTimelineSummary(summarySnapshot, effectiveBucketCount);
|
synchronized (session.getTimelineSummaryCache()) {
|
Map<String, Object> cachedSummary = session.getTimelineSummaryCache().get(effectiveBucketCount);
|
if (cachedSummary != null) {
|
return cachedSummary;
|
}
|
session.getTimelineSummaryCache().put(effectiveBucketCount, summary);
|
return summary;
|
}
|
}
|
|
private TimelineSummarySnapshot buildTimelineSummarySnapshot(ReplaySessionContext session) {
|
TimelineSummarySnapshot snapshot = new TimelineSummarySnapshot();
|
synchronized (session) {
|
snapshot.sessionId = session.getSessionId();
|
snapshot.day = session.getDay();
|
snapshot.startTimeMs = safeLong(session.getStartTimeMs());
|
snapshot.endTimeMs = safeLong(session.getEndTimeMs());
|
snapshot.windowStartTimeMs = safeLong(session.getWindowStartTimeMs());
|
snapshot.windowEndTimeMs = safeLong(session.getWindowEndTimeMs());
|
snapshot.timelineChunks = new ArrayList<>(session.getTimelineChunks());
|
snapshot.manifests = new ArrayList<>(session.getManifestByStream().values());
|
}
|
return snapshot;
|
}
|
|
private Map<String, Object> buildTimelineSummary(TimelineSummarySnapshot snapshot, int bucketCount) {
|
Map<String, Object> summary = new LinkedHashMap<>();
|
List<Map<String, Object>> buckets = new ArrayList<>();
|
if (bucketCount <= 0 || snapshot.timelineChunks.isEmpty()) {
|
summary.put("sessionId", snapshot.sessionId);
|
summary.put("day", snapshot.day);
|
summary.put("bucketCount", 0);
|
summary.put("startTimeMs", snapshot.startTimeMs);
|
summary.put("endTimeMs", snapshot.endTimeMs);
|
summary.put("buckets", buckets);
|
return summary;
|
}
|
|
long totalDurationMs = Math.max(1L, snapshot.endTimeMs - snapshot.startTimeMs + 1L);
|
long bucketDurationMs = Math.max(1L, (long) Math.ceil(totalDurationMs / (double) bucketCount));
|
for (int bucketIndex = 0; bucketIndex < bucketCount; bucketIndex++) {
|
long bucketStartTime = snapshot.startTimeMs + bucketIndex * bucketDurationMs;
|
if (bucketStartTime > snapshot.endTimeMs) {
|
bucketStartTime = snapshot.endTimeMs;
|
}
|
long bucketEndTime = Math.min(snapshot.endTimeMs, bucketStartTime + bucketDurationMs - 1L);
|
int startChunkIndex = resolveChunkIndex(snapshot.timelineChunks, bucketStartTime, 0L);
|
int endChunkIndex = resolveChunkIndex(snapshot.timelineChunks, bucketEndTime, 0L);
|
if (endChunkIndex < startChunkIndex) {
|
endChunkIndex = startChunkIndex;
|
}
|
buckets.add(buildSummaryBucket(snapshot, bucketIndex, bucketStartTime, bucketEndTime, startChunkIndex, endChunkIndex));
|
}
|
summary.put("sessionId", snapshot.sessionId);
|
summary.put("day", snapshot.day);
|
summary.put("bucketCount", bucketCount);
|
summary.put("startTimeMs", snapshot.startTimeMs);
|
summary.put("endTimeMs", snapshot.endTimeMs);
|
summary.put("buckets", buckets);
|
return summary;
|
}
|
|
private Map<String, Object> buildSummaryBucket(TimelineSummarySnapshot snapshot,
|
int bucketIndex,
|
long bucketStartTime,
|
long bucketEndTime,
|
int startChunkIndex,
|
int endChunkIndex) {
|
Map<String, Object> bucket = new LinkedHashMap<>();
|
bucket.put("bucketIndex", bucketIndex);
|
bucket.put("startTimeMs", clampTimestampToWindow(snapshot, bucketStartTime));
|
bucket.put("endTimeMs", clampTimestampToWindow(snapshot, bucketEndTime));
|
|
int firstDataChunkIndex = -1;
|
int anchorChunkIndex = -1;
|
int anchorFrameIndex = -1;
|
int firstAbnormalChunkIndex = -1;
|
int firstAbnormalFrameIndex = -1;
|
String topEventType = "";
|
int bucketEstimatedSamples = 0;
|
|
for (int chunkIndex = startChunkIndex; chunkIndex <= endChunkIndex; chunkIndex++) {
|
Map<String, Object> chunkMeta = resolveTimelineChunk(snapshot.timelineChunks, chunkIndex);
|
int estimatedSamples = intValue(chunkMeta.get("estimatedSamples"));
|
bucketEstimatedSamples += Math.max(0, estimatedSamples);
|
if (firstDataChunkIndex >= 0 || estimatedSamples <= 0) {
|
continue;
|
}
|
firstDataChunkIndex = chunkIndex;
|
}
|
|
boolean hasReplayData = bucketEstimatedSamples > 0 && firstDataChunkIndex >= 0;
|
int activityScore = hasReplayData ? 100 : 0;
|
int abnormalCount = 0;
|
int errorCount = 0;
|
int blockCount = 0;
|
boolean hasCrnData = false;
|
boolean hasStationData = false;
|
boolean hasRgvData = false;
|
|
if (hasReplayData) {
|
long visibleBucketStartTime = clampTimestampToWindow(snapshot, bucketStartTime);
|
long visibleBucketEndTime = clampTimestampToWindow(snapshot, bucketEndTime);
|
for (DeviceReplayManifest manifest : snapshot.manifests) {
|
if (manifest == null) {
|
continue;
|
}
|
long manifestStartTime = safeLong(manifest.getFirstSampleTimeMs());
|
long manifestEndTime = safeLong(manifest.getLastSampleTimeMs());
|
if (manifestEndTime < visibleBucketStartTime || manifestStartTime > visibleBucketEndTime) {
|
continue;
|
}
|
if ("Devp".equalsIgnoreCase(manifest.getType())) {
|
hasStationData = true;
|
} else if ("Rgv".equalsIgnoreCase(manifest.getType())) {
|
hasRgvData = true;
|
} else if ("Crn".equalsIgnoreCase(manifest.getType()) || "DualCrn".equalsIgnoreCase(manifest.getType())) {
|
hasCrnData = true;
|
}
|
}
|
}
|
|
if (hasReplayData && anchorChunkIndex < 0) {
|
anchorChunkIndex = firstDataChunkIndex;
|
anchorFrameIndex = 0;
|
}
|
|
bucket.put("activityScore", activityScore);
|
bucket.put("abnormalCount", abnormalCount);
|
bucket.put("errorCount", errorCount);
|
bucket.put("blockCount", blockCount);
|
bucket.put("hasReplayData", hasReplayData);
|
bucket.put("hasCrnData", hasCrnData);
|
bucket.put("hasStationData", hasStationData);
|
bucket.put("hasRgvData", hasRgvData);
|
bucket.put("anchorChunkIndex", anchorChunkIndex);
|
bucket.put("anchorFrameIndex", anchorFrameIndex);
|
bucket.put("firstAbnormalChunkIndex", firstAbnormalChunkIndex);
|
bucket.put("firstAbnormalFrameIndex", firstAbnormalFrameIndex);
|
bucket.put("topEventType", topEventType);
|
return bucket;
|
}
|
|
@Override
|
public ReplayChunk loadChunk(String sessionId, Integer chunkIndex) {
|
ReplaySessionContext session = requireSession(sessionId);
|
return projectChunkForClient(session, loadFullChunk(session, chunkIndex));
|
}
|
|
private ReplayChunk loadFullChunk(ReplaySessionContext session, Integer chunkIndex) {
|
ReplayChunk cachedChunk = getCachedChunk(session, chunkIndex, true);
|
if (cachedChunk != null) {
|
return cachedChunk;
|
}
|
String manifestSnapshotKey = session.getManifestSnapshotKey();
|
ReplayChunk chunk = buildChunk(session, chunkIndex, true);
|
if (isSameManifestSnapshot(session, manifestSnapshotKey)) {
|
cacheChunk(session, chunkIndex, chunk, true);
|
} else {
|
chunk = buildChunk(session, chunkIndex, true);
|
}
|
return chunk;
|
}
|
|
@Override
|
public ReplayChunk loadChunk(String sessionId, Integer chunkIndex, boolean fullFrames) {
|
ReplaySessionContext session = requireSession(sessionId);
|
ReplayChunk cachedChunk = getCachedChunk(session, chunkIndex, fullFrames);
|
if (cachedChunk != null) {
|
return projectChunkForClient(session, cachedChunk);
|
}
|
String manifestSnapshotKey = session.getManifestSnapshotKey();
|
ReplayChunk chunk = fullFrames
|
? buildChunk(session, chunkIndex, true)
|
: buildCompactChunk(session, chunkIndex);
|
if (isSameManifestSnapshot(session, manifestSnapshotKey)) {
|
cacheChunk(session, chunkIndex, chunk, fullFrames);
|
} else {
|
chunk = fullFrames
|
? buildChunk(session, chunkIndex, true)
|
: buildCompactChunk(session, chunkIndex);
|
}
|
return projectChunkForClient(session, chunk);
|
}
|
|
private ReplayChunk buildCompactChunk(ReplaySessionContext session, Integer chunkIndex) {
|
Map<String, Object> chunkMeta = resolveTimelineChunk(session, chunkIndex);
|
long startTime = numberValue(chunkMeta.get("startTimeMs"));
|
long endTime = numberValue(chunkMeta.get("endTimeMs"));
|
long sampleSeq = numberValue(chunkMeta.get("lastSampleSeq"));
|
|
ReplayChunk chunk = new ReplayChunk();
|
chunk.setChunkIndex(chunkIndex);
|
chunk.setStartTimeMs(startTime);
|
chunk.setEndTimeMs(endTime);
|
|
Map<String, Object> summary = new LinkedHashMap<>();
|
summary.put("stationCount", 0);
|
summary.put("crnCount", 0);
|
summary.put("dualCrnCount", 0);
|
summary.put("rgvCount", 0);
|
summary.put("abnormalCount", 0);
|
summary.put("errorCount", 0);
|
summary.put("blockCount", 0);
|
summary.put("estimatedSamples", intValue(chunkMeta.get("estimatedSamples")));
|
|
Map<String, Object> frame = new LinkedHashMap<>();
|
frame.put("timestamp", endTime > 0 ? endTime : startTime);
|
frame.put("sampleSeq", sampleSeq > 0 ? sampleSeq : 0L);
|
frame.put("abnormalList", new ArrayList<>());
|
frame.put("summary", summary);
|
appendFrame(chunk, frame);
|
chunk.setFrameCount(chunk.getFrames().size());
|
return chunk;
|
}
|
|
private ReplayChunk buildChunk(ReplaySessionContext session, Integer chunkIndex, boolean fullFrames) {
|
String manifestSnapshotKey = session.getManifestSnapshotKey();
|
Map<String, Object> chunkMeta = resolveTimelineChunk(session, chunkIndex);
|
long startTime = numberValue(chunkMeta.get("startTimeMs"));
|
long endTime = numberValue(chunkMeta.get("endTimeMs"));
|
|
Map<String, Object> currentStateByStream = new LinkedHashMap<>();
|
List<DeviceReplayState> eventList = new ArrayList<>();
|
ChunkCheckpointSeed checkpointSeed = resolveCheckpointSeed(session, chunkIndex);
|
boolean enableBaselineRecovery = fullFrames || checkpointSeed != null;
|
if (checkpointSeed != null && checkpointSeed.stateByStream != null) {
|
currentStateByStream.putAll(checkpointSeed.stateByStream);
|
}
|
long scanStartExclusiveTimeMs = checkpointSeed == null ? 0L : checkpointSeed.scanStartExclusiveTimeMs;
|
boolean allowBaselineLookup = fullFrames && checkpointSeed == null;
|
|
for (DeviceReplayManifest manifest : resolveChunkManifests(session, chunkIndex)) {
|
StreamReplayScanResult scanResult = scanManifestRange(
|
manifest,
|
startTime,
|
endTime,
|
allowBaselineLookup,
|
scanStartExclusiveTimeMs);
|
if (enableBaselineRecovery && scanResult.baselineState != null) {
|
currentStateByStream.put(manifest.getStreamId(), scanResult.baselineState);
|
}
|
eventList.addAll(scanResult.eventList);
|
}
|
eventList.sort(Comparator
|
.comparing(DeviceReplayState::getTimestamp, Comparator.nullsLast(Long::compareTo))
|
.thenComparing(DeviceReplayState::getSampleSeq, Comparator.nullsLast(Long::compareTo))
|
.thenComparing(DeviceReplayState::getType, Comparator.nullsLast(String::compareTo))
|
.thenComparing(DeviceReplayState::getDeviceNo, Comparator.nullsLast(Integer::compareTo))
|
.thenComparing(DeviceReplayState::getStationId, Comparator.nullsLast(Integer::compareTo)));
|
|
ReplayChunk chunk = new ReplayChunk();
|
chunk.setChunkIndex(chunkIndex);
|
chunk.setStartTimeMs(startTime);
|
chunk.setEndTimeMs(endTime);
|
|
if (fullFrames && !currentStateByStream.isEmpty()) {
|
Map<String, Object> baseState = replayNormalizer.buildFrame(startTime, 0L, currentStateByStream);
|
chunk.setBaseState(baseState);
|
appendFrame(chunk, baseState);
|
}
|
|
List<Integer> keptGroupIndexes = fullFrames ? null : resolveCompactGroupIndexes(eventList);
|
int groupIndex = -1;
|
int cursor = 0;
|
while (cursor < eventList.size()) {
|
groupIndex++;
|
DeviceReplayState currentEvent = eventList.get(cursor);
|
Long frameTime = currentEvent.getTimestamp();
|
Long frameSeq = currentEvent.getSampleSeq();
|
while (cursor < eventList.size()) {
|
DeviceReplayState nextEvent = eventList.get(cursor);
|
if (!equalsLong(frameTime, nextEvent.getTimestamp())
|
|| !equalsLong(frameSeq, nextEvent.getSampleSeq())) {
|
break;
|
}
|
String streamId = buildStreamId(session.getDay(), nextEvent.getType(), nextEvent.getDeviceNo(), nextEvent.getStationId());
|
currentStateByStream.put(streamId, nextEvent.getPayload());
|
frameSeq = nextEvent.getSampleSeq();
|
cursor++;
|
}
|
if (fullFrames || keptGroupIndexes.contains(groupIndex)) {
|
Map<String, Object> frame = fullFrames
|
? replayNormalizer.buildFrame(frameTime, frameSeq, currentStateByStream)
|
: replayNormalizer.buildCompactFrame(frameTime, frameSeq, currentStateByStream);
|
appendFrame(chunk, frame);
|
}
|
}
|
|
if (chunk.getFrames().isEmpty()) {
|
Map<String, Object> emptyFrame = fullFrames
|
? replayNormalizer.buildFrame(startTime, 0L, currentStateByStream)
|
: replayNormalizer.buildCompactFrame(startTime, 0L, currentStateByStream);
|
appendFrame(chunk, emptyFrame);
|
}
|
if (fullFrames) {
|
chunk.setFullFrames(new ArrayList<>(chunk.getFrames()));
|
chunk.setFullFrameSummaryList(new ArrayList<>(chunk.getFrameSummaryList()));
|
shrinkChunkFrames(chunk);
|
}
|
chunk.setFrameCount(chunk.getFrames().size());
|
if (fullFrames && !currentStateByStream.isEmpty() && isSameManifestSnapshot(session, manifestSnapshotKey)) {
|
cacheChunkCheckpoint(session, chunkIndex, currentStateByStream);
|
}
|
return chunk;
|
}
|
|
private ReplayChunk projectChunkForClient(ReplaySessionContext session, ReplayChunk sourceChunk) {
|
ReplayChunk chunk = new ReplayChunk();
|
chunk.setChunkIndex(sourceChunk.getChunkIndex());
|
chunk.setStartTimeMs(clampTimestampToWindow(session, safeLong(sourceChunk.getStartTimeMs())));
|
chunk.setEndTimeMs(clampTimestampToWindow(session, safeLong(sourceChunk.getEndTimeMs())));
|
chunk.setBaseState(copyObjectForClient(sourceChunk.getBaseState()));
|
copyVisibleFrames(sourceChunk, chunk);
|
chunk.setFrameCount(chunk.getFrames().size());
|
return chunk;
|
}
|
|
@SuppressWarnings("unchecked")
|
private Map<String, Object> copyObjectForClient(Map<String, Object> sourceMap) {
|
if (sourceMap == null) {
|
return null;
|
}
|
return (Map<String, Object>) copyValueForClient(sourceMap);
|
}
|
|
@SuppressWarnings("unchecked")
|
private Object copyValueForClient(Object sourceValue) {
|
if (sourceValue instanceof Map<?, ?> sourceMap) {
|
Map<String, Object> copiedMap = new LinkedHashMap<>();
|
for (Map.Entry<?, ?> entry : sourceMap.entrySet()) {
|
copiedMap.put(String.valueOf(entry.getKey()), copyValueForClient(entry.getValue()));
|
}
|
return copiedMap;
|
}
|
if (sourceValue instanceof List<?> sourceList) {
|
List<Object> copiedList = new ArrayList<>(sourceList.size());
|
for (Object item : sourceList) {
|
copiedList.add(copyValueForClient(item));
|
}
|
return copiedList;
|
}
|
return sourceValue;
|
}
|
|
private void copyVisibleFrames(ReplayChunk sourceChunk, ReplayChunk targetChunk) {
|
List<Map<String, Object>> sourceFrames = resolveFullFrameList(sourceChunk);
|
List<ReplayFrameSummary> sourceSummaries = resolveFullFrameSummaryList(sourceChunk);
|
long startTime = safeLong(targetChunk.getStartTimeMs());
|
long endTime = safeLong(targetChunk.getEndTimeMs());
|
for (int frameIndex = 0; frameIndex < sourceFrames.size(); frameIndex++) {
|
Map<String, Object> frame = sourceFrames.get(frameIndex);
|
long frameTime = numberValue(frame.get("timestamp"));
|
if (frameTime < startTime || frameTime > endTime) {
|
continue;
|
}
|
targetChunk.getFrames().add(copyObjectForClient(frame));
|
ReplayFrameSummary summary = frameIndex < sourceSummaries.size() ? sourceSummaries.get(frameIndex) : null;
|
targetChunk.getFrameSummaryList().add(copyFrameSummary(summary, targetChunk.getFrameSummaryList().size(), frameTime));
|
}
|
}
|
|
private List<ReplayFrameSummary> resolveFullFrameSummaryList(ReplayChunk replayChunk) {
|
if (replayChunk == null) {
|
return new ArrayList<>();
|
}
|
if (replayChunk.getFullFrameSummaryList() != null && !replayChunk.getFullFrameSummaryList().isEmpty()) {
|
return replayChunk.getFullFrameSummaryList();
|
}
|
return replayChunk.getFrameSummaryList();
|
}
|
|
private ReplayFrameSummary copyFrameSummary(ReplayFrameSummary sourceSummary, int frameIndex, long frameTime) {
|
ReplayFrameSummary summary = new ReplayFrameSummary();
|
summary.setFrameIndex(frameIndex);
|
summary.setTimestamp(frameTime);
|
summary.setAbnormalCount(sourceSummary == null ? 0 : sourceSummary.getAbnormalCount());
|
summary.setErrorCount(sourceSummary == null ? 0 : sourceSummary.getErrorCount());
|
summary.setBlockCount(sourceSummary == null ? 0 : sourceSummary.getBlockCount());
|
summary.setManualCount(sourceSummary == null ? 0 : sourceSummary.getManualCount());
|
return summary;
|
}
|
|
private List<Map<String, Object>> resolveFullFrameList(ReplayChunk replayChunk) {
|
if (replayChunk == null) {
|
return new ArrayList<>();
|
}
|
if (replayChunk.getFullFrames() != null && !replayChunk.getFullFrames().isEmpty()) {
|
return replayChunk.getFullFrames();
|
}
|
return replayChunk.getFrames();
|
}
|
|
private void shrinkChunkFrames(ReplayChunk chunk) {
|
if (chunk == null || chunk.getFrames() == null || chunk.getFrames().isEmpty()) {
|
return;
|
}
|
int frameLimit = Math.max(16, safePositive(maxChunkFrames, 32));
|
if (chunk.getFrames().size() <= frameLimit) {
|
return;
|
}
|
|
LinkedHashSet<Integer> keptIndexes = new LinkedHashSet<>();
|
keptIndexes.add(0);
|
keptIndexes.add(chunk.getFrames().size() - 1);
|
|
for (int index = 0; index < chunk.getFrameSummaryList().size(); index++) {
|
ReplayFrameSummary frameSummary = chunk.getFrameSummaryList().get(index);
|
if (frameSummary != null && safeSummaryActivity(frameSummary) > 0) {
|
keptIndexes.add(index);
|
}
|
}
|
|
int remainingSlots = Math.max(0, frameLimit - keptIndexes.size());
|
if (remainingSlots > 0) {
|
double step = (chunk.getFrames().size() - 1D) / Math.max(1D, remainingSlots);
|
for (int slot = 1; slot <= remainingSlots; slot++) {
|
int sampledIndex = (int) Math.round(slot * step);
|
sampledIndex = Math.max(0, Math.min(chunk.getFrames().size() - 1, sampledIndex));
|
keptIndexes.add(sampledIndex);
|
if (keptIndexes.size() >= frameLimit) {
|
break;
|
}
|
}
|
}
|
|
List<Integer> orderedIndexes = new ArrayList<>(keptIndexes);
|
orderedIndexes.sort(Integer::compareTo);
|
|
List<Map<String, Object>> compactFrames = new ArrayList<>(orderedIndexes.size());
|
List<ReplayFrameSummary> compactSummaryList = new ArrayList<>(orderedIndexes.size());
|
for (int compactIndex = 0; compactIndex < orderedIndexes.size(); compactIndex++) {
|
int originalIndex = orderedIndexes.get(compactIndex);
|
compactFrames.add(chunk.getFrames().get(originalIndex));
|
ReplayFrameSummary originalSummary = originalIndex < chunk.getFrameSummaryList().size()
|
? chunk.getFrameSummaryList().get(originalIndex)
|
: null;
|
if (originalSummary == null) {
|
originalSummary = new ReplayFrameSummary();
|
}
|
ReplayFrameSummary compactSummary = new ReplayFrameSummary();
|
compactSummary.setFrameIndex(compactIndex);
|
compactSummary.setTimestamp(originalSummary.getTimestamp());
|
compactSummary.setAbnormalCount(originalSummary.getAbnormalCount());
|
compactSummary.setErrorCount(originalSummary.getErrorCount());
|
compactSummary.setBlockCount(originalSummary.getBlockCount());
|
compactSummary.setManualCount(originalSummary.getManualCount());
|
compactSummaryList.add(compactSummary);
|
}
|
chunk.setFrames(compactFrames);
|
chunk.setFrameSummaryList(compactSummaryList);
|
}
|
|
private List<Integer> resolveCompactGroupIndexes(List<DeviceReplayState> eventList) {
|
List<Integer> keptIndexes = new ArrayList<>();
|
if (eventList == null || eventList.isEmpty()) {
|
return keptIndexes;
|
}
|
int groupCount = 0;
|
int cursor = 0;
|
while (cursor < eventList.size()) {
|
DeviceReplayState currentEvent = eventList.get(cursor);
|
Long frameTime = currentEvent.getTimestamp();
|
Long frameSeq = currentEvent.getSampleSeq();
|
while (cursor < eventList.size()) {
|
DeviceReplayState nextEvent = eventList.get(cursor);
|
if (!equalsLong(frameTime, nextEvent.getTimestamp())
|
|| !equalsLong(frameSeq, nextEvent.getSampleSeq())) {
|
break;
|
}
|
cursor++;
|
}
|
groupCount++;
|
}
|
int compactFrameLimit = Math.max(8, safePositive(maxChunkFrames, 32));
|
if (groupCount <= compactFrameLimit) {
|
for (int index = 0; index < groupCount; index++) {
|
keptIndexes.add(index);
|
}
|
return keptIndexes;
|
}
|
keptIndexes.add(0);
|
keptIndexes.add(groupCount - 1);
|
int remainingSlots = Math.max(0, compactFrameLimit - keptIndexes.size());
|
double step = (groupCount - 1D) / Math.max(1D, remainingSlots);
|
for (int slot = 1; slot <= remainingSlots; slot++) {
|
int sampledIndex = (int) Math.round(slot * step);
|
sampledIndex = Math.max(0, Math.min(groupCount - 1, sampledIndex));
|
if (!keptIndexes.contains(sampledIndex)) {
|
keptIndexes.add(sampledIndex);
|
}
|
}
|
keptIndexes.sort(Integer::compareTo);
|
return keptIndexes;
|
}
|
|
private int safeSummaryActivity(ReplayFrameSummary frameSummary) {
|
if (frameSummary == null) {
|
return 0;
|
}
|
return intValue(frameSummary.getAbnormalCount())
|
+ intValue(frameSummary.getErrorCount())
|
+ intValue(frameSummary.getBlockCount());
|
}
|
|
@Override
|
public ReplaySeekResult seek(String sessionId, Long timestamp, Long sampleSeq) {
|
ReplaySessionContext session = requireSession(sessionId);
|
long requestTime = timestamp == null || timestamp <= 0 ? session.getStartTimeMs() : timestamp;
|
rejectTimestampOutsideWindow(session, requestTime, "seek 目标时间不在当前回放窗口内");
|
long requestSeq = sampleSeq == null || sampleSeq <= 0 ? 0L : sampleSeq;
|
rejectSampleSeqOutsideWindow(session, requestSeq);
|
int initialChunkIndex = resolveChunkIndex(session, requestTime, requestSeq);
|
if (requestSeq <= 0) {
|
ReplaySeekResult result = new ReplaySeekResult();
|
result.setChunkIndex(initialChunkIndex);
|
result.setFrameIndex(0);
|
result.setRequestedTimestamp(requestTime);
|
result.setResolvedTimestamp(requestTime);
|
result.setRequestedSampleSeq(null);
|
result.setResolvedSampleSeq(null);
|
result.setDriftMs(0L);
|
result.setResolvedFrame(null);
|
return result;
|
}
|
ResolvedSeekFrame resolvedSeekFrame = findNearestSeekFrame(session, sessionId, initialChunkIndex, requestTime, requestSeq);
|
int chunkIndex = resolvedSeekFrame == null ? initialChunkIndex : resolvedSeekFrame.chunkIndex;
|
ReplayChunk chunk = resolvedSeekFrame == null ? loadFullChunk(session, chunkIndex) : resolvedSeekFrame.chunk;
|
long resolvedTime = resolvedSeekFrame == null ? requestTime : numberValue(resolvedSeekFrame.frame.get("timestamp"));
|
long resolvedSeq = resolvedSeekFrame == null ? requestSeq : numberValue(resolvedSeekFrame.frame.get("sampleSeq"));
|
int displayFrameIndex = findNearestDisplayFrameIndex(chunk, resolvedTime);
|
ReplaySeekResult result = new ReplaySeekResult();
|
result.setChunkIndex(chunkIndex);
|
result.setFrameIndex(displayFrameIndex);
|
result.setRequestedTimestamp(requestTime);
|
result.setResolvedTimestamp(resolvedTime);
|
result.setRequestedSampleSeq(requestSeq > 0 ? requestSeq : null);
|
result.setResolvedSampleSeq(resolvedSeq > 0 ? resolvedSeq : null);
|
result.setDriftMs(requestSeq > 0
|
? Math.abs(resolvedSeq - requestSeq)
|
: Math.abs(resolvedTime - requestTime));
|
result.setResolvedFrame(resolvedSeekFrame == null ? null : resolvedSeekFrame.frame);
|
return result;
|
}
|
|
private ResolvedSeekFrame findNearestSeekFrame(ReplaySessionContext session,
|
String sessionId,
|
int initialChunkIndex,
|
long requestTime,
|
long requestSeq) {
|
ResolvedSeekFrame bestMatch = evaluateSeekChunk(session, initialChunkIndex, requestTime, requestSeq);
|
long bestDrift = bestMatch == null ? Long.MAX_VALUE : bestMatch.drift;
|
int maxSearchDistance = Math.min(session.getTimelineChunks().size(), 512);
|
|
for (int distance = 1; distance < maxSearchDistance; distance++) {
|
boolean checkedAny = false;
|
|
int leftChunkIndex = initialChunkIndex - distance;
|
if (leftChunkIndex >= 0) {
|
long lowerBoundDrift = resolveChunkBoundaryDrift(session, leftChunkIndex, requestTime);
|
if (lowerBoundDrift <= bestDrift) {
|
ResolvedSeekFrame leftMatch = evaluateSeekChunk(session, leftChunkIndex, requestTime, requestSeq);
|
bestMatch = chooseBetterSeekFrame(bestMatch, leftMatch);
|
bestDrift = bestMatch == null ? Long.MAX_VALUE : bestMatch.drift;
|
checkedAny = true;
|
}
|
}
|
|
int rightChunkIndex = initialChunkIndex + distance;
|
if (rightChunkIndex < session.getTimelineChunks().size()) {
|
long lowerBoundDrift = resolveChunkBoundaryDrift(session, rightChunkIndex, requestTime);
|
if (lowerBoundDrift <= bestDrift) {
|
ResolvedSeekFrame rightMatch = evaluateSeekChunk(session, rightChunkIndex, requestTime, requestSeq);
|
bestMatch = chooseBetterSeekFrame(bestMatch, rightMatch);
|
bestDrift = bestMatch == null ? Long.MAX_VALUE : bestMatch.drift;
|
checkedAny = true;
|
}
|
}
|
|
if (!checkedAny && bestMatch != null) {
|
break;
|
}
|
}
|
return bestMatch;
|
}
|
|
private ResolvedSeekFrame evaluateSeekChunk(ReplaySessionContext session, int chunkIndex, long requestTime, long requestSeq) {
|
if (chunkIndex < 0) {
|
return null;
|
}
|
ReplayChunk chunk = loadFullChunk(session, chunkIndex);
|
List<Map<String, Object>> seekFrames = chunk.getFullFrames() == null || chunk.getFullFrames().isEmpty()
|
? chunk.getFrames()
|
: chunk.getFullFrames();
|
if (seekFrames == null || seekFrames.isEmpty()) {
|
return null;
|
}
|
|
Map<String, Object> bestFrame = null;
|
long bestDrift = Long.MAX_VALUE;
|
for (Map<String, Object> frame : seekFrames) {
|
long drift = requestSeq > 0
|
? Math.abs(numberValue(frame.get("sampleSeq")) - requestSeq)
|
: Math.abs(numberValue(frame.get("timestamp")) - requestTime);
|
if (drift <= bestDrift) {
|
bestDrift = drift;
|
bestFrame = frame;
|
}
|
}
|
if (bestFrame == null) {
|
return null;
|
}
|
|
ResolvedSeekFrame resolvedSeekFrame = new ResolvedSeekFrame();
|
resolvedSeekFrame.chunkIndex = chunkIndex;
|
resolvedSeekFrame.chunk = chunk;
|
resolvedSeekFrame.frame = bestFrame;
|
resolvedSeekFrame.drift = bestDrift;
|
return resolvedSeekFrame;
|
}
|
|
private ResolvedSeekFrame chooseBetterSeekFrame(ResolvedSeekFrame currentBest, ResolvedSeekFrame challenger) {
|
if (challenger == null) {
|
return currentBest;
|
}
|
if (currentBest == null || challenger.drift <= currentBest.drift) {
|
return challenger;
|
}
|
return currentBest;
|
}
|
|
private long resolveChunkBoundaryDrift(ReplaySessionContext session, int chunkIndex, long requestTime) {
|
Map<String, Object> chunkMeta = resolveTimelineChunk(session, chunkIndex);
|
long chunkStartTime = numberValue(chunkMeta.get("startTimeMs"));
|
long chunkEndTime = numberValue(chunkMeta.get("endTimeMs"));
|
if (requestTime < chunkStartTime) {
|
return chunkStartTime - requestTime;
|
}
|
if (requestTime > chunkEndTime) {
|
return requestTime - chunkEndTime;
|
}
|
return 0L;
|
}
|
|
private int findNearestDisplayFrameIndex(ReplayChunk chunk, long resolvedTime) {
|
List<Map<String, Object>> displayFrames = chunk.getFrames();
|
if (displayFrames == null || displayFrames.isEmpty()) {
|
return 0;
|
}
|
int matchedIndex = 0;
|
long bestDrift = Long.MAX_VALUE;
|
for (int index = 0; index < displayFrames.size(); index++) {
|
Map<String, Object> frame = displayFrames.get(index);
|
long drift = Math.abs(numberValue(frame.get("timestamp")) - resolvedTime);
|
if (drift <= bestDrift) {
|
bestDrift = drift;
|
matchedIndex = index;
|
}
|
}
|
return matchedIndex;
|
}
|
|
@Override
|
public Map<String, Object> resolveDevice(String sessionId, String type, Integer deviceNo, Integer stationId, Long timestamp) {
|
ReplaySessionContext session = requireSession(sessionId);
|
DeviceReplayManifest manifest = resolveDeviceManifest(session, type, deviceNo, stationId);
|
if (manifest == null) {
|
throw new IllegalArgumentException("未找到对应设备的历史日志");
|
}
|
long targetTimestamp = timestamp == null || timestamp <= 0 ? session.getStartTimeMs() : timestamp;
|
rejectTimestampOutsideWindow(session, targetTimestamp, "设备详情时间不在当前回放窗口内");
|
Object resolvedState = resolveDeviceStateAt(manifest, targetTimestamp);
|
if (resolvedState == null) {
|
throw new IllegalArgumentException("当前时间没有该设备的历史详情");
|
}
|
Map<String, Object> data = new LinkedHashMap<>();
|
data.put("day", session.getDay());
|
data.put("type", type);
|
data.put("deviceNo", deviceNo);
|
data.put("stationId", stationId);
|
data.put("timestamp", targetTimestamp);
|
data.put("detail", JSON.parseObject(JSON.toJSONString(resolvedState)));
|
return data;
|
}
|
|
private DeviceReplayManifest resolveDeviceManifest(ReplaySessionContext session, String type, Integer deviceNo, Integer stationId) {
|
String manifestKey = buildStreamId(session.getDay(), type, deviceNo, stationId);
|
DeviceReplayManifest manifest = session.getManifestByStream().get(manifestKey);
|
if (manifest != null) {
|
return manifest;
|
}
|
if (!"Devp".equalsIgnoreCase(type) || stationId == null) {
|
return null;
|
}
|
for (DeviceReplayManifest currentManifest : session.getManifestByStream().values()) {
|
if (currentManifest == null) {
|
continue;
|
}
|
if (!"Devp".equalsIgnoreCase(currentManifest.getType())) {
|
continue;
|
}
|
if (stationId.equals(currentManifest.getStationId())) {
|
return currentManifest;
|
}
|
}
|
return null;
|
}
|
|
@Override
|
public void closeSession(String sessionId) {
|
if (sessionId != null) {
|
sessions.remove(sessionId);
|
}
|
}
|
|
@PreDestroy
|
public void shutdownReplayWarmupExecutor() {
|
replayWarmupExecutor.shutdownNow();
|
}
|
|
private TimelineChunkSnapshotData buildTimelineChunkSnapshot(List<DeviceReplayManifest> manifests,
|
long windowStartTimeMs,
|
long windowEndTimeMs) {
|
List<ChunkWindowMeta> allChunkMetas = collectTimelineChunkMetas(manifests, windowStartTimeMs, windowEndTimeMs);
|
if (allChunkMetas.isEmpty() || windowEndTimeMs < windowStartTimeMs) {
|
return TimelineChunkSnapshotData.empty();
|
}
|
|
List<Map<String, Object>> chunks = new ArrayList<>();
|
Map<Integer, List<String>> chunkStreamIdsByChunk = new LinkedHashMap<>();
|
|
long maxWindowMs = safePositive(maxWindowMinutes, 240) * 60L * 1000L;
|
long bucketWindowMs = safePositive(timelineBucketSeconds, 20) * 1000L;
|
if (maxWindowMs > 0) {
|
bucketWindowMs = Math.min(bucketWindowMs, maxWindowMs);
|
}
|
bucketWindowMs = Math.max(1_000L, bucketWindowMs);
|
|
int chunkIndex = 0;
|
for (long windowStart = windowStartTimeMs; windowStart <= windowEndTimeMs; windowStart += bucketWindowMs) {
|
long windowEnd = Math.min(windowEndTimeMs, windowStart + bucketWindowMs - 1);
|
int estimatedSamples = estimateWindowSamples(allChunkMetas, windowStart, windowEnd);
|
int targetChunkSampleSize = Math.max(1, safePositive(chunkSampleSize, 400));
|
int desiredSplitCount = Math.max(1, (int) Math.ceil(estimatedSamples / (double) targetChunkSampleSize));
|
int splitCount = Math.min(128, desiredSplitCount);
|
long windowDurationMs = Math.max(1L, windowEnd - windowStart + 1L);
|
long splitDurationMs = Math.max(1L, (long) Math.ceil(windowDurationMs / (double) splitCount));
|
|
for (int splitIndex = 0; splitIndex < splitCount; splitIndex++) {
|
long splitStartTime = windowStart + splitIndex * splitDurationMs;
|
if (splitStartTime > windowEnd) {
|
break;
|
}
|
long splitEndTime = Math.min(windowEnd, splitStartTime + splitDurationMs - 1L);
|
int splitEstimatedSamples = estimateWindowSamples(allChunkMetas, splitStartTime, splitEndTime);
|
long firstSampleSeq = resolveWindowFirstSampleSeq(allChunkMetas, splitStartTime, splitEndTime);
|
long lastSampleSeq = resolveWindowLastSampleSeq(allChunkMetas, splitStartTime, splitEndTime);
|
List<String> activeStreamIds = resolveWindowStreamIds(allChunkMetas, splitStartTime, splitEndTime);
|
chunks.add(buildChunkMeta(
|
chunkIndex++,
|
splitStartTime,
|
splitEndTime,
|
splitEstimatedSamples,
|
firstSampleSeq,
|
lastSampleSeq));
|
chunkStreamIdsByChunk.put(chunkIndex - 1, activeStreamIds);
|
}
|
}
|
TimelineChunkSnapshotData snapshotData = new TimelineChunkSnapshotData();
|
snapshotData.timelineChunks = Collections.unmodifiableList(new ArrayList<>(chunks));
|
snapshotData.chunkStreamIdsByChunk = Collections.unmodifiableMap(new LinkedHashMap<>(chunkStreamIdsByChunk));
|
return snapshotData;
|
}
|
|
private Map<String, DeviceReplayManifest> buildManifestSnapshot(List<DeviceReplayManifest> manifests) {
|
Map<String, DeviceReplayManifest> manifestByStream = new LinkedHashMap<>();
|
for (DeviceReplayManifest manifest : manifests) {
|
manifestByStream.put(manifest.getStreamId(), manifest);
|
}
|
return Collections.unmodifiableMap(manifestByStream);
|
}
|
|
private int estimateWindowSamples(List<ChunkWindowMeta> chunkMetas, long windowStart, long windowEnd) {
|
int estimatedSamples = 0;
|
for (ChunkWindowMeta chunkWindowMeta : chunkMetas) {
|
DeviceReplayChunkMeta chunkMeta = chunkWindowMeta.chunkMeta;
|
long chunkStartTime = safeLong(chunkMeta.getFirstSampleTimeMs());
|
long chunkEndTime = safeLong(chunkMeta.getLastSampleTimeMs());
|
if (chunkEndTime < windowStart || chunkStartTime > windowEnd) {
|
continue;
|
}
|
estimatedSamples += Math.max(1, chunkMeta.getSampleCount() == null ? 1 : chunkMeta.getSampleCount());
|
}
|
return estimatedSamples;
|
}
|
|
private StreamReplayScanResult scanManifestRange(DeviceReplayManifest manifest,
|
long startTime,
|
long endTime,
|
boolean allowBaselineLookup,
|
long scanStartExclusiveTimeMs) {
|
StreamReplayScanResult result = new StreamReplayScanResult();
|
List<DeviceReplayChunkMeta> candidates = new ArrayList<>();
|
DeviceReplayChunkMeta baselineChunk = null;
|
for (DeviceReplayChunkMeta chunkMeta : manifest.getChunks()) {
|
long chunkStart = safeLong(chunkMeta.getFirstSampleTimeMs());
|
long chunkEnd = safeLong(chunkMeta.getLastSampleTimeMs());
|
if (allowBaselineLookup && chunkEnd < startTime) {
|
baselineChunk = chunkMeta;
|
continue;
|
}
|
if (chunkEnd <= scanStartExclusiveTimeMs) {
|
continue;
|
}
|
if (chunkStart > endTime) {
|
break;
|
}
|
candidates.add(chunkMeta);
|
}
|
if (allowBaselineLookup && baselineChunk != null) {
|
candidates.add(0, baselineChunk);
|
}
|
for (DeviceReplayChunkMeta chunkMeta : candidates) {
|
Path logPath = resolveLogPath(manifest, chunkMeta);
|
if (!Files.exists(logPath)) {
|
continue;
|
}
|
try (BufferedReader reader = Files.newBufferedReader(logPath, StandardCharsets.UTF_8)) {
|
String line;
|
while ((line = reader.readLine()) != null) {
|
if (line == null || line.isBlank()) {
|
continue;
|
}
|
DeviceDataLog logItem;
|
try {
|
logItem = JSON.parseObject(line, DeviceDataLog.class);
|
} catch (Exception parseError) {
|
log.debug("跳过损坏的 replay 日志行, path={}", logPath, parseError);
|
continue;
|
}
|
if (logItem == null) {
|
continue;
|
}
|
long sampleTime = resolveSampleTime(logItem);
|
Object normalizedState = replayNormalizer.normalizeState(logItem);
|
if (normalizedState == null) {
|
continue;
|
}
|
if (allowBaselineLookup && sampleTime < startTime) {
|
result.baselineState = normalizedState;
|
continue;
|
}
|
if (sampleTime <= scanStartExclusiveTimeMs) {
|
continue;
|
}
|
if (sampleTime > endTime) {
|
break;
|
}
|
DeviceReplayState replayState = new DeviceReplayState();
|
replayState.setType(manifest.getType());
|
replayState.setDeviceNo(manifest.getDeviceNo());
|
replayState.setStationId(manifest.getStationId());
|
replayState.setTimestamp(sampleTime);
|
replayState.setSampleSeq(resolveSampleSeq(logItem));
|
replayState.setPayload(normalizedState);
|
result.eventList.add(replayState);
|
}
|
} catch (Exception e) {
|
log.warn("读取 replay chunk 失败, path={}", logPath, e);
|
}
|
}
|
return result;
|
}
|
|
private Object resolveDeviceStateAt(DeviceReplayManifest manifest, long targetTimestamp) {
|
Object latestState = null;
|
for (DeviceReplayChunkMeta chunkMeta : manifest.getChunks()) {
|
long chunkStart = safeLong(chunkMeta.getFirstSampleTimeMs());
|
if (chunkStart > targetTimestamp) {
|
break;
|
}
|
Path logPath = resolveLogPath(manifest, chunkMeta);
|
if (!Files.exists(logPath)) {
|
continue;
|
}
|
try (BufferedReader reader = Files.newBufferedReader(logPath, StandardCharsets.UTF_8)) {
|
String line;
|
while ((line = reader.readLine()) != null) {
|
if (line == null || line.isBlank()) {
|
continue;
|
}
|
DeviceDataLog logItem;
|
try {
|
logItem = JSON.parseObject(line, DeviceDataLog.class);
|
} catch (Exception parseError) {
|
log.debug("跳过损坏的回放设备详情日志行, path={}", logPath, parseError);
|
continue;
|
}
|
if (logItem == null) {
|
continue;
|
}
|
long sampleTime = resolveSampleTime(logItem);
|
if (sampleTime > targetTimestamp) {
|
return latestState;
|
}
|
Object normalizedState = replayNormalizer.normalizeState(logItem);
|
if (normalizedState != null) {
|
latestState = normalizedState;
|
}
|
}
|
} catch (Exception e) {
|
log.warn("读取回放设备详情失败, path={}", logPath, e);
|
}
|
}
|
return latestState;
|
}
|
|
private ChunkCheckpointSeed resolveCheckpointSeed(ReplaySessionContext session, Integer chunkIndex) {
|
if (session == null || chunkIndex == null || chunkIndex <= 0) {
|
return null;
|
}
|
ChunkCheckpointSeed hotCheckpoint = resolveCheckpointSeed(session.getChunkStateCheckpointCache(), session, chunkIndex);
|
if (hotCheckpoint != null) {
|
return hotCheckpoint;
|
}
|
return resolveCheckpointSeed(session.getSparseChunkStateCheckpointCache(), session, chunkIndex);
|
}
|
|
private ChunkCheckpointSeed resolveCheckpointSeed(Map<Integer, Map<String, Object>> checkpointCache,
|
ReplaySessionContext session,
|
Integer chunkIndex) {
|
synchronized (checkpointCache) {
|
for (int checkpointChunkIndex = chunkIndex - 1; checkpointChunkIndex >= 0; checkpointChunkIndex--) {
|
Map<String, Object> checkpointState = checkpointCache.get(checkpointChunkIndex);
|
if (checkpointState == null || checkpointState.isEmpty()) {
|
continue;
|
}
|
Map<String, Object> checkpointChunkMeta = resolveTimelineChunk(session, checkpointChunkIndex);
|
ChunkCheckpointSeed checkpointSeed = new ChunkCheckpointSeed();
|
checkpointSeed.chunkIndex = checkpointChunkIndex;
|
checkpointSeed.scanStartExclusiveTimeMs = numberValue(checkpointChunkMeta.get("endTimeMs"));
|
checkpointSeed.stateByStream = new LinkedHashMap<>(checkpointState);
|
return checkpointSeed;
|
}
|
}
|
return null;
|
}
|
|
private void appendFrame(ReplayChunk chunk, Map<String, Object> frame) {
|
int frameIndex = chunk.getFrames().size();
|
chunk.getFrames().add(frame);
|
ReplayFrameSummary summary = new ReplayFrameSummary();
|
summary.setFrameIndex(frameIndex);
|
summary.setTimestamp(numberValue(frame.get("timestamp")));
|
Map<String, Object> frameSummary = frame.get("summary") instanceof Map
|
? (Map<String, Object>) frame.get("summary")
|
: new LinkedHashMap<>();
|
summary.setAbnormalCount(intValue(frameSummary.get("abnormalCount")));
|
summary.setErrorCount(intValue(frameSummary.get("errorCount")));
|
summary.setBlockCount(intValue(frameSummary.get("blockCount")));
|
summary.setManualCount(0);
|
chunk.getFrameSummaryList().add(summary);
|
}
|
|
private Map<String, Object> resolveTimelineChunk(ReplaySessionContext session, Integer chunkIndex) {
|
return resolveTimelineChunk(session.getTimelineChunks(), chunkIndex);
|
}
|
|
private Map<String, Object> resolveTimelineChunk(List<Map<String, Object>> timelineChunks, Integer chunkIndex) {
|
if (chunkIndex == null || chunkIndex < 0 || chunkIndex >= timelineChunks.size()) {
|
throw new IllegalArgumentException("回放分片不存在");
|
}
|
return timelineChunks.get(chunkIndex);
|
}
|
|
private int resolveChunkIndex(ReplaySessionContext session, long timestamp, long sampleSeq) {
|
return resolveChunkIndex(session.getTimelineChunks(), timestamp, sampleSeq);
|
}
|
|
private int resolveChunkIndex(List<Map<String, Object>> timelineChunks, long timestamp, long sampleSeq) {
|
if (sampleSeq > 0) {
|
for (int i = 0; i < timelineChunks.size(); i++) {
|
Map<String, Object> chunkMeta = timelineChunks.get(i);
|
long firstSampleSeq = numberValue(chunkMeta.get("firstSampleSeq"));
|
long lastSampleSeq = numberValue(chunkMeta.get("lastSampleSeq"));
|
if (sampleSeq >= firstSampleSeq && sampleSeq <= lastSampleSeq) {
|
return i;
|
}
|
}
|
}
|
for (int i = 0; i < timelineChunks.size(); i++) {
|
Map<String, Object> chunkMeta = timelineChunks.get(i);
|
long startTime = numberValue(chunkMeta.get("startTimeMs"));
|
long endTime = numberValue(chunkMeta.get("endTimeMs"));
|
if (timestamp >= startTime && timestamp <= endTime) {
|
return i;
|
}
|
}
|
return Math.max(0, timelineChunks.size() - 1);
|
}
|
|
private int resolveSummaryBucketCount(ReplaySessionContext session, Integer bucketCount) {
|
int totalChunkCount = session.getTimelineChunks().size();
|
if (totalChunkCount <= 0) {
|
return 0;
|
}
|
int defaultBucketCount = Math.min(totalChunkCount, 96);
|
int safeBucketCount = bucketCount == null || bucketCount <= 0 ? defaultBucketCount : bucketCount;
|
return Math.max(1, Math.min(totalChunkCount, safeBucketCount));
|
}
|
|
private Map<String, Object> buildChunkMeta(int chunkIndex,
|
long startTimeMs,
|
long endTimeMs,
|
int estimatedSamples,
|
long firstSampleSeq,
|
long lastSampleSeq) {
|
Map<String, Object> chunkMeta = new LinkedHashMap<>();
|
chunkMeta.put("chunkIndex", chunkIndex);
|
chunkMeta.put("startTimeMs", startTimeMs);
|
chunkMeta.put("endTimeMs", endTimeMs);
|
chunkMeta.put("estimatedSamples", estimatedSamples);
|
chunkMeta.put("firstSampleSeq", firstSampleSeq);
|
chunkMeta.put("lastSampleSeq", lastSampleSeq);
|
return chunkMeta;
|
}
|
|
private long resolveWindowFirstSampleSeq(List<ChunkWindowMeta> chunkMetas, long windowStart, long windowEnd) {
|
long firstSampleSeq = Long.MAX_VALUE;
|
for (ChunkWindowMeta chunkWindowMeta : chunkMetas) {
|
DeviceReplayChunkMeta chunkMeta = chunkWindowMeta.chunkMeta;
|
long chunkStartTime = safeLong(chunkMeta.getFirstSampleTimeMs());
|
long chunkEndTime = safeLong(chunkMeta.getLastSampleTimeMs());
|
if (chunkEndTime < windowStart || chunkStartTime > windowEnd) {
|
continue;
|
}
|
long candidate = safeLong(chunkMeta.getFirstSampleSeq());
|
if (candidate > 0) {
|
firstSampleSeq = Math.min(firstSampleSeq, candidate);
|
}
|
}
|
return firstSampleSeq == Long.MAX_VALUE ? 0L : firstSampleSeq;
|
}
|
|
private long resolveWindowLastSampleSeq(List<ChunkWindowMeta> chunkMetas, long windowStart, long windowEnd) {
|
long lastSampleSeq = 0L;
|
for (ChunkWindowMeta chunkWindowMeta : chunkMetas) {
|
DeviceReplayChunkMeta chunkMeta = chunkWindowMeta.chunkMeta;
|
long chunkStartTime = safeLong(chunkMeta.getFirstSampleTimeMs());
|
long chunkEndTime = safeLong(chunkMeta.getLastSampleTimeMs());
|
if (chunkEndTime < windowStart || chunkStartTime > windowEnd) {
|
continue;
|
}
|
lastSampleSeq = Math.max(lastSampleSeq, safeLong(chunkMeta.getLastSampleSeq()));
|
}
|
return lastSampleSeq;
|
}
|
|
private List<DeviceReplayManifest> resolveChunkManifests(ReplaySessionContext session, Integer chunkIndex) {
|
if (session == null || chunkIndex == null) {
|
return new ArrayList<>();
|
}
|
List<String> streamIds = session.getChunkStreamIdsByChunk().get(chunkIndex);
|
if (streamIds == null) {
|
return new ArrayList<>(session.getManifestByStream().values());
|
}
|
List<DeviceReplayManifest> manifests = new ArrayList<>(streamIds.size());
|
for (String streamId : streamIds) {
|
DeviceReplayManifest manifest = session.getManifestByStream().get(streamId);
|
if (manifest != null) {
|
manifests.add(manifest);
|
}
|
}
|
return manifests;
|
}
|
|
private List<ChunkWindowMeta> collectTimelineChunkMetas(List<DeviceReplayManifest> manifests,
|
long windowStartTimeMs,
|
long windowEndTimeMs) {
|
List<ChunkWindowMeta> chunkMetas = new ArrayList<>();
|
for (DeviceReplayManifest manifest : manifests) {
|
if (manifest == null || manifest.getChunks() == null || manifest.getChunks().isEmpty()) {
|
continue;
|
}
|
for (DeviceReplayChunkMeta chunkMeta : manifest.getChunks()) {
|
long chunkStartTime = safeLong(chunkMeta.getFirstSampleTimeMs());
|
long chunkEndTime = safeLong(chunkMeta.getLastSampleTimeMs());
|
if (chunkStartTime <= 0 || chunkEndTime < chunkStartTime) {
|
continue;
|
}
|
if (chunkEndTime < windowStartTimeMs || chunkStartTime > windowEndTimeMs) {
|
continue;
|
}
|
ChunkWindowMeta chunkWindowMeta = new ChunkWindowMeta();
|
chunkWindowMeta.streamId = manifest.getStreamId();
|
chunkWindowMeta.chunkMeta = chunkMeta;
|
chunkMetas.add(chunkWindowMeta);
|
}
|
}
|
return chunkMetas;
|
}
|
|
private List<String> resolveWindowStreamIds(List<ChunkWindowMeta> chunkMetas,
|
long windowStartTimeMs,
|
long windowEndTimeMs) {
|
LinkedHashSet<String> streamIds = new LinkedHashSet<>();
|
for (ChunkWindowMeta chunkWindowMeta : chunkMetas) {
|
DeviceReplayChunkMeta chunkMeta = chunkWindowMeta.chunkMeta;
|
long chunkStartTime = safeLong(chunkMeta.getFirstSampleTimeMs());
|
long chunkEndTime = safeLong(chunkMeta.getLastSampleTimeMs());
|
if (chunkEndTime < windowStartTimeMs || chunkStartTime > windowEndTimeMs) {
|
continue;
|
}
|
streamIds.add(chunkWindowMeta.streamId);
|
}
|
return new ArrayList<>(streamIds);
|
}
|
|
private long clampTimestampToWindow(ReplaySessionContext session, long timestamp) {
|
long windowStartTimeMs = safeLong(session.getWindowStartTimeMs());
|
long windowEndTimeMs = safeLong(session.getWindowEndTimeMs());
|
if (timestamp < windowStartTimeMs) {
|
return windowStartTimeMs;
|
}
|
if (timestamp > windowEndTimeMs) {
|
return windowEndTimeMs;
|
}
|
return timestamp;
|
}
|
|
private long clampTimestampToWindow(TimelineSummarySnapshot snapshot, long timestamp) {
|
if (timestamp < snapshot.windowStartTimeMs) {
|
return snapshot.windowStartTimeMs;
|
}
|
if (timestamp > snapshot.windowEndTimeMs) {
|
return snapshot.windowEndTimeMs;
|
}
|
return timestamp;
|
}
|
|
private void rejectTimestampOutsideWindow(ReplaySessionContext session, long timestamp, String message) {
|
long windowStartTimeMs = safeLong(session.getWindowStartTimeMs());
|
long windowEndTimeMs = safeLong(session.getWindowEndTimeMs());
|
if (timestamp < windowStartTimeMs || timestamp > windowEndTimeMs) {
|
throwReplayWindowException(REPLAY_SEEK_OUT_OF_WINDOW, message);
|
}
|
}
|
|
private void rejectSampleSeqOutsideWindow(ReplaySessionContext session, long sampleSeq) {
|
if (sampleSeq <= 0) {
|
return;
|
}
|
long firstWindowSampleSeq = Long.MAX_VALUE;
|
long lastWindowSampleSeq = 0L;
|
for (Map<String, Object> chunkMeta : session.getTimelineChunks()) {
|
long firstSampleSeq = numberValue(chunkMeta.get("firstSampleSeq"));
|
long lastSampleSeq = numberValue(chunkMeta.get("lastSampleSeq"));
|
if (firstSampleSeq <= 0 || lastSampleSeq <= 0) {
|
continue;
|
}
|
firstWindowSampleSeq = Math.min(firstWindowSampleSeq, firstSampleSeq);
|
lastWindowSampleSeq = Math.max(lastWindowSampleSeq, lastSampleSeq);
|
}
|
if (firstWindowSampleSeq == Long.MAX_VALUE || lastWindowSampleSeq <= 0) {
|
return;
|
}
|
if (sampleSeq < firstWindowSampleSeq || sampleSeq > lastWindowSampleSeq) {
|
throwReplayWindowException(REPLAY_SEEK_OUT_OF_WINDOW, "seek 采样序号不在当前回放窗口内");
|
}
|
}
|
|
private boolean isSameManifestSnapshot(ReplaySessionContext session, String manifestSnapshotKey) {
|
if (session == null) {
|
return false;
|
}
|
String currentManifestSnapshotKey = session.getManifestSnapshotKey();
|
return manifestSnapshotKey == null
|
? currentManifestSnapshotKey == null
|
: manifestSnapshotKey.equals(currentManifestSnapshotKey);
|
}
|
|
private ReplaySessionContext requireSession(String sessionId) {
|
cleanupExpiredSessions();
|
ReplaySessionContext session = sessions.get(sessionId);
|
if (session == null) {
|
throw new IllegalArgumentException("回放会话不存在或已过期");
|
}
|
session.setLastAccessAtMs(System.currentTimeMillis());
|
refreshCurrentDaySessionIfNeeded(session);
|
return session;
|
}
|
|
private void refreshCurrentDaySessionIfNeeded(ReplaySessionContext session) {
|
if (session == null || !isCurrentReplayDay(session.getDay())) {
|
return;
|
}
|
long now = System.currentTimeMillis();
|
Long lastRefreshAtMs = session.getLastManifestRefreshAtMs();
|
if (lastRefreshAtMs != null && (now - lastRefreshAtMs) < CURRENT_DAY_MANIFEST_REFRESH_INTERVAL_MS) {
|
return;
|
}
|
|
String sessionDay = session.getDay();
|
long windowStartTimeMs = safeLong(session.getWindowStartTimeMs());
|
long windowEndTimeMs = safeLong(session.getWindowEndTimeMs());
|
List<DeviceReplayManifest> manifests = loadSessionManifests(sessionDay, true, false);
|
if (manifests.isEmpty()) {
|
manifestService.scheduleDayManifestRefresh(sessionDay);
|
synchronized (session) {
|
long currentTime = System.currentTimeMillis();
|
Long currentLastRefreshAtMs = session.getLastManifestRefreshAtMs();
|
if (currentLastRefreshAtMs != null
|
&& (currentTime - currentLastRefreshAtMs) < CURRENT_DAY_MANIFEST_REFRESH_INTERVAL_MS) {
|
return;
|
}
|
session.setLastManifestRefreshAtMs(currentTime);
|
}
|
return;
|
}
|
|
boolean manifestChanged;
|
String manifestSnapshotKey = manifests.isEmpty() ? "" : buildManifestSnapshotKey(manifests);
|
Map<String, DeviceReplayManifest> manifestByStream = manifests.isEmpty()
|
? Collections.emptyMap()
|
: buildManifestSnapshot(manifests);
|
TimelineChunkSnapshotData timelineSnapshot = manifests.isEmpty()
|
? TimelineChunkSnapshotData.empty()
|
: buildTimelineChunkSnapshot(manifests, windowStartTimeMs, windowEndTimeMs);
|
synchronized (session) {
|
long currentTime = System.currentTimeMillis();
|
Long currentLastRefreshAtMs = session.getLastManifestRefreshAtMs();
|
if (currentLastRefreshAtMs != null
|
&& (currentTime - currentLastRefreshAtMs) < CURRENT_DAY_MANIFEST_REFRESH_INTERVAL_MS) {
|
return;
|
}
|
manifestChanged = !manifests.isEmpty() && !manifestSnapshotKey.equals(session.getManifestSnapshotKey());
|
if (manifestChanged) {
|
session.setManifestByStream(manifestByStream);
|
session.setTimelineChunks(timelineSnapshot.timelineChunks);
|
session.setChunkStreamIdsByChunk(timelineSnapshot.chunkStreamIdsByChunk);
|
session.setStartTimeMs(windowStartTimeMs);
|
session.setEndTimeMs(windowEndTimeMs);
|
session.setManifestSnapshotKey(manifestSnapshotKey);
|
session.setTailCheckpointWarmupScheduled(false);
|
clearSessionCaches(session);
|
}
|
session.setLastManifestRefreshAtMs(currentTime);
|
}
|
|
manifestService.scheduleDayManifestRefresh(sessionDay);
|
if (manifestChanged) {
|
scheduleTailCheckpointWarmup(session.getSessionId());
|
}
|
}
|
|
private List<DeviceReplayManifest> loadSessionManifests(String day,
|
boolean currentReplayDay,
|
boolean allowBuildIfMissing) {
|
List<DeviceReplayManifest> manifests = manifestService.loadDayManifests(day, false);
|
if (!manifests.isEmpty()) {
|
return manifests;
|
}
|
if (allowBuildIfMissing) {
|
return manifestService.loadDayManifests(day, true);
|
}
|
if (!currentReplayDay) {
|
return manifests;
|
}
|
return manifests;
|
}
|
|
private void applyManifestSnapshot(ReplaySessionContext session,
|
List<DeviceReplayManifest> manifests,
|
String manifestSnapshotKey) {
|
if (session == null || manifests == null || manifests.isEmpty()) {
|
return;
|
}
|
long windowStartTimeMs = safeLong(session.getWindowStartTimeMs());
|
long windowEndTimeMs = safeLong(session.getWindowEndTimeMs());
|
Map<String, DeviceReplayManifest> manifestByStream = buildManifestSnapshot(manifests);
|
TimelineChunkSnapshotData timelineSnapshot = buildTimelineChunkSnapshot(manifests, windowStartTimeMs, windowEndTimeMs);
|
synchronized (session) {
|
session.setManifestByStream(manifestByStream);
|
session.setTimelineChunks(timelineSnapshot.timelineChunks);
|
session.setChunkStreamIdsByChunk(timelineSnapshot.chunkStreamIdsByChunk);
|
session.setStartTimeMs(windowStartTimeMs);
|
session.setEndTimeMs(windowEndTimeMs);
|
session.setManifestSnapshotKey(manifestSnapshotKey);
|
session.setTailCheckpointWarmupScheduled(false);
|
clearSessionCaches(session);
|
}
|
}
|
|
private void clearSessionCaches(ReplaySessionContext session) {
|
synchronized (session.getChunkCache()) {
|
session.getChunkCache().clear();
|
}
|
synchronized (session.getCompactChunkCache()) {
|
session.getCompactChunkCache().clear();
|
}
|
synchronized (session.getChunkStateCheckpointCache()) {
|
session.getChunkStateCheckpointCache().clear();
|
}
|
synchronized (session.getSparseChunkStateCheckpointCache()) {
|
session.getSparseChunkStateCheckpointCache().clear();
|
}
|
synchronized (session.getTimelineSummaryCache()) {
|
session.getTimelineSummaryCache().clear();
|
}
|
}
|
|
private ReplayChunk getCachedChunk(ReplaySessionContext session, Integer chunkIndex, boolean fullFrames) {
|
if (session == null || chunkIndex == null) {
|
return null;
|
}
|
Map<Integer, ReplayChunk> cache = fullFrames ? session.getChunkCache() : session.getCompactChunkCache();
|
synchronized (cache) {
|
return cache.get(chunkIndex);
|
}
|
}
|
|
private void cacheChunk(ReplaySessionContext session, Integer chunkIndex, ReplayChunk chunk, boolean fullFrames) {
|
if (session == null || chunkIndex == null || chunk == null) {
|
return;
|
}
|
Map<Integer, ReplayChunk> cache = fullFrames ? session.getChunkCache() : session.getCompactChunkCache();
|
int configuredMaxCacheSize = safePositive(maxChunkCacheEntries, 128);
|
int minimumWarmCacheSize = Math.max(fullFrames ? 24 : 48, safePositive(maxPrefetchChunks, 3) + 8);
|
int maxCacheSize = Math.max(minimumWarmCacheSize, configuredMaxCacheSize);
|
synchronized (cache) {
|
cache.remove(chunkIndex);
|
cache.put(chunkIndex, chunk);
|
while (cache.size() > maxCacheSize) {
|
Integer eldestKey = cache.keySet().iterator().next();
|
cache.remove(eldestKey);
|
}
|
}
|
}
|
|
private void cacheChunkCheckpoint(ReplaySessionContext session, Integer chunkIndex, Map<String, Object> currentStateByStream) {
|
if (session == null || chunkIndex == null || currentStateByStream == null || currentStateByStream.isEmpty()) {
|
return;
|
}
|
Map<Integer, Map<String, Object>> cache = session.getChunkStateCheckpointCache();
|
int configuredMaxCacheSize = safePositive(maxChunkCacheEntries, 128);
|
int maxCacheSize = Math.max(32, configuredMaxCacheSize);
|
synchronized (cache) {
|
cache.remove(chunkIndex);
|
cache.put(chunkIndex, new LinkedHashMap<>(currentStateByStream));
|
while (cache.size() > maxCacheSize) {
|
Integer eldestKey = cache.keySet().iterator().next();
|
cache.remove(eldestKey);
|
}
|
}
|
Map<Integer, Map<String, Object>> sparseCache = session.getSparseChunkStateCheckpointCache();
|
int sparseMaxCacheSize = Math.max(512, safePositive(maxChunkCacheEntries, 128) * 8);
|
synchronized (sparseCache) {
|
sparseCache.remove(chunkIndex);
|
sparseCache.put(chunkIndex, new LinkedHashMap<>(currentStateByStream));
|
while (sparseCache.size() > sparseMaxCacheSize) {
|
Integer eldestKey = sparseCache.keySet().iterator().next();
|
sparseCache.remove(eldestKey);
|
}
|
}
|
}
|
|
private boolean isCurrentReplayDay(String day) {
|
if (day == null || day.isBlank()) {
|
return false;
|
}
|
return day.equals(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE));
|
}
|
|
private ReplayWindow resolveReplayWindow(String day,
|
Long targetTimestamp,
|
Long windowStartTimeMs,
|
Long windowEndTimeMs) {
|
DayBoundary dayBoundary = resolveDayBoundary(day);
|
boolean hasTargetTimestamp = targetTimestamp != null;
|
boolean hasWindowStart = windowStartTimeMs != null;
|
boolean hasWindowEnd = windowEndTimeMs != null;
|
|
if (!hasTargetTimestamp && !hasWindowStart && !hasWindowEnd) {
|
throwReplayWindowException(REPLAY_WINDOW_MISSING, "请选择历史回放目标时间或时间窗口");
|
}
|
if (hasWindowStart != hasWindowEnd) {
|
throwReplayWindowException(REPLAY_WINDOW_INVALID, "历史回放时间窗口开始和结束时间必须同时传入");
|
}
|
if (hasWindowStart) {
|
return validateExplicitReplayWindow(dayBoundary, windowStartTimeMs, windowEndTimeMs);
|
}
|
return resolveTargetReplayWindow(dayBoundary, targetTimestamp);
|
}
|
|
private DayBoundary resolveDayBoundary(String day) {
|
try {
|
LocalDate replayDay = LocalDate.parse(day, DateTimeFormatter.BASIC_ISO_DATE);
|
long dayStartTimeMs = replayDay.atStartOfDay(ZoneId.systemDefault()).toInstant().toEpochMilli();
|
return new DayBoundary(dayStartTimeMs, dayStartTimeMs + DAY_END_OFFSET_MS);
|
} catch (DateTimeParseException e) {
|
throwReplayWindowException(REPLAY_WINDOW_INVALID, "历史回放日期格式无效");
|
return null;
|
}
|
}
|
|
private ReplayWindow validateExplicitReplayWindow(DayBoundary dayBoundary,
|
Long windowStartTimeMs,
|
Long windowEndTimeMs) {
|
if (windowStartTimeMs >= windowEndTimeMs) {
|
throwReplayWindowException(REPLAY_WINDOW_INVALID, "历史回放时间窗口开始时间必须早于结束时间");
|
}
|
if (!isWithinDay(dayBoundary, windowStartTimeMs) || !isWithinDay(dayBoundary, windowEndTimeMs)) {
|
throwReplayWindowException(REPLAY_WINDOW_CROSS_DAY, "历史回放时间窗口不能跨自然日");
|
}
|
validateReplayWindowSize(windowStartTimeMs, windowEndTimeMs);
|
return new ReplayWindow(windowStartTimeMs, windowEndTimeMs);
|
}
|
|
private ReplayWindow resolveTargetReplayWindow(DayBoundary dayBoundary, Long targetTimestamp) {
|
if (!isWithinDay(dayBoundary, targetTimestamp)) {
|
throwReplayWindowException(REPLAY_WINDOW_CROSS_DAY, "历史回放目标时间不在所选日期内");
|
}
|
long windowStartTimeMs = Math.max(dayBoundary.startTimeMs, targetTimestamp - DEFAULT_REPLAY_WINDOW_HALF_MS);
|
long windowEndTimeMs = Math.min(dayBoundary.endTimeMs, targetTimestamp + DEFAULT_REPLAY_WINDOW_HALF_MS);
|
if (windowStartTimeMs >= windowEndTimeMs) {
|
throwReplayWindowException(REPLAY_WINDOW_INVALID, "历史回放时间窗口无效");
|
}
|
validateReplayWindowSize(windowStartTimeMs, windowEndTimeMs);
|
return new ReplayWindow(windowStartTimeMs, windowEndTimeMs);
|
}
|
|
private boolean isWithinDay(DayBoundary dayBoundary, Long timestamp) {
|
return dayBoundary != null
|
&& timestamp != null
|
&& timestamp >= dayBoundary.startTimeMs
|
&& timestamp <= dayBoundary.endTimeMs;
|
}
|
|
private void validateReplayWindowSize(long windowStartTimeMs, long windowEndTimeMs) {
|
long maxWindowMs = safePositive(maxWindowMinutes, 240) * 60L * 1000L;
|
if ((windowEndTimeMs - windowStartTimeMs) > maxWindowMs) {
|
throwReplayWindowException(REPLAY_WINDOW_TOO_LARGE, "历史回放时间窗口超过最大允许范围");
|
}
|
}
|
|
private void throwReplayWindowException(String errorCode, String message) {
|
throw new DeviceLogReplayService.ReplayWindowException(errorCode, message);
|
}
|
|
private static class DayBoundary {
|
private final long startTimeMs;
|
private final long endTimeMs;
|
|
private DayBoundary(long startTimeMs, long endTimeMs) {
|
this.startTimeMs = startTimeMs;
|
this.endTimeMs = endTimeMs;
|
}
|
}
|
|
private static class ReplayWindow {
|
private final long startTimeMs;
|
private final long endTimeMs;
|
|
private ReplayWindow(long startTimeMs, long endTimeMs) {
|
this.startTimeMs = startTimeMs;
|
this.endTimeMs = endTimeMs;
|
}
|
}
|
|
private void scheduleTailCheckpointWarmup(String sessionId) {
|
ReplaySessionContext session = sessions.get(sessionId);
|
if (session == null) {
|
return;
|
}
|
synchronized (session) {
|
if (session.isTailCheckpointWarmupScheduled()) {
|
return;
|
}
|
session.setTailCheckpointWarmupScheduled(true);
|
}
|
replayWarmupExecutor.submit(() -> warmupTailCheckpoints(sessionId));
|
}
|
|
private void warmupTailCheckpoints(String sessionId) {
|
ReplaySessionContext session = sessions.get(sessionId);
|
if (session == null || session.getTimelineChunks() == null || session.getTimelineChunks().isEmpty()) {
|
return;
|
}
|
int chunkCount = session.getTimelineChunks().size();
|
int warmupSpan = Math.min(chunkCount, 2048);
|
int warmupStep = 32;
|
int startChunkIndex = Math.max(0, chunkCount - warmupSpan);
|
for (int chunkIndex = startChunkIndex; chunkIndex < chunkCount; chunkIndex += warmupStep) {
|
if (!sessions.containsKey(sessionId)) {
|
return;
|
}
|
try {
|
loadFullChunk(session, chunkIndex);
|
} catch (Exception e) {
|
log.debug("预热 replay 尾部 checkpoint 失败, sessionId={}, chunkIndex={}", sessionId, chunkIndex, e);
|
}
|
}
|
}
|
|
private void cleanupExpiredSessions() {
|
long ttlMs = safePositive(sessionTtlSeconds, 1800) * 1000L;
|
long now = System.currentTimeMillis();
|
for (Map.Entry<String, ReplaySessionContext> entry : sessions.entrySet()) {
|
ReplaySessionContext session = entry.getValue();
|
if (session == null || session.getLastAccessAtMs() == null || (now - session.getLastAccessAtMs()) > ttlMs) {
|
sessions.remove(entry.getKey());
|
}
|
}
|
}
|
|
private void ensureFileMode() {
|
if (!"file".equalsIgnoreCase(storageType)) {
|
throw new IllegalStateException("当前仅支持 file 模式历史回放");
|
}
|
}
|
|
private Path resolveLogPath(DeviceReplayManifest manifest, DeviceReplayChunkMeta chunkMeta) {
|
return Paths.get(loggingPath, manifest.getDay(), chunkMeta.getRelativePath());
|
}
|
|
private long resolveSampleTime(DeviceDataLog logItem) {
|
if (logItem.getSampleTimeMs() != null && logItem.getSampleTimeMs() > 0) {
|
return logItem.getSampleTimeMs();
|
}
|
if (logItem.getCreateTime() != null) {
|
return logItem.getCreateTime().getTime();
|
}
|
return 0L;
|
}
|
|
private long resolveSampleSeq(DeviceDataLog logItem) {
|
if (logItem.getSampleSeq() != null && logItem.getSampleSeq() > 0) {
|
return logItem.getSampleSeq();
|
}
|
return resolveSampleTime(logItem);
|
}
|
|
private long safeLong(Long value) {
|
return value == null ? 0L : value;
|
}
|
|
private int safePositive(Integer value, int fallback) {
|
return value == null || value <= 0 ? fallback : value;
|
}
|
|
private long numberValue(Object value) {
|
if (value instanceof Number) {
|
return ((Number) value).longValue();
|
}
|
if (value == null) {
|
return 0L;
|
}
|
try {
|
return Long.parseLong(String.valueOf(value));
|
} catch (Exception e) {
|
return 0L;
|
}
|
}
|
|
private int intValue(Object value) {
|
if (value instanceof Number) {
|
return ((Number) value).intValue();
|
}
|
if (value == null) {
|
return 0;
|
}
|
try {
|
return Integer.parseInt(String.valueOf(value));
|
} catch (Exception e) {
|
return 0;
|
}
|
}
|
|
private String stringValue(Object value) {
|
return value == null ? "" : String.valueOf(value);
|
}
|
|
private int resolveAbnormalPriority(String level) {
|
if ("ERROR".equalsIgnoreCase(level)) {
|
return 0;
|
}
|
if ("BLOCK".equalsIgnoreCase(level)) {
|
return 1;
|
}
|
return 2;
|
}
|
|
private void warmupEntryManifest(String day, String type, Integer deviceNo, Integer stationId) {
|
if (day == null || type == null || type.isBlank() || deviceNo == null) {
|
return;
|
}
|
manifestService.loadManifest(DeviceReplayStreamKey.of(day, type, deviceNo, stationId), true);
|
}
|
|
private String buildManifestSnapshotKey(List<DeviceReplayManifest> manifests) {
|
if (manifests == null || manifests.isEmpty()) {
|
return "";
|
}
|
StringBuilder builder = new StringBuilder();
|
for (DeviceReplayManifest manifest : manifests) {
|
if (manifest == null) {
|
continue;
|
}
|
builder.append(manifest.getStreamId())
|
.append('#')
|
.append(manifest.getChunks() == null ? 0 : manifest.getChunks().size())
|
.append('#')
|
.append(safeLong(manifest.getFirstSampleSeq()))
|
.append('#')
|
.append(safeLong(manifest.getLastSampleSeq()))
|
.append('#')
|
.append(safeLong(manifest.getFirstSampleTimeMs()))
|
.append('#')
|
.append(safeLong(manifest.getLastSampleTimeMs()))
|
.append(';');
|
}
|
return builder.toString();
|
}
|
|
private boolean equalsLong(Long left, Long right) {
|
if (left == null) {
|
return right == null;
|
}
|
return left.equals(right);
|
}
|
|
private String buildStreamId(String day, String type, Integer deviceNo, Integer stationId) {
|
StringBuilder builder = new StringBuilder();
|
if (day != null) {
|
builder.append(day).append('|');
|
}
|
builder.append(type == null ? "" : type)
|
.append('|')
|
.append(deviceNo == null ? "" : deviceNo);
|
if (stationId != null) {
|
builder.append('|').append(stationId);
|
}
|
return builder.toString();
|
}
|
|
private static class StreamReplayScanResult {
|
private Object baselineState;
|
private List<DeviceReplayState> eventList = new ArrayList<>();
|
}
|
|
private static class TimelineSummarySnapshot {
|
private String sessionId;
|
private String day;
|
private long startTimeMs;
|
private long endTimeMs;
|
private long windowStartTimeMs;
|
private long windowEndTimeMs;
|
private List<Map<String, Object>> timelineChunks = new ArrayList<>();
|
private List<DeviceReplayManifest> manifests = new ArrayList<>();
|
}
|
|
private static class ChunkCheckpointSeed {
|
private Integer chunkIndex;
|
private long scanStartExclusiveTimeMs;
|
private Map<String, Object> stateByStream;
|
}
|
|
private static class ChunkWindowMeta {
|
private String streamId;
|
private DeviceReplayChunkMeta chunkMeta;
|
}
|
|
private static class TimelineChunkSnapshotData {
|
private List<Map<String, Object>> timelineChunks = Collections.emptyList();
|
private Map<Integer, List<String>> chunkStreamIdsByChunk = Collections.emptyMap();
|
|
private static TimelineChunkSnapshotData empty() {
|
return new TimelineChunkSnapshotData();
|
}
|
}
|
|
private static class ResolvedSeekFrame {
|
private Integer chunkIndex;
|
private ReplayChunk chunk;
|
private Map<String, Object> frame;
|
private long drift;
|
}
|
}
|
