#

Junjie

1 天以前 6c8ee211ae63de5f4ce0bd13b9c34f0524cd656f

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
package com.zy.core.task;
 
import com.zy.asrs.entity.DeviceDataLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
 
import java.util.concurrent.atomic.AtomicLong;
 
@Slf4j
@Component
public class DeviceAsyncLogPublisher {
 
    private static final int REPLAY_SCHEMA_VERSION = 1;
 
    private final AtomicLong sampleSequence = new AtomicLong(System.currentTimeMillis() * 1000L);
    private final DeviceLogFileWriter deviceLogFileWriter;
 
    public DeviceAsyncLogPublisher(DeviceLogFileWriter deviceLogFileWriter) {
        this.deviceLogFileWriter = deviceLogFileWriter;
    }
 
    public void publishLatest(DeviceDataLog deviceDataLog) {
        enrichReplayFields(deviceDataLog);
        String laneKey = buildLaneKey(deviceDataLog);
        if (laneKey == null) {
            return;
        }
        // Writer owns lane state, buffering, flush, rotation and shutdown semantics.
        deviceLogFileWriter.publishLatest(laneKey, deviceDataLog);
    }
 
    private String buildLaneKey(DeviceDataLog deviceDataLog) {
        if (deviceDataLog == null || deviceDataLog.getType() == null || deviceDataLog.getDeviceNo() == null) {
            return null;
        }
        if ("Devp".equals(deviceDataLog.getType())) {
            if (deviceDataLog.getStationId() == null) {
                return null;
            }
            return deviceDataLog.getType() + ":" + deviceDataLog.getDeviceNo() + ":" + deviceDataLog.getStationId();
        }
        return deviceDataLog.getType() + ":" + deviceDataLog.getDeviceNo();
    }
 
    private void enrichReplayFields(DeviceDataLog deviceDataLog) {
        if (deviceDataLog == null) {
            return;
        }
        if (deviceDataLog.getSampleSeq() == null || deviceDataLog.getSampleSeq() <= 0) {
            deviceDataLog.setSampleSeq(sampleSequence.incrementAndGet());
        }
        if (deviceDataLog.getSampleTimeMs() == null || deviceDataLog.getSampleTimeMs() <= 0) {
            long timestamp = deviceDataLog.getCreateTime() == null
                    ? System.currentTimeMillis()
                    : deviceDataLog.getCreateTime().getTime();
            deviceDataLog.setSampleTimeMs(timestamp);
        }
        if (deviceDataLog.getSchemaVersion() == null || deviceDataLog.getSchemaVersion() <= 0) {
            deviceDataLog.setSchemaVersion(REPLAY_SCHEMA_VERSION);
        }
    }
 
}