Compare commits
13 Commits
| Author | SHA1 | Date |
|---|---|---|
review512jwy@163.com
|
4736e21568 | 2 weeks ago |
|
review512jwy@163.com
|
a676119130 | 3 weeks ago |
 zhczyx
|
424260a386 | 1 month ago |
|
review512jwy@163.com
|
9514deac58 | 1 month ago |
|
zhczyx
|
e1c6481f17 | 1 month ago |
|
review512jwy@163.com
|
2850828812 | 2 months ago |
|
review512jwy@163.com
|
8b98cc808c | 3 months ago |
|
zhczyx
|
6237fc996e | 3 months ago |
|
review512jwy@163.com
|
4d621f7a7b | 3 months ago |
|
review512jwy@163.com
|
8e359b0b9e | 3 months ago |
|
review512jwy@163.com
|
e49f1b8948 | 3 months ago |
|
review512jwy@163.com
|
2341784be7 | 3 months ago |
|
review512jwy@163.com
|
4b8db40fb6 | 3 months ago |
21 changed files with 1859 additions and 35 deletions
@ -0,0 +1 @@ |
|||
17 |
|||
@ -0,0 +1,363 @@ |
|||
package com.techsor.datacenter.receiver.listener; |
|||
|
|||
import com.google.gson.Gson; |
|||
import com.techsor.datacenter.receiver.config.DataCenterEnvConfig; |
|||
import com.techsor.datacenter.receiver.utils.DefaultHttpRequestUtil; |
|||
import com.techsor.datacenter.receiver.utils.MyHTTPResponse; |
|||
import jakarta.annotation.PostConstruct; |
|||
import jakarta.annotation.PreDestroy; |
|||
import jakarta.annotation.Resource; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.domain.Range; |
|||
import org.springframework.data.redis.connection.stream.*; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.data.redis.core.StringRedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.net.InetAddress; |
|||
import java.time.Duration; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.*; |
|||
import java.util.concurrent.atomic.LongAdder; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
public class RedisStreamConsumer { |
|||
|
|||
@Autowired |
|||
private StringRedisTemplate redisTemplate; |
|||
@Resource |
|||
private DefaultHttpRequestUtil defaultHttpRequestUtil; |
|||
@Resource |
|||
private DataCenterEnvConfig dataCenterEnvConfig; |
|||
|
|||
private static final int PARTITIONS = 16; |
|||
private static final String STREAM_PREFIX = "aeon_tcp_stream_"; |
|||
private static final String GROUP = "aeon_tcp_stream_consumer_group"; |
|||
private static final String DLQ_STREAM = "aeon_tcp_stream_dlq"; |
|||
|
|||
private static final int MAX_RETRY = 5; |
|||
private static final Duration IDLE_TIMEOUT = Duration.ofSeconds(30); |
|||
|
|||
/** |
|||
* 读取 Pending 消息和 reclaim 时 批量处理消息数 |
|||
*/ |
|||
private static final int BATCH_SIZE = 50; |
|||
|
|||
private volatile boolean running = true; |
|||
private String consumerName; |
|||
|
|||
/** |
|||
* 接收消息总数 |
|||
*/ |
|||
private final LongAdder receiveCounter = new LongAdder(); |
|||
|
|||
private final ExecutorService consumerExecutor = |
|||
Executors.newFixedThreadPool(PARTITIONS); |
|||
|
|||
private final ExecutorService workerExecutor = |
|||
new ThreadPoolExecutor( |
|||
16, |
|||
64, |
|||
60, |
|||
TimeUnit.SECONDS, |
|||
new LinkedBlockingQueue<>(10000), |
|||
new ThreadPoolExecutor.CallerRunsPolicy() |
|||
); |
|||
|
|||
/** |
|||
* 防止消息丢失或长时间未处理（死消息） |
|||
*/ |
|||
private final ScheduledExecutorService reclaimScheduler = |
|||
Executors.newSingleThreadScheduledExecutor(); |
|||
|
|||
/** |
|||
* 统计接收次数任务 |
|||
*/ |
|||
private final ScheduledExecutorService statsScheduler = |
|||
Executors.newSingleThreadScheduledExecutor(); |
|||
|
|||
@PostConstruct |
|||
public void start() throws Exception { |
|||
consumerName = buildConsumerName(); |
|||
log.info("RedisStreamConsumer start, consumer={}", consumerName); |
|||
|
|||
for (int i = 0; i < PARTITIONS; i++) { |
|||
String stream = STREAM_PREFIX + i; |
|||
|
|||
cleanConsumers(stream); |
|||
|
|||
createGroup(stream); |
|||
|
|||
int partition = i; |
|||
consumerExecutor.submit(() -> { |
|||
// 启动时处理未 ack 消息
|
|||
recoverPending(stream); |
|||
// 主消费循环
|
|||
consumeLoop(stream, partition); |
|||
}); |
|||
} |
|||
|
|||
// 定时 reclaim idle message
|
|||
reclaimScheduler.scheduleWithFixedDelay( |
|||
this::reclaimIdleMessages, |
|||
10, |
|||
10, |
|||
TimeUnit.SECONDS |
|||
); |
|||
|
|||
// 定时统计接收次数
|
|||
statsScheduler.scheduleAtFixedRate(() -> { |
|||
long count = receiveCounter.sum(); |
|||
log.info("RedisStreamConsumer receive count={}", count); |
|||
}, 10, 10, TimeUnit.SECONDS); |
|||
} |
|||
|
|||
private String buildConsumerName() throws Exception { |
|||
String host = InetAddress.getLocalHost().getHostName(); |
|||
return "consumer-" + host + "-" + UUID.randomUUID(); |
|||
} |
|||
|
|||
/** |
|||
* 启动时判定清理 consumer |
|||
*/ |
|||
private void cleanConsumers(String stream) { |
|||
try { |
|||
StreamInfo.XInfoConsumers consumers = |
|||
redisTemplate.opsForStream() |
|||
.consumers(stream, GROUP); |
|||
if (consumers == null) { |
|||
return; |
|||
} |
|||
consumers.forEach(c -> { |
|||
try { |
|||
if (c.idleTimeMs() > 86400000) { // 1天
|
|||
redisTemplate.opsForStream() |
|||
.deleteConsumer( |
|||
stream, |
|||
Consumer.from(GROUP, c.consumerName()) |
|||
); |
|||
log.info("startup clean consumer {}", c.consumerName()); |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("delete consumer error {}", c.consumerName(), e); |
|||
} |
|||
}); |
|||
} catch (Exception e) { |
|||
log.error("clean consumer error stream={}", stream, e); |
|||
} |
|||
} |
|||
|
|||
private void createGroup(String stream) { |
|||
try { |
|||
redisTemplate.opsForStream() |
|||
.createGroup(stream, ReadOffset.from("0-0"), GROUP); |
|||
log.info("create group success stream={}", stream); |
|||
} catch (Exception e) { |
|||
log.info("group exists stream={}", stream); |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* 启动时恢复未 ack 消息 |
|||
*/ |
|||
private void recoverPending(String stream) { |
|||
try { |
|||
// 分批次处理 Pending
|
|||
while (true) { |
|||
PendingMessages pending = redisTemplate.opsForStream() |
|||
.pending(stream, GROUP, Range.unbounded(), BATCH_SIZE); |
|||
|
|||
if (pending == null || pending.isEmpty()) break; |
|||
|
|||
for (PendingMessage msg : pending) { |
|||
// claim 未 ack 消息
|
|||
List<MapRecord<String, Object, Object>> claimed = |
|||
redisTemplate.opsForStream().claim( |
|||
stream, |
|||
GROUP, |
|||
consumerName, |
|||
IDLE_TIMEOUT, |
|||
msg.getId() |
|||
); |
|||
|
|||
for (MapRecord<String, Object, Object> record : claimed) { |
|||
receiveCounter.increment(); |
|||
workerExecutor.submit(() -> process(record, stream, msg.getTotalDeliveryCount())); |
|||
} |
|||
} |
|||
|
|||
// 如果少于 BATCH_SIZE，说明已处理完
|
|||
if (pending.size() < BATCH_SIZE) break; |
|||
} |
|||
|
|||
log.info("recoverPending stream={}", stream); |
|||
} catch (Exception e) { |
|||
log.error("recoverPending error stream={}", stream, e); |
|||
} |
|||
} |
|||
|
|||
private void consumeLoop(String stream, int partition) { |
|||
while (running) { |
|||
try { |
|||
List<MapRecord<String, Object, Object>> messages = |
|||
redisTemplate.opsForStream().read( |
|||
Consumer.from(GROUP, consumerName), |
|||
StreamReadOptions.empty() |
|||
.count(20) |
|||
.block(Duration.ofSeconds(1)), |
|||
StreamOffset.create(stream, ReadOffset.lastConsumed()) |
|||
); |
|||
|
|||
if (messages == null || messages.isEmpty()) continue; |
|||
|
|||
receiveCounter.add(messages.size()); |
|||
|
|||
for (MapRecord<String, Object, Object> message : messages) { |
|||
workerExecutor.submit(() -> process(message, stream, 1)); |
|||
} |
|||
|
|||
} catch (Exception e) { |
|||
log.error("consume error stream={}", stream, e); |
|||
sleep(200); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void process(MapRecord<String, Object, Object> message, |
|||
String stream, |
|||
long deliveryCount) { |
|||
|
|||
boolean success = false; |
|||
try { |
|||
Map<Object, Object> body = message.getValue(); |
|||
log.info("stream={}, id={}, body={}", |
|||
message.getStream(), |
|||
message.getId(), |
|||
body); |
|||
|
|||
/** |
|||
* ===== 业务逻辑 ===== |
|||
*/ |
|||
Gson gson = new Gson(); |
|||
String jsonParams = gson.toJson(body); |
|||
MyHTTPResponse response = this.defaultHttpRequestUtil.postJson(dataCenterEnvConfig.getF10ApiUrl(), jsonParams); |
|||
if (response.getCode() == 200) { |
|||
log.info("F10 data sent successfully...."); |
|||
} else { |
|||
log.error("F10 data sent failed...."); |
|||
} |
|||
|
|||
success = true; |
|||
} catch (Exception e) { |
|||
log.error("process error id={}", message.getId(), e); |
|||
} finally { |
|||
if (success) { |
|||
ack(stream, message); |
|||
} else { |
|||
// 不 ack，等待 reclaim
|
|||
if (deliveryCount >= MAX_RETRY) { |
|||
moveToDlq(stream, message); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void ack(String stream, MapRecord<String, Object, Object> message) { |
|||
try { |
|||
redisTemplate.opsForStream().acknowledge(stream, GROUP, message.getId()); |
|||
} catch (Exception e) { |
|||
log.error("ack error id={}", message.getId(), e); |
|||
} |
|||
} |
|||
|
|||
private void moveToDlq(String stream, MapRecord<String, Object, Object> message) { |
|||
try { |
|||
redisTemplate.opsForStream().add( |
|||
StreamRecords.mapBacked(message.getValue()) |
|||
.withStreamKey(DLQ_STREAM) |
|||
); |
|||
ack(stream, message); |
|||
log.error("move to DLQ :{}", message); |
|||
} catch (Exception e) { |
|||
log.error("dlq error :{}", message, e); |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* 定时 reclaim idle 消息 |
|||
* 防止消息被遗忘或丢失 |
|||
*/ |
|||
private void reclaimIdleMessages() { |
|||
for (int i = 0; i < PARTITIONS; i++) { |
|||
String stream = STREAM_PREFIX + i; |
|||
try { |
|||
while (true) { |
|||
PendingMessages pending = |
|||
redisTemplate.opsForStream().pending(stream, GROUP, Range.unbounded(), BATCH_SIZE); |
|||
|
|||
if (pending == null || pending.isEmpty()) break; |
|||
|
|||
for (PendingMessage p : pending) { |
|||
if (p.getElapsedTimeSinceLastDelivery().compareTo(IDLE_TIMEOUT) < 0) continue; |
|||
|
|||
List<MapRecord<String, Object, Object>> claimed = |
|||
redisTemplate.opsForStream().claim( |
|||
stream, |
|||
GROUP, |
|||
consumerName, |
|||
IDLE_TIMEOUT, |
|||
p.getId() |
|||
); |
|||
|
|||
for (MapRecord<String, Object, Object> r : claimed) { |
|||
receiveCounter.increment(); |
|||
workerExecutor.submit(() -> process(r, stream, p.getTotalDeliveryCount())); |
|||
} |
|||
} |
|||
|
|||
if (pending.size() < BATCH_SIZE) break; |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("reclaim error stream={}", stream, e); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public long getReceiveCount() { |
|||
return receiveCounter.sum(); |
|||
} |
|||
|
|||
private void sleep(long ms) { |
|||
try { |
|||
Thread.sleep(ms); |
|||
} catch (InterruptedException e) { |
|||
Thread.currentThread().interrupt(); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdown() throws InterruptedException { |
|||
running = false; |
|||
log.info("RedisStreamConsumer shutdown"); |
|||
|
|||
consumerExecutor.shutdown(); |
|||
workerExecutor.shutdown(); |
|||
reclaimScheduler.shutdown(); |
|||
statsScheduler.shutdown(); |
|||
|
|||
boolean consumerTerminated = consumerExecutor.awaitTermination(10, TimeUnit.SECONDS); |
|||
boolean workerTerminated = workerExecutor.awaitTermination(10, TimeUnit.SECONDS); |
|||
boolean reclaimTerminated = reclaimScheduler.awaitTermination(10, TimeUnit.SECONDS); |
|||
boolean statsTerminated = statsScheduler.awaitTermination(10, TimeUnit.SECONDS); |
|||
|
|||
log.info("RedisStreamConsumer shutdown done, consumerTerminated={}, workerTerminated={}, reclaimTerminated={}, statsTerminated={}", |
|||
consumerTerminated, |
|||
workerTerminated, |
|||
reclaimTerminated, |
|||
statsTerminated); |
|||
} |
|||
} |
|||
@ -0,0 +1,181 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj; |
|||
|
|||
import com.google.gson.Gson; |
|||
import com.techsor.datacenter.receiver.config.DataCenterEnvConfig; |
|||
import com.techsor.datacenter.receiver.listener.csdj.config.CsdjProperties; |
|||
import com.techsor.datacenter.receiver.listener.csdj.entity.CsdjEntity; |
|||
import com.techsor.datacenter.receiver.listener.csdj.session.CsdjSession; |
|||
import com.techsor.datacenter.receiver.listener.csdj.protocol.CsdjFrame; |
|||
import com.techsor.datacenter.receiver.utils.DefaultHttpRequestUtil; |
|||
import com.techsor.datacenter.receiver.utils.MyHTTPResponse; |
|||
import jakarta.annotation.PostConstruct; |
|||
import jakarta.annotation.PreDestroy; |
|||
import jakarta.annotation.Resource; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.io.IOException; |
|||
import java.net.ServerSocket; |
|||
import java.net.Socket; |
|||
import java.util.Arrays; |
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class CsdjServer { |
|||
|
|||
private final DefaultHttpRequestUtil defaultHttpRequestUtil; |
|||
|
|||
private final DataCenterEnvConfig dataCenterEnvConfig; |
|||
|
|||
private final CsdjProperties properties; |
|||
|
|||
private final RedisTemplate<String, Object> redisTemplate; |
|||
|
|||
private final ExecutorService executor = Executors.newCachedThreadPool(); |
|||
|
|||
// HTTP异步处理专用线程池
|
|||
private final ExecutorService httpExecutor = Executors.newCachedThreadPool(); |
|||
|
|||
@PostConstruct |
|||
public void start() { |
|||
if (!properties.getServer().isEnabled()) { |
|||
log.info("CSDJ Server is disabled"); |
|||
return; |
|||
} |
|||
|
|||
executor.submit(this::serverLoop); |
|||
} |
|||
|
|||
private void serverLoop() { |
|||
try (ServerSocket serverSocket = new java.net.ServerSocket(properties.getServer().getPort())) { |
|||
log.info("CSDJ Server started on port {}", properties.getServer().getPort()); |
|||
|
|||
while (!serverSocket.isClosed() && !Thread.currentThread().isInterrupted()) { |
|||
try { |
|||
Socket socket = serverSocket.accept(); |
|||
log.info("New connection from {}", socket.getInetAddress()); |
|||
executor.submit(() -> handleConnection(socket)); |
|||
} catch (IOException e) { |
|||
if (!serverSocket.isClosed()) { |
|||
log.error("Accept error", e); |
|||
} |
|||
} |
|||
} |
|||
} catch (IOException e) { |
|||
log.error("Server error", e); |
|||
} |
|||
} |
|||
|
|||
private void handleConnection(Socket socket) { |
|||
CsdjSession session = null; |
|||
try { |
|||
session = createSession(socket); |
|||
session.start(); |
|||
session.initiateAuthentication(); |
|||
|
|||
// 优雅等待：使用CountDownLatch阻塞直到连接关闭
|
|||
session.awaitClose(); |
|||
|
|||
} catch (InterruptedException e) { |
|||
log.info("Connection interrupted"); |
|||
Thread.currentThread().interrupt(); |
|||
} catch (Exception e) { |
|||
log.error("Connection error", e); |
|||
} finally { |
|||
if (session != null) { |
|||
session.close(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private CsdjSession createSession(Socket socket) throws IOException { |
|||
return new CsdjSession(socket, properties, true, redisTemplate) { |
|||
|
|||
@Override |
|||
protected void processInformation(CsdjFrame frame) { |
|||
|
|||
// ========== 打印设备上报数据 ==========
|
|||
printDeviceData(frame); |
|||
|
|||
} |
|||
|
|||
/** |
|||
* 打印设备上报数据 |
|||
*/ |
|||
private void printDeviceData(CsdjFrame frame) { |
|||
String terminalId = frame.getTerminalId(); |
|||
byte[] data = frame.getInfoPart(); |
|||
|
|||
log.info("========================================"); |
|||
log.info("Received device notification data"); |
|||
log.info("----------------------------------------"); |
|||
log.info("Terminal ID: {}", terminalId); |
|||
|
|||
if (data != null && data.length > 0) { |
|||
|
|||
String hexData = bytesToHex(data); |
|||
log.info("Data length: {} bytes", data.length); |
|||
log.info("Received Hex: {}", hexData); |
|||
log.info("Decimal array: {}", Arrays.toString(data)); |
|||
|
|||
httpExecutor.submit(() -> { |
|||
try { |
|||
//要用异步，不然会和这个csdj数据接收冲突阻塞
|
|||
sendDataAsync(terminalId, hexData); |
|||
} catch (Exception e) { |
|||
log.error("Failed to send data asynchronously", e); |
|||
} |
|||
}); |
|||
} else { |
|||
log.info("Data: empty"); |
|||
} |
|||
|
|||
log.info("========================================"); |
|||
} |
|||
}; |
|||
} |
|||
|
|||
private void sendDataAsync(String terminalId, String hexData) { |
|||
CsdjEntity csdjEntity = new CsdjEntity(); |
|||
csdjEntity.setTerminalId(terminalId); |
|||
csdjEntity.setData(hexData); |
|||
csdjEntity.setTs(System.currentTimeMillis()); |
|||
|
|||
Gson gson = new Gson(); |
|||
String jsonParams = gson.toJson(csdjEntity); |
|||
MyHTTPResponse response = defaultHttpRequestUtil.postJson(dataCenterEnvConfig.getCsdjApiUrl(), jsonParams); |
|||
if (response.getCode() == 200) { |
|||
log.info("csdj data sent successfully...."); |
|||
} else { |
|||
log.error("csdj data sent failed...."); |
|||
} |
|||
} |
|||
|
|||
|
|||
|
|||
/** |
|||
* 字节数组转十六进制字符串 |
|||
*/ |
|||
private String bytesToHex(byte[] bytes) { |
|||
StringBuilder sb = new StringBuilder(); |
|||
for (byte b : bytes) { |
|||
sb.append(String.format("%02X ", b)); |
|||
} |
|||
return sb.toString().trim(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void stop() { |
|||
log.info("正在关闭 CSDJ Server..."); |
|||
executor.shutdown(); |
|||
httpExecutor.shutdown(); |
|||
log.info("CSDJ Server 已关闭"); |
|||
} |
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.client; |
|||
|
|||
import com.techsor.datacenter.receiver.listener.csdj.config.CsdjProperties; |
|||
import com.techsor.datacenter.receiver.listener.csdj.protocol.CsdjFrame; |
|||
import com.techsor.datacenter.receiver.listener.csdj.session.CsdjSession; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.io.IOException; |
|||
import java.net.Socket; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class CsdjClient { |
|||
|
|||
private final CsdjProperties properties; |
|||
|
|||
private final RedisTemplate<String, Object> redisTemplate; |
|||
|
|||
public CsdjSession connect(String terminalId, String userId, String password) throws IOException { |
|||
if (!properties.getClient().isEnabled()) { |
|||
throw new IllegalStateException("CSDJ Client is disabled"); |
|||
} |
|||
|
|||
Socket socket = new Socket(properties.getClient().getHost(), properties.getClient().getPort()); |
|||
log.info("Connected to server: {}:{}", properties.getClient().getHost(), properties.getClient().getPort()); |
|||
|
|||
CsdjSession session = new CsdjSession(socket, properties, false, redisTemplate); |
|||
session.start(); |
|||
return session; |
|||
} |
|||
|
|||
public void sendNotification(CsdjSession session, String terminalId, byte[] data) { |
|||
CsdjFrame frame = CsdjFrame.createInfoFrame(terminalId, data, true, 0); |
|||
session.queueFrame(frame); |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.config; |
|||
|
|||
import lombok.Data; |
|||
import org.springframework.boot.context.properties.ConfigurationProperties; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
|
|||
@Data |
|||
@Component |
|||
@ConfigurationProperties(prefix = "csdj") |
|||
public class CsdjProperties { |
|||
private Server server = new Server(); |
|||
private Client client = new Client(); |
|||
private List<User> users = new ArrayList<>(); |
|||
private Timeout timeout = new Timeout(); |
|||
|
|||
@Data |
|||
public static class Server { |
|||
private int port = 7114; |
|||
private boolean enabled = true; |
|||
} |
|||
|
|||
@Data |
|||
public static class Client { |
|||
private boolean enabled = false; |
|||
private String host = "127.0.0.1"; |
|||
private int port = 7114; |
|||
} |
|||
|
|||
@Data |
|||
public static class User { |
|||
private String userId; |
|||
private String password; |
|||
} |
|||
|
|||
@Data |
|||
public static class Timeout { |
|||
private long ts0 = 30000; |
|||
private long ts1 = 10000; |
|||
private long ts2 = 10000; |
|||
private long ts3 = 10000; |
|||
private long ts4 = 10000; |
|||
private long tr1 = 10000; |
|||
private long tr2 = 10000; |
|||
private long tr4 = 10000; |
|||
private long tr5 = 10000; |
|||
} |
|||
} |
|||
@ -0,0 +1,12 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.entity; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
@Data |
|||
public class CsdjEntity implements Serializable { |
|||
private String terminalId; |
|||
private Long ts; |
|||
private String data; |
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.protocol; |
|||
|
|||
import lombok.Getter; |
|||
|
|||
@Getter |
|||
public enum ControlCommand { |
|||
INTEGRATED_VALUE_CLEAR((byte) 0x01, "積算値クリア要求"), |
|||
TERMINAL_STATUS_NOTIFY((byte) 0x02, "端子状態通知要求"), |
|||
TERMINAL_STATUS_NOTIFY_CSDJ((byte) 0x03, "端子状態通知要求(CSDJ用)"), |
|||
DIGITAL_OUTPUT_CONTROL((byte) 0x04, "デジタル出力制御要求"), |
|||
OUTPUT_TERMINAL_CONTROL((byte) 0x05, "出力端子制御要求"), |
|||
HISTORY_COLLECT_1((byte) 0x0A, "履歴収集要求1"), |
|||
HISTORY_COLLECT_1_OLD((byte) 0x0C, "履歴収集要求1(旧)"), |
|||
HISTORY_COLLECT_2((byte) 0x2A, "履歴収集要求2"), |
|||
HISTORY_COLLECT_2_OLD((byte) 0x2C, "履歴収集要求2(旧)"), |
|||
DATA_CONTROL_REQUEST((byte) 0x00, "データコントロール要求"), |
|||
TIME_INQUIRY((byte) 0x0E, "時刻問い合わせ要求"), |
|||
TIME_SETTING((byte) 0x0F, "時刻設定要求"), |
|||
TERMINAL_INFO((byte) 0x18, "端末情報要求"), |
|||
TERMINAL_INFO_OLD((byte) 0x11, "端末情報要求(旧)"), |
|||
RESET((byte) 0x19, "リセット要求(CSDJ用)"), |
|||
|
|||
INTEGRATED_VALUE_CLEAR_RESP((byte) 0x41, "積算値クリア通知"), |
|||
TERMINAL_STATUS_RESP((byte) 0x42, "端子状態通知"), |
|||
TERMINAL_STATUS_RESP_CSDJ((byte) 0x43, "端子状態通知(CSDJ用)"), |
|||
DIGITAL_OUTPUT_CONTROL_RESP((byte) 0x44, "デジタル出力制御完了通知"), |
|||
OUTPUT_TERMINAL_CONTROL_RESP((byte) 0x45, "出力端子制御完了通知"), |
|||
HISTORY_COLLECT_1_RESP((byte) 0x4A, "履歴収集応答1"), |
|||
HISTORY_COLLECT_1_RESP_OLD((byte) 0x4C, "履歴収集応答1(旧)"), |
|||
HISTORY_COLLECT_1_RESP_MEM_ERROR((byte) 0xCB, "履歴収集応答1(メモリ異常)"), |
|||
HISTORY_COLLECT_1_RESP_ERROR_OLD((byte) 0xCD, "履歴収集応答1(メモリ異常旧)"), |
|||
HISTORY_COLLECT_2_RESP((byte) 0x6A, "履歴収集応答2"), |
|||
HISTORY_COLLECT_2_RESP_OLD((byte) 0x6C, "履歴収集応答2(旧)"), |
|||
DATA_CONTROL_REQUEST_RESP((byte) 0x40, "データコントロール要求応答"), |
|||
TIME_INQUIRY_RESP((byte) 0x4E, "時刻問い合わせ応答"), |
|||
TIME_SETTING_RESP((byte) 0x4F, "時刻設定応答"), |
|||
TERMINAL_INFO_RESP((byte) 0x58, "端末情報応答"), |
|||
TERMINAL_INFO_RESP_OLD((byte) 0x51, "端末情報応答(旧)"), |
|||
TERMINAL_INFO_RESP_ERROR((byte) 0xD8, "端末情報応答(異常)"), |
|||
TERMINAL_INFO_RESP_ERROR_OLD((byte) 0xD1, "端末情報応答(異常旧)"), |
|||
RESET_RESP((byte) 0x59, "リセット応答(CSDJ用)"), |
|||
INVALID_COMMAND((byte) 0xFF, "不正コマンド／無応答通知"); |
|||
|
|||
private final byte code; |
|||
private final String description; |
|||
|
|||
ControlCommand(byte code, String description) { |
|||
this.code = code; |
|||
this.description = description; |
|||
} |
|||
|
|||
public static ControlCommand fromCode(byte code) { |
|||
for (ControlCommand cmd : values()) { |
|||
if (cmd.code == code) { |
|||
return cmd; |
|||
} |
|||
} |
|||
return INVALID_COMMAND; |
|||
} |
|||
} |
|||
@ -0,0 +1,46 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.protocol; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@Builder |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class ControlPart { |
|||
private FrameIdentifier identifier; |
|||
private boolean endBit; |
|||
private int sequenceNumber; |
|||
|
|||
public byte[] toBytes() { |
|||
byte[] bytes = new byte[2]; |
|||
bytes[0] = identifier.getCode(); |
|||
byte attr = (byte) ((endBit ? 0x80 : 0x00) | (sequenceNumber & 0x7F)); |
|||
bytes[1] = attr; |
|||
return bytes; |
|||
} |
|||
|
|||
public static ControlPart fromBytes(byte[] bytes) { |
|||
if (bytes.length != 2) { |
|||
throw new IllegalArgumentException("Control part must be 2 bytes"); |
|||
} |
|||
FrameIdentifier id = FrameIdentifier.fromCode(bytes[0]); |
|||
boolean endBit = (bytes[1] & 0x80) != 0; |
|||
int seq = bytes[1] & 0x7F; |
|||
return ControlPart.builder() |
|||
.identifier(id) |
|||
.endBit(endBit) |
|||
.sequenceNumber(seq) |
|||
.build(); |
|||
} |
|||
|
|||
public static ControlPart create(FrameIdentifier id) { |
|||
return ControlPart.builder() |
|||
.identifier(id) |
|||
.endBit(true) |
|||
.sequenceNumber(0) |
|||
.build(); |
|||
} |
|||
} |
|||
@ -0,0 +1,103 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.protocol; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.nio.ByteBuffer; |
|||
import java.nio.ByteOrder; |
|||
import java.nio.charset.StandardCharsets; |
|||
import java.util.Arrays; |
|||
|
|||
@Data |
|||
@Builder |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class CsdjFrame { |
|||
public static final int HEADER_LENGTH = 16; |
|||
public static final int MAX_FRAME_LENGTH = 6144; |
|||
public static final int MAX_INFO_LENGTH = 6128; |
|||
public static final int TERMINAL_ID_LENGTH = 12; |
|||
|
|||
private int dataLength; |
|||
private String terminalId; |
|||
private ControlPart controlPart; |
|||
private byte[] infoPart; |
|||
|
|||
public byte[] toBytes() { |
|||
int totalLength = HEADER_LENGTH + (infoPart != null ? infoPart.length : 0); |
|||
ByteBuffer buffer = ByteBuffer.allocate(totalLength); |
|||
buffer.order(ByteOrder.BIG_ENDIAN); |
|||
|
|||
buffer.putShort((short) totalLength); |
|||
|
|||
byte[] idBytes = new byte[TERMINAL_ID_LENGTH]; |
|||
if (terminalId != null) { |
|||
byte[] srcBytes = terminalId.getBytes(StandardCharsets.US_ASCII); |
|||
int len = Math.min(srcBytes.length, TERMINAL_ID_LENGTH); |
|||
System.arraycopy(srcBytes, 0, idBytes, 0, len); |
|||
} |
|||
buffer.put(idBytes); |
|||
|
|||
buffer.put(controlPart.toBytes()); |
|||
|
|||
if (infoPart != null && infoPart.length > 0) { |
|||
buffer.put(infoPart); |
|||
} |
|||
|
|||
return buffer.array(); |
|||
} |
|||
|
|||
public static CsdjFrame fromBytes(byte[] bytes) { |
|||
if (bytes.length < HEADER_LENGTH) { |
|||
throw new IllegalArgumentException("Frame too short"); |
|||
} |
|||
ByteBuffer buffer = ByteBuffer.wrap(bytes); |
|||
buffer.order(ByteOrder.BIG_ENDIAN); |
|||
|
|||
int dataLength = buffer.getShort() & 0xFFFF; |
|||
|
|||
byte[] idBytes = new byte[TERMINAL_ID_LENGTH]; |
|||
buffer.get(idBytes); |
|||
String terminalId = new String(idBytes, StandardCharsets.US_ASCII).trim(); |
|||
|
|||
byte[] ctrlBytes = new byte[2]; |
|||
buffer.get(ctrlBytes); |
|||
ControlPart controlPart = ControlPart.fromBytes(ctrlBytes); |
|||
|
|||
int infoLength = dataLength - HEADER_LENGTH; |
|||
byte[] infoPart = null; |
|||
if (infoLength > 0) { |
|||
infoPart = new byte[infoLength]; |
|||
buffer.get(infoPart); |
|||
} |
|||
|
|||
return CsdjFrame.builder() |
|||
.dataLength(dataLength) |
|||
.terminalId(terminalId) |
|||
.controlPart(controlPart) |
|||
.infoPart(infoPart) |
|||
.build(); |
|||
} |
|||
|
|||
public static CsdjFrame createSimpleFrame(FrameIdentifier id, String terminalId) { |
|||
return CsdjFrame.builder() |
|||
.terminalId(terminalId) |
|||
.controlPart(ControlPart.create(id)) |
|||
.infoPart(null) |
|||
.build(); |
|||
} |
|||
|
|||
public static CsdjFrame createInfoFrame(String terminalId, byte[] info, boolean endBit, int seq) { |
|||
return CsdjFrame.builder() |
|||
.terminalId(terminalId) |
|||
.controlPart(ControlPart.builder() |
|||
.identifier(FrameIdentifier.INFORMATION) |
|||
.endBit(endBit) |
|||
.sequenceNumber(seq) |
|||
.build()) |
|||
.infoPart(info) |
|||
.build(); |
|||
} |
|||
} |
|||
@ -0,0 +1,106 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.protocol; |
|||
|
|||
import lombok.Getter; |
|||
|
|||
/** |
|||
* 帧识别码枚举 |
|||
* 定义了CSDJ协议中所有帧类型的识别码 |
|||
*/ |
|||
@Getter |
|||
public enum FrameIdentifier { |
|||
/** |
|||
* 信息帧 (Information) |
|||
* 用于发送通報数据、数据控制命令等业务数据 |
|||
*/ |
|||
INFORMATION((byte) 0x01, "I", "信息（通報データ等の送信）"), |
|||
|
|||
/** |
|||
* 接收确认 (Receive Ready) |
|||
* 用于确认收到I帧 |
|||
*/ |
|||
RECEIVE_READY((byte) 0x81, "RR", "Iフレームの受信確認"), |
|||
|
|||
/** |
|||
* 接收未就绪 (Receive Not Ready) |
|||
* 用于要求对方重发I帧（例如校验错误或丢包） |
|||
*/ |
|||
RECEIVE_NOT_READY((byte) 0x82, "RNR", "Iフレームの再送要求"), |
|||
|
|||
/** |
|||
* 拒绝 (Reject) |
|||
* 用于强制终止通信 |
|||
*/ |
|||
REJECT((byte) 0x83, "REJ", "通信強制終了"), |
|||
|
|||
/** |
|||
* ID/密码请求 (ID/PassWord) |
|||
* 服务器端发送，要求终端发送认证信息 |
|||
*/ |
|||
ID_PASSWORD((byte) 0xC1, "IDPWD", "ユーザID／パスワード要求"), |
|||
|
|||
/** |
|||
* ID/密码确认 (ID/PassWord-Unnumbered Acknowledge) |
|||
* 终端发送，包含用户ID和密码进行认证 |
|||
*/ |
|||
ID_PASSWORD_ACK((byte) 0xC2, "IDPWD-UA", "ユーザID／パスワードの送信"), |
|||
|
|||
/** |
|||
* 新消息发送准备 (New Message Send Ready) |
|||
* 表示"我准备好接收数据了"或"我发完了，轮到你了" |
|||
*/ |
|||
NEW_MESSAGE_SEND_READY((byte) 0xC3, "NMSR", "新規メッセージ送信確認"), |
|||
|
|||
/** |
|||
* 断开请求 (DisConnect) |
|||
* 请求断开连接 |
|||
*/ |
|||
DISCONNECT((byte) 0xC4, "DISC", "切断要求"), |
|||
|
|||
/** |
|||
* 断开确认 (DisConnect-Unnumbered Acknowledge) |
|||
* 确认断开连接 |
|||
*/ |
|||
DISCONNECT_ACK((byte) 0xC5, "DISC-UA", "(DISCに対する)確認応答"); |
|||
|
|||
/** |
|||
* 识别码（1字节） |
|||
*/ |
|||
private final byte code; |
|||
|
|||
/** |
|||
* 简称（如I、RR等） |
|||
*/ |
|||
private final String name; |
|||
|
|||
/** |
|||
* 详细描述 |
|||
*/ |
|||
private final String description; |
|||
|
|||
/** |
|||
* 构造函数 |
|||
* @param code 识别码字节 |
|||
* @param name 简称 |
|||
* @param description 详细描述 |
|||
*/ |
|||
FrameIdentifier(byte code, String name, String description) { |
|||
this.code = code; |
|||
this.name = name; |
|||
this.description = description; |
|||
} |
|||
|
|||
/** |
|||
* 根据字节码查找对应的识别码枚举 |
|||
* @param code 识别码字节 |
|||
* @return 对应的帧识别码枚举 |
|||
* @throws IllegalArgumentException 如果找不到匹配的识别码 |
|||
*/ |
|||
public static FrameIdentifier fromCode(byte code) { |
|||
for (FrameIdentifier id : values()) { |
|||
if (id.code == code) { |
|||
return id; |
|||
} |
|||
} |
|||
throw new IllegalArgumentException("Unknown frame identifier: " + code); |
|||
} |
|||
} |
|||
@ -0,0 +1,54 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.protocol; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.nio.ByteBuffer; |
|||
import java.nio.charset.StandardCharsets; |
|||
|
|||
@Data |
|||
@Builder |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class IdPasswordAck { |
|||
private String userId; |
|||
private String password; |
|||
|
|||
public byte[] toBytes() { |
|||
ByteBuffer buffer = ByteBuffer.allocate(32); |
|||
putString(buffer, userId, 16); |
|||
putString(buffer, password, 16); |
|||
return buffer.array(); |
|||
} |
|||
|
|||
public static IdPasswordAck fromBytes(byte[] bytes) { |
|||
if (bytes.length < 32) { |
|||
throw new IllegalArgumentException("ID/PWD ACK must be 32 bytes"); |
|||
} |
|||
ByteBuffer buffer = ByteBuffer.wrap(bytes); |
|||
String userId = getString(buffer, 16); |
|||
String password = getString(buffer, 16); |
|||
return IdPasswordAck.builder() |
|||
.userId(userId) |
|||
.password(password) |
|||
.build(); |
|||
} |
|||
|
|||
private static void putString(ByteBuffer buffer, String str, int length) { |
|||
byte[] bytes = new byte[length]; |
|||
if (str != null) { |
|||
byte[] src = str.getBytes(StandardCharsets.US_ASCII); |
|||
int len = Math.min(src.length, length); |
|||
System.arraycopy(src, 0, bytes, 0, len); |
|||
} |
|||
buffer.put(bytes); |
|||
} |
|||
|
|||
private static String getString(ByteBuffer buffer, int length) { |
|||
byte[] bytes = new byte[length]; |
|||
buffer.get(bytes); |
|||
return new String(bytes, StandardCharsets.US_ASCII).trim(); |
|||
} |
|||
} |
|||
@ -0,0 +1,336 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.session; |
|||
|
|||
import com.alibaba.fastjson2.JSONObject; |
|||
import com.techsor.datacenter.receiver.listener.csdj.config.CsdjProperties; |
|||
import com.techsor.datacenter.receiver.listener.csdj.protocol.*; |
|||
import com.techsor.datacenter.receiver.utils.RedisUtils; |
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.data.redis.core.RedisTemplate; |
|||
|
|||
import java.io.IOException; |
|||
import java.io.InputStream; |
|||
import java.io.OutputStream; |
|||
import java.net.Socket; |
|||
import java.nio.ByteBuffer; |
|||
import java.nio.ByteOrder; |
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.concurrent.*; |
|||
|
|||
@Slf4j |
|||
public class CsdjSession implements AutoCloseable { |
|||
|
|||
private static final String AP_GATEWAY_AUTH_KEY = "ap_gateway_auth"; |
|||
|
|||
private final Socket socket; |
|||
private final InputStream input; |
|||
private final OutputStream output; |
|||
private final CsdjProperties properties; |
|||
private final boolean isServer; |
|||
private final List<CsdjFrame> pendingFrames = new ArrayList<>(); |
|||
|
|||
private final RedisTemplate<String, Object> redisTemplate; |
|||
|
|||
// 认证失败计数器，最多重发 2 次 IDPWD
|
|||
private int authFailCount = 0; |
|||
|
|||
@Getter |
|||
private SessionState state = SessionState.INIT; |
|||
@Getter |
|||
private String terminalId = ""; |
|||
@Getter |
|||
private boolean authenticated = false; |
|||
|
|||
private final ExecutorService executor = Executors.newSingleThreadExecutor(); |
|||
private Future<?> readFuture; |
|||
private volatile boolean active = true; |
|||
private final CountDownLatch closeLatch = new CountDownLatch(1); |
|||
|
|||
public boolean isActive() { |
|||
return active && !socket.isClosed() && state != SessionState.CLOSED; |
|||
} |
|||
|
|||
public void awaitClose() throws InterruptedException { |
|||
closeLatch.await(); |
|||
} |
|||
|
|||
public boolean awaitClose(long timeout, TimeUnit unit) throws InterruptedException { |
|||
return closeLatch.await(timeout, unit); |
|||
} |
|||
|
|||
public CsdjSession(Socket socket, CsdjProperties properties, boolean isServer, RedisTemplate<String, Object> redisTemplate) throws IOException { |
|||
this.socket = socket; |
|||
this.input = socket.getInputStream(); |
|||
this.output = socket.getOutputStream(); |
|||
this.properties = properties; |
|||
this.isServer = isServer; |
|||
this.redisTemplate = redisTemplate; |
|||
} |
|||
|
|||
public void start() { |
|||
readFuture = executor.submit(this::readLoop); |
|||
} |
|||
|
|||
private void readLoop() { |
|||
try { |
|||
while (!socket.isClosed() && !Thread.currentThread().isInterrupted()) { |
|||
CsdjFrame frame = readFrame(); |
|||
if (frame != null) { |
|||
handleFrame(frame); |
|||
} |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("Read loop error", e); |
|||
} finally { |
|||
close(); |
|||
} |
|||
} |
|||
|
|||
private CsdjFrame readFrame() throws IOException { |
|||
byte[] header = new byte[CsdjFrame.HEADER_LENGTH]; |
|||
int read = input.read(header); |
|||
if (read != CsdjFrame.HEADER_LENGTH) { |
|||
return null; |
|||
} |
|||
|
|||
ByteBuffer buffer = ByteBuffer.wrap(header); |
|||
buffer.order(ByteOrder.BIG_ENDIAN); |
|||
int dataLength = buffer.getShort() & 0xFFFF; |
|||
|
|||
int infoLength = dataLength - CsdjFrame.HEADER_LENGTH; |
|||
byte[] fullFrame = new byte[dataLength]; |
|||
System.arraycopy(header, 0, fullFrame, 0, CsdjFrame.HEADER_LENGTH); |
|||
|
|||
if (infoLength > 0) { |
|||
read = input.read(fullFrame, CsdjFrame.HEADER_LENGTH, infoLength); |
|||
if (read != infoLength) { |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
return CsdjFrame.fromBytes(fullFrame); |
|||
} |
|||
|
|||
private void handleFrame(CsdjFrame frame) { |
|||
log.debug("Received frame: {}", frame.getControlPart().getIdentifier().getName()); |
|||
|
|||
FrameIdentifier id = frame.getControlPart().getIdentifier(); |
|||
|
|||
switch (id) { |
|||
case ID_PASSWORD: |
|||
handleIdPassword(frame); |
|||
break; |
|||
case ID_PASSWORD_ACK: |
|||
handleIdPasswordAck(frame); |
|||
break; |
|||
case NEW_MESSAGE_SEND_READY: |
|||
handleNmsr(frame); |
|||
break; |
|||
case INFORMATION: |
|||
handleInformation(frame); |
|||
break; |
|||
case RECEIVE_READY: |
|||
handleRr(frame); |
|||
break; |
|||
case DISCONNECT: |
|||
handleDisc(frame); |
|||
break; |
|||
case DISCONNECT_ACK: |
|||
handleDiscAck(frame); |
|||
break; |
|||
default: |
|||
log.warn("Unknown frame: {}", id); |
|||
} |
|||
} |
|||
|
|||
private void handleIdPassword(CsdjFrame frame) { |
|||
if (isServer) { |
|||
state = SessionState.WAITING_IDPWD_UA; |
|||
} |
|||
} |
|||
|
|||
private void handleIdPasswordAck(CsdjFrame frame) { |
|||
if (isServer) { |
|||
this.terminalId = frame.getTerminalId(); |
|||
IdPasswordAck auth = IdPasswordAck.fromBytes(frame.getInfoPart()); |
|||
authenticated = authenticate(auth.getUserId(), auth.getPassword(), terminalId); |
|||
if (authenticated) { |
|||
log.info("Authentication success: {}", auth.getUserId()); |
|||
|
|||
authFailCount = 0; // 重置计数器
|
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.NEW_MESSAGE_SEND_READY, terminalId)); |
|||
state = SessionState.WAITING_I; |
|||
} else { |
|||
log.warn("Authentication failed: {}, fail count: {}", auth.getUserId(), authFailCount); |
|||
// 文档 ： CSDJ データ通信仕様書_20231114.pdf
|
|||
// 页码 ：第 25-26页
|
|||
// 规定-认证失败后，最多重发 2 次 IDPWD，2 次重发后，发 DISC 断开连接
|
|||
authFailCount++; |
|||
if (authFailCount <= 2) { |
|||
log.info("Re-sending IDPWD, attempt: {}", authFailCount); |
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.ID_PASSWORD, terminalId)); |
|||
state = SessionState.WAITING_IDPWD_UA; |
|||
} else { |
|||
// 协议规定：2 回再送後 DISC 送信
|
|||
log.warn("Max auth fail attempts (2) reached, sending DISC and closing"); |
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.DISCONNECT, terminalId)); |
|||
state = SessionState.CLOSING; |
|||
} |
|||
} |
|||
} else { |
|||
state = SessionState.WAITING_NMSR; |
|||
} |
|||
} |
|||
|
|||
private boolean authenticate(String userId, String password, String terminalId) { |
|||
// return properties.getUsers().stream()
|
|||
// .anyMatch(u -> u.getUserId().equals(userId) && u.getPassword().equals(password));
|
|||
try { |
|||
// 从 Redis Hash "ap_gateway_auth" 查询认证信息
|
|||
// Hash key: "ap_gateway_auth"
|
|||
// Hash field: terminalId (就是 imei)
|
|||
Object authJson = redisTemplate.opsForHash().get(AP_GATEWAY_AUTH_KEY, terminalId); |
|||
|
|||
if (authJson == null) { |
|||
log.warn("未在 Redis 找到终端ID的认证信息: {}", terminalId); |
|||
return false; |
|||
} |
|||
|
|||
// 解析 JSON，获取 authUserId 和 authPwd
|
|||
JSONObject authObj = JSONObject.parseObject(authJson.toString()); |
|||
String redisUserId = authObj.getString("authUserId"); |
|||
String redisPassword = authObj.getString("authPwd"); |
|||
|
|||
// 核对账号密码
|
|||
if (userId.equals(redisUserId) && password.equals(redisPassword)) { |
|||
log.info("Redis 认证成功: 终端ID: {}, 用户: {}", terminalId, userId); |
|||
return true; |
|||
} else { |
|||
log.warn("Redis 认证失败: 终端ID: {}, 用户: {}, 密码不匹配", terminalId, userId); |
|||
return false; |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("Redis 认证异常: 终端ID: {}, 错误: {}", terminalId, e.getMessage(), e); |
|||
// 异常时尝试使用配置里的认证信息
|
|||
return properties.getUsers().stream() |
|||
.anyMatch(u -> u.getUserId().equals(userId) && u.getPassword().equals(password)); |
|||
} |
|||
} |
|||
|
|||
private void handleNmsr(CsdjFrame frame) { |
|||
log.info("Received NMSR"); |
|||
|
|||
if (!pendingFrames.isEmpty()) { |
|||
sendPendingFrame(); |
|||
} else if (isServer) { |
|||
// 服务器端，没有数据要发，直接断开
|
|||
log.info("No data to send, sending DISC to disconnect."); |
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.DISCONNECT, terminalId)); |
|||
state = SessionState.CLOSING; |
|||
} else { |
|||
state = SessionState.WAITING_I; |
|||
} |
|||
} |
|||
|
|||
private void handleInformation(CsdjFrame frame) { |
|||
log.info("Received information frame from terminal: {}", frame.getTerminalId()); |
|||
this.terminalId = frame.getTerminalId(); |
|||
|
|||
// 1. 先发送 RR 确认收到
|
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.RECEIVE_READY, terminalId)); |
|||
|
|||
// 2. 先设置状态为等待 NMSR
|
|||
state = SessionState.WAITING_I; |
|||
|
|||
// 3. 再处理业务数据
|
|||
processInformation(frame); |
|||
} |
|||
|
|||
protected void processInformation(CsdjFrame frame) { |
|||
log.info("Processing information: {} bytes", frame.getInfoPart() != null ? frame.getInfoPart().length : 0); |
|||
} |
|||
|
|||
private void handleRr(CsdjFrame frame) { |
|||
if (!pendingFrames.isEmpty()) { |
|||
pendingFrames.remove(0); |
|||
if (!pendingFrames.isEmpty()) { |
|||
sendPendingFrame(); |
|||
} else { |
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.NEW_MESSAGE_SEND_READY, terminalId)); |
|||
state = SessionState.WAITING_I; |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void handleDisc(CsdjFrame frame) { |
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.DISCONNECT_ACK, terminalId)); |
|||
state = SessionState.CLOSED; |
|||
} |
|||
|
|||
private void handleDiscAck(CsdjFrame frame) { |
|||
state = SessionState.CLOSED; |
|||
log.info("Received disc ack, disconnected..."); |
|||
close(); |
|||
} |
|||
|
|||
private void sendPendingFrame() { |
|||
if (!pendingFrames.isEmpty()) { |
|||
sendFrame(pendingFrames.get(0)); |
|||
state = SessionState.WAITING_RR; |
|||
} |
|||
} |
|||
|
|||
public void sendFrame(CsdjFrame frame) { |
|||
try { |
|||
byte[] bytes = frame.toBytes(); |
|||
output.write(bytes); |
|||
output.flush(); |
|||
log.debug("Sent frame: {}", frame.getControlPart().getIdentifier().getName()); |
|||
} catch (IOException e) { |
|||
log.error("Send frame error", e); |
|||
close(); |
|||
} |
|||
} |
|||
|
|||
public void queueFrame(CsdjFrame frame) { |
|||
pendingFrames.add(frame); |
|||
} |
|||
|
|||
public void initiateAuthentication() { |
|||
sendFrame(CsdjFrame.createSimpleFrame(FrameIdentifier.ID_PASSWORD, terminalId)); |
|||
state = SessionState.WAITING_IDPWD_UA; |
|||
} |
|||
|
|||
public void sendAuthentication(String userId, String password) { |
|||
IdPasswordAck auth = IdPasswordAck.builder() |
|||
.userId(userId) |
|||
.password(password) |
|||
.build(); |
|||
CsdjFrame frame = CsdjFrame.builder() |
|||
.terminalId(terminalId) |
|||
.controlPart(ControlPart.create(FrameIdentifier.ID_PASSWORD_ACK)) |
|||
.infoPart(auth.toBytes()) |
|||
.build(); |
|||
sendFrame(frame); |
|||
} |
|||
|
|||
@Override |
|||
public void close() { |
|||
if (!active) { |
|||
return; // 避免重复关闭
|
|||
} |
|||
active = false; |
|||
state = SessionState.CLOSED; |
|||
if (readFuture != null) { |
|||
readFuture.cancel(true); |
|||
} |
|||
try { |
|||
socket.close(); |
|||
} catch (IOException e) { |
|||
log.error("Close socket error", e); |
|||
} |
|||
executor.shutdown(); |
|||
closeLatch.countDown(); // 释放锁，唤醒awaitClose()
|
|||
} |
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
package com.techsor.datacenter.receiver.listener.csdj.session; |
|||
|
|||
/** |
|||
* CSDJ会话状态枚举 |
|||
*/ |
|||
public enum SessionState { |
|||
/** |
|||
* 初始化状态 |
|||
* - 会话刚建立，准备开始 |
|||
*/ |
|||
INIT, |
|||
|
|||
/** |
|||
* 等待IDPWD-UA |
|||
* - 服务器发IDPWD，等设备回应IDPWD-UA |
|||
*/ |
|||
WAITING_IDPWD_UA, |
|||
|
|||
/** |
|||
* 等待NMSR |
|||
* - 发送NMSR，等对方回应NMSR或I帧 |
|||
*/ |
|||
WAITING_NMSR, |
|||
|
|||
/** |
|||
* 等待I帧 |
|||
* - 等待设备发通報数据或响应 |
|||
*/ |
|||
WAITING_I, |
|||
|
|||
/** |
|||
* 等待RR |
|||
* - 发送了I帧，等RR确认 |
|||
*/ |
|||
WAITING_RR, |
|||
|
|||
/** |
|||
* 等待响应 |
|||
* - 发送了数据控制命令，等设备响应 |
|||
*/ |
|||
WAITING_RESPONSE, |
|||
|
|||
/** |
|||
* 发送中 |
|||
* - 正在发送帧中 |
|||
*/ |
|||
SENDING, |
|||
|
|||
/** |
|||
* 关闭中 |
|||
* - 发送了DISC，等DISC-UA |
|||
*/ |
|||
CLOSING, |
|||
|
|||
/** |
|||
* 已关闭 |
|||
* - 会话结束 |
|||
*/ |
|||
CLOSED |
|||
} |
|||
@ -0,0 +1,369 @@ |
|||
package com.techsor.datacenter; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.io.IOException; |
|||
import java.io.InputStream; |
|||
import java.io.OutputStream; |
|||
import java.net.Socket; |
|||
import java.nio.ByteBuffer; |
|||
import java.nio.ByteOrder; |
|||
import java.util.Arrays; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* CSDJ设备模拟器 - 带状态机校验 |
|||
*/ |
|||
@Slf4j |
|||
public class CsdjClientSimulator { |
|||
|
|||
// 配置
|
|||
private static final String SERVER_HOST = "127.0.0.1"; |
|||
private static final int SERVER_PORT = 7114; |
|||
private static final String TERMINAL_ID = "TEST002"; |
|||
private static final String USER_ID = "user11"; |
|||
private static final String PASSWORD = "pwd1"; |
|||
|
|||
// 帧类型定义
|
|||
private static final byte FRAME_ID_PASSWORD = (byte) 0xC1; |
|||
private static final byte FRAME_ID_PASSWORD_UA = (byte) 0xC2; |
|||
private static final byte FRAME_NEW_MESSAGE_SEND_READY = (byte) 0xC3; |
|||
private static final byte FRAME_INFORMATION = (byte) 0x01; |
|||
private static final byte FRAME_RECEIVE_READY = (byte) 0x81; |
|||
private static final byte FRAME_DISCONNECT = (byte) 0xC4; |
|||
private static final byte FRAME_DISCONNECT_UA = (byte) 0xC5; |
|||
|
|||
// 状态机
|
|||
private enum State { |
|||
WAIT_IDPWD, |
|||
WAIT_NMSR1, |
|||
WAIT_RR, |
|||
WAIT_DISC, |
|||
WAIT_SERVER_CLOSE, |
|||
FINISHED |
|||
} |
|||
|
|||
public static void main(String[] args) throws InterruptedException { |
|||
log.info("CSDJ设备模拟器启动..."); |
|||
log.info("连接服务器: {}:{}", SERVER_HOST, SERVER_PORT); |
|||
|
|||
Socket socket = null; |
|||
try { |
|||
socket = new Socket(SERVER_HOST, SERVER_PORT); |
|||
log.info("连接成功！"); |
|||
|
|||
InputStream input = socket.getInputStream(); |
|||
OutputStream output = socket.getOutputStream(); |
|||
|
|||
State state = State.WAIT_IDPWD; |
|||
int authAttemptCount = 0; |
|||
|
|||
while (state != State.FINISHED && !socket.isClosed()) { |
|||
switch (state) { |
|||
case WAIT_IDPWD: |
|||
// ========== 步骤1: 等待 IDPWD ==========
|
|||
log.info("---------- 步骤1: 等待 IDPWD ----------"); |
|||
CsdjFrame idpwdFrame = expectFrame(input, FRAME_ID_PASSWORD); |
|||
log.info("✅ 收到: IDPWD (0xC1)"); |
|||
logFrame(idpwdFrame); |
|||
state = State.WAIT_NMSR1; |
|||
|
|||
// ========== 步骤2: 发送 IDPWD-UA ==========
|
|||
log.info("---------- 步骤2: 发送 IDPWD-UA ----------"); |
|||
CsdjFrame idpwdUaFrame = createIdpwdUaFrame(USER_ID, PASSWORD); |
|||
sendFrame(output, idpwdUaFrame); |
|||
log.info("✅ 已发送: IDPWD-UA (0xC2)"); |
|||
logFrame(idpwdUaFrame); |
|||
authAttemptCount++; |
|||
break; |
|||
|
|||
case WAIT_NMSR1: |
|||
// ========== 步骤3: 等待 NMSR 或者 IDPWD(重发) 或者 DISC ==========
|
|||
log.info("---------- 步骤3: 等待 NMSR 或 IDPWD 或 DISC ----------"); |
|||
// 先读取帧，再判断类型
|
|||
CsdjFrame frame = readFrame(input); |
|||
|
|||
if (frame.controlId == FRAME_NEW_MESSAGE_SEND_READY) { |
|||
// 收到 NMSR，认证成功，继续正常流程
|
|||
log.info("✅ 收到: NMSR (0xC3) - 认证成功"); |
|||
logFrame(frame); |
|||
state = State.WAIT_RR; |
|||
continueNormalFlow(input, output, socket); |
|||
state = State.WAIT_SERVER_CLOSE; |
|||
} else if (frame.controlId == FRAME_ID_PASSWORD) { |
|||
// 收到 IDPWD，服务器在重发认证请求，我们也要重发 IDPWD-UA
|
|||
log.info("🔄 收到: IDPWD (0xC1) - 服务器在重发认证请求 (attempt {})", authAttemptCount); |
|||
logFrame(frame); |
|||
log.info("---------- 重发 IDPWD-UA ----------"); |
|||
CsdjFrame idpwdUaFrame2 = createIdpwdUaFrame(USER_ID, PASSWORD); |
|||
sendFrame(output, idpwdUaFrame2); |
|||
log.info("✅ 已重发: IDPWD-UA (0xC2)"); |
|||
logFrame(idpwdUaFrame2); |
|||
authAttemptCount++; |
|||
state = State.WAIT_NMSR1; |
|||
} else if (frame.controlId == FRAME_DISCONNECT) { |
|||
// 收到 DISC，服务器要断开
|
|||
log.info("❌ 收到: DISC (0xC4) - 服务器要断开连接 (失败 {} 次后)", authAttemptCount); |
|||
logFrame(frame); |
|||
log.info("发送 DISC-UA 确认断开"); |
|||
CsdjFrame discUaFrame = createSimpleFrame(FRAME_DISCONNECT_UA); |
|||
sendFrame(output, discUaFrame); |
|||
log.info("✅ 已发送: DISC-UA (0xC5)"); |
|||
logFrame(discUaFrame); |
|||
state = State.WAIT_SERVER_CLOSE; |
|||
} else { |
|||
String receivedName = getFrameName(frame.controlId); |
|||
throw new IOException(String.format("❌ 帧类型错误！期望: NMSR/IDPWD/DISC, 实际: %s(0x%02X)", |
|||
receivedName, frame.controlId & 0xFF)); |
|||
} |
|||
break; |
|||
|
|||
case WAIT_SERVER_CLOSE: |
|||
// ========== 等待服务器关闭连接 ==========
|
|||
log.info("---------- 等待服务器关闭连接 ----------"); |
|||
// 继续读取，直到连接关闭
|
|||
CsdjFrame waitFrame = readFrame(input); |
|||
// 如果读到了帧，可能是服务器又发了什么，继续处理
|
|||
if (waitFrame != null) { |
|||
log.info("收到帧: {}", getFrameName(waitFrame.controlId)); |
|||
logFrame(waitFrame); |
|||
// 如果收到 DISC，再次回复 DISC-UA
|
|||
if (waitFrame.controlId == FRAME_DISCONNECT) { |
|||
log.info("发送 DISC-UA 确认断开"); |
|||
CsdjFrame discUaFrame2 = createSimpleFrame(FRAME_DISCONNECT_UA); |
|||
sendFrame(output, discUaFrame2); |
|||
log.info("✅ 已发送: DISC-UA (0xC5)"); |
|||
logFrame(discUaFrame2); |
|||
} |
|||
} |
|||
break; |
|||
|
|||
default: |
|||
log.warn("Unknown state: {}", state); |
|||
state = State.FINISHED; |
|||
break; |
|||
} |
|||
} |
|||
|
|||
if (state == State.FINISHED) { |
|||
log.info("========================================"); |
|||
log.info("✅ 流程结束！"); |
|||
log.info("========================================"); |
|||
} |
|||
|
|||
} catch (IOException e) { |
|||
log.error("❌ 连接错误: {}", e.getMessage(), e); |
|||
} |
|||
TimeUnit.SECONDS.sleep(1); |
|||
} |
|||
|
|||
/** |
|||
* 正常流程（认证成功后的流程） |
|||
*/ |
|||
private static void continueNormalFlow(InputStream input, OutputStream output, Socket socket) throws IOException { |
|||
State state = State.WAIT_RR; |
|||
|
|||
// ========== 步骤4: 发送 I帧（通報数据）==========
|
|||
log.info("---------- 步骤4: 发送通報数据 ----------"); |
|||
byte[] sensorData = createMockSensorData(); |
|||
CsdjFrame iFrame = createInfoFrame(sensorData); |
|||
sendFrame(output, iFrame); |
|||
log.info("✅ 已发送: I帧 (0x01)"); |
|||
logFrame(iFrame); |
|||
log.info("通報数据: {}", bytesToHex(sensorData)); |
|||
|
|||
// ========== 步骤5: 等待 RR ==========
|
|||
log.info("---------- 步骤5: 等待 RR ----------"); |
|||
CsdjFrame rrFrame = expectFrame(input, FRAME_RECEIVE_READY); |
|||
log.info("✅ 收到: RR (0x81)"); |
|||
logFrame(rrFrame); |
|||
state = State.WAIT_DISC; |
|||
|
|||
// ========== 步骤6: 发送 NMSR ==========
|
|||
log.info("---------- 步骤6: 发送 NMSR ----------"); |
|||
CsdjFrame nmsrFrame2 = createSimpleFrame(FRAME_NEW_MESSAGE_SEND_READY); |
|||
sendFrame(output, nmsrFrame2); |
|||
log.info("✅ 已发送: NMSR (0xC3)"); |
|||
logFrame(nmsrFrame2); |
|||
|
|||
// ========== 步骤7: 等待 DISC ==========
|
|||
log.info("---------- 步骤7: 等待 DISC ----------"); |
|||
CsdjFrame discFrame = expectFrame(input, FRAME_DISCONNECT); |
|||
log.info("✅ 收到: DISC (0xC4)"); |
|||
logFrame(discFrame); |
|||
|
|||
// ========== 步骤8: 发送 DISC-UA ==========
|
|||
log.info("---------- 步骤8: 发送 DISC-UA ----------"); |
|||
CsdjFrame discUaFrame = createSimpleFrame(FRAME_DISCONNECT_UA); |
|||
sendFrame(output, discUaFrame); |
|||
log.info("✅ 已发送: DISC-UA (0xC5)"); |
|||
logFrame(discUaFrame); |
|||
|
|||
// 不设置 FINISHED，让主循环继续等待服务器关闭 socket
|
|||
} |
|||
|
|||
// ==================== 状态机校验帧 ====================
|
|||
private static CsdjFrame expectFrame(InputStream input, byte expectedFrameId) throws IOException { |
|||
CsdjFrame frame = readFrame(input); |
|||
if (frame.controlId != expectedFrameId) { |
|||
String receivedName = getFrameName(frame.controlId); |
|||
String expectedName = getFrameName(expectedFrameId); |
|||
throw new IOException(String.format("❌ 帧类型错误！期望: %s(0x%02X), 实际: %s(0x%02X)", |
|||
expectedName, expectedFrameId & 0xFF, receivedName, frame.controlId & 0xFF)); |
|||
} |
|||
return frame; |
|||
} |
|||
|
|||
// ==================== 帧处理方法 ====================
|
|||
private static CsdjFrame readFrame(InputStream input) throws IOException { |
|||
// 先读头部 (2字节长度 + 12字节终端ID + 2字节控制部 = 16字节)
|
|||
byte[] header = new byte[16]; |
|||
int readLen = input.read(header); |
|||
if (readLen != 16) { |
|||
throw new IOException("读取头部失败: " + readLen + " != 16"); |
|||
} |
|||
|
|||
// 解析长度
|
|||
ByteBuffer buffer = ByteBuffer.wrap(header); |
|||
buffer.order(ByteOrder.BIG_ENDIAN); |
|||
int length = buffer.getShort() & 0xFFFF; |
|||
|
|||
// 读信息部
|
|||
int infoLen = length - 16; |
|||
byte[] info = null; |
|||
if (infoLen > 0) { |
|||
info = new byte[infoLen]; |
|||
readLen = input.read(info); |
|||
if (readLen != infoLen) { throw new IOException("读取信息部失败: " + readLen + " != " + infoLen); |
|||
} |
|||
} |
|||
|
|||
return new CsdjFrame(header, info); |
|||
} |
|||
|
|||
private static void sendFrame(OutputStream output, CsdjFrame frame) throws IOException { |
|||
output.write(frame.toBytes()); |
|||
output.flush(); |
|||
} |
|||
|
|||
// ==================== 创建帧方法 ====================
|
|||
private static CsdjFrame createIdpwdUaFrame(String userId, String password) { |
|||
// 信息部: userId (16字节) + password (16字节) = 32字节
|
|||
byte[] info = new byte[32]; |
|||
byte[] userIdBytes = userId.getBytes(); |
|||
byte[] passwordBytes = password.getBytes(); |
|||
System.arraycopy(userIdBytes, 0, info, 0, Math.min(userIdBytes.length, 16)); |
|||
System.arraycopy(passwordBytes, 0, info, 16, Math.min(passwordBytes.length, 16)); |
|||
|
|||
return createFrame(FRAME_ID_PASSWORD_UA, info); |
|||
} |
|||
|
|||
private static CsdjFrame createInfoFrame(byte[] info) { |
|||
return createFrame(FRAME_INFORMATION, info); |
|||
} |
|||
|
|||
private static CsdjFrame createSimpleFrame(byte frameId) { |
|||
return createFrame(frameId, null); |
|||
} |
|||
|
|||
private static CsdjFrame createFrame(byte frameId, byte[] info) { |
|||
int infoLen = (info != null) ? info.length : 0; |
|||
int length = 16 + infoLen; |
|||
|
|||
ByteBuffer buffer = ByteBuffer.allocate(length); |
|||
buffer.order(ByteOrder.BIG_ENDIAN); |
|||
|
|||
// 数据长度
|
|||
buffer.putShort((short) length); |
|||
|
|||
// 终端ID
|
|||
byte[] terminalIdBytes = new byte[12]; |
|||
byte[] src = TERMINAL_ID.getBytes(); |
|||
System.arraycopy(src, 0, terminalIdBytes, 0, Math.min(src.length, 12)); |
|||
buffer.put(terminalIdBytes); |
|||
|
|||
// 控制部: 识别码 + 属性 (0x80: 结束位)
|
|||
buffer.put(frameId); |
|||
buffer.put((byte) 0x80); |
|||
|
|||
// 信息部
|
|||
if (info != null) { |
|||
buffer.put(info); |
|||
} |
|||
|
|||
return new CsdjFrame(buffer.array(), info); |
|||
} |
|||
|
|||
// ==================== 模拟数据 ====================
|
|||
private static byte[] createMockSensorData() { |
|||
// 模拟数据: 温度=25.6℃，湿度=60%
|
|||
byte[] data = new byte[4]; |
|||
data[0] = 0x19; // 温度高8位 (25)
|
|||
data[1] = 0x06; // 温度低8位 (0.6)
|
|||
data[2] = 0x3C; // 湿度60 (0x3C)
|
|||
data[3] = 0x01; // 状态:正常
|
|||
return data; |
|||
} |
|||
|
|||
// ==================== 日志打印 ====================
|
|||
private static String getFrameName(byte frameId) { |
|||
switch (frameId) { |
|||
case FRAME_ID_PASSWORD: return "IDPWD"; |
|||
case FRAME_ID_PASSWORD_UA: return "IDPWD-UA"; |
|||
case FRAME_NEW_MESSAGE_SEND_READY: return "NMSR"; |
|||
case FRAME_INFORMATION: return "I帧"; |
|||
case FRAME_RECEIVE_READY: return "RR"; |
|||
case FRAME_DISCONNECT: return "DISC"; |
|||
case FRAME_DISCONNECT_UA: return "DISC-UA"; |
|||
default: return String.format("UNKNOWN(0x%02X)", frameId); |
|||
} |
|||
} |
|||
|
|||
private static void logFrame(CsdjFrame frame) { |
|||
log.info(" 长度: {} 字节", frame.data.length); |
|||
log.info(" 终端ID: {}", frame.terminalId); |
|||
log.info(" 控制码: 0x{}", String.format("%02X", frame.controlId)); |
|||
if (frame.info != null) { |
|||
log.info(" 信息部: {}", bytesToHex(frame.info)); |
|||
} |
|||
} |
|||
|
|||
private static String bytesToHex(byte[] bytes) { |
|||
StringBuilder sb = new StringBuilder(); |
|||
for (byte b : bytes) { |
|||
sb.append(String.format("%02X ", b)); |
|||
} |
|||
return sb.toString().trim(); |
|||
} |
|||
|
|||
// ==================== 内部类 ====================
|
|||
static class CsdjFrame { |
|||
byte[] data; |
|||
String terminalId; |
|||
byte controlId; |
|||
byte[] info; |
|||
|
|||
CsdjFrame(byte[] header, byte[] info) { |
|||
this.data = new byte[header.length + (info != null ? info.length : 0)]; |
|||
System.arraycopy(header, 0, data, 0, header.length); |
|||
if (info != null) { |
|||
System.arraycopy(info, 0, data, header.length, info.length); |
|||
} |
|||
|
|||
// 解析终端ID
|
|||
byte[] idBytes = new byte[12]; |
|||
System.arraycopy(header, 2, idBytes, 0, 12); |
|||
this.terminalId = new String(idBytes).trim(); |
|||
|
|||
// 解析控制码
|
|||
this.controlId = header[14]; |
|||
|
|||
// 信息部
|
|||
this.info = info; |
|||
} |
|||
|
|||
byte[] toBytes() { |
|||
return data; |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue