diff --git a/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/handler/MqttReceiveHandler.java b/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/handler/MqttReceiveHandler.java index 3ca5ff1cc..03b3d79cf 100644 --- a/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/handler/MqttReceiveHandler.java +++ b/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/handler/MqttReceiveHandler.java @@ -80,14 +80,14 @@ public class MqttReceiveHandler { messageHeader.getMqttReceivedTopic(), messageHeader.getMqttReceivedQos()); return; } - MqttScheduleJob.messageCount.getAndIncrement(); + MqttScheduleJob.MESSAGE_COUNT.getAndIncrement(); MqttMessage mqttMessage = MqttMessage.builder().header(messageHeader).payload(payload).build(); log.debug("Mqtt inbound, From: {}, Qos: {}, Payload: {}", messageHeader.getMqttReceivedTopic(), messageHeader.getMqttReceivedQos(), payload); // Determine whether to process data in batch based on transmission speed Integer batchSpeed = mqttProperties.getBatch().getSpeed(); - if (MqttScheduleJob.messageSpeed.get() < batchSpeed) { + if (MqttScheduleJob.MESSAGE_SPEED.get() < batchSpeed) { virtualThreadExecutor.execute(() -> receiveSingle(mqttMessage)); } else { // Save message to batch schedule for processing diff --git a/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJob.java b/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJob.java index 57872ed18..0a2b55fbc 100644 --- a/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJob.java +++ b/dc3-common/dc3-common-mqtt/src/main/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJob.java @@ -50,13 +50,13 @@ import java.util.concurrent.locks.ReentrantReadWriteLock; @ConditionalOnBean(MqttReceiveService.class) public class MqttScheduleJob extends QuartzJobBean { - public static final ReentrantReadWriteLock messageLock = new ReentrantReadWriteLock(); + public static final ReentrantReadWriteLock MESSAGE_LOCK = new ReentrantReadWriteLock(); - public static final AtomicLong messageCount = new AtomicLong(0); + public static final AtomicLong MESSAGE_COUNT = new AtomicLong(0); - public static final AtomicLong messageSpeed = new AtomicLong(0); + public static final AtomicLong MESSAGE_SPEED = new AtomicLong(0); - private static final List mqttMessages = new ArrayList<>(); + private static final List MQTT_MESSAGES = new ArrayList<>(); @Resource private MqttProperties mqttProperties; @@ -73,11 +73,11 @@ public class MqttScheduleJob extends QuartzJobBean { * @return message size */ public static int getMqttMessagesSize() { - messageLock.readLock().lock(); + MESSAGE_LOCK.readLock().lock(); try { - return mqttMessages.size(); + return MQTT_MESSAGES.size(); } finally { - messageLock.readLock().unlock(); + MESSAGE_LOCK.readLock().unlock(); } } @@ -85,11 +85,11 @@ public class MqttScheduleJob extends QuartzJobBean { * Clear MqttMessage list */ public static void clearMqttMessages() { - messageLock.writeLock().lock(); + MESSAGE_LOCK.writeLock().lock(); try { - mqttMessages.clear(); + MQTT_MESSAGES.clear(); } finally { - messageLock.writeLock().unlock(); + MESSAGE_LOCK.writeLock().unlock(); } } @@ -99,11 +99,11 @@ public class MqttScheduleJob extends QuartzJobBean { * @param mqttMessage MqttMessage */ public static void addMqttMessages(MqttMessage mqttMessage) { - messageLock.writeLock().lock(); + MESSAGE_LOCK.writeLock().lock(); try { - mqttMessages.add(mqttMessage); + MQTT_MESSAGES.add(mqttMessage); } finally { - messageLock.writeLock().unlock(); + MESSAGE_LOCK.writeLock().unlock(); } } @@ -118,8 +118,8 @@ public class MqttScheduleJob extends QuartzJobBean { // Calculate MQTT message receive rate Integer interval = mqttProperties.getBatch().getInterval(); Integer batchSpeed = mqttProperties.getBatch().getSpeed(); - long speed = messageCount.getAndSet(0) / interval; - messageSpeed.set(speed); + long speed = MESSAGE_COUNT.getAndSet(0) / interval; + MESSAGE_SPEED.set(speed); if (speed >= batchSpeed) { log.debug("Mqtt message receiver speed: {} /s, value size: {}, interval: {}", speed, getMqttMessagesSize(), interval); @@ -128,15 +128,15 @@ public class MqttScheduleJob extends QuartzJobBean { // Process a private snapshot outside the lock so inbound MQTT threads are not // blocked by downstream parsing or publishing. List snapshot; - messageLock.writeLock().lock(); + MESSAGE_LOCK.writeLock().lock(); try { - if (mqttMessages.isEmpty()) { + if (MQTT_MESSAGES.isEmpty()) { return; } - snapshot = new ArrayList<>(mqttMessages); - mqttMessages.clear(); + snapshot = new ArrayList<>(MQTT_MESSAGES); + MQTT_MESSAGES.clear(); } finally { - messageLock.writeLock().unlock(); + MESSAGE_LOCK.writeLock().unlock(); } virtualThreadExecutor.execute(() -> receiveBatch(snapshot)); diff --git a/dc3-common/dc3-common-mqtt/src/test/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJobTest.java b/dc3-common/dc3-common-mqtt/src/test/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJobTest.java index ac813d516..9dc6b1ab2 100644 --- a/dc3-common/dc3-common-mqtt/src/test/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJobTest.java +++ b/dc3-common/dc3-common-mqtt/src/test/java/io/github/pnoker/common/mqtt/service/job/MqttScheduleJobTest.java @@ -72,27 +72,27 @@ class MqttScheduleJobTest { injectField(job, "mqttReceiveService", mqttReceiveService); injectField(job, "virtualThreadExecutor", executor); - MqttScheduleJob.messageCount.set(0); - MqttScheduleJob.messageSpeed.set(0); + MqttScheduleJob.MESSAGE_COUNT.set(0); + MqttScheduleJob.MESSAGE_SPEED.set(0); MqttScheduleJob.clearMqttMessages(); } @AfterEach void tearDown() { executor.shutdownNow(); - MqttScheduleJob.messageCount.set(0); - MqttScheduleJob.messageSpeed.set(0); + MqttScheduleJob.MESSAGE_COUNT.set(0); + MqttScheduleJob.MESSAGE_SPEED.set(0); MqttScheduleJob.clearMqttMessages(); } @Test void executeRotatesMessageCountIntoPerSecondSpeed() throws Exception { - MqttScheduleJob.messageCount.set(50); + MqttScheduleJob.MESSAGE_COUNT.set(50); job.executeInternal(jobExecutionContext); - assertThat(MqttScheduleJob.messageSpeed.get()).isEqualTo(10); - assertThat(MqttScheduleJob.messageCount.get()).isZero(); + assertThat(MqttScheduleJob.MESSAGE_SPEED.get()).isEqualTo(10); + assertThat(MqttScheduleJob.MESSAGE_COUNT.get()).isZero(); } @Test diff --git a/dc3-common/dc3-common-repository/src/main/java/io/github/pnoker/common/strategy/RepositoryStrategyFactory.java b/dc3-common/dc3-common-repository/src/main/java/io/github/pnoker/common/strategy/RepositoryStrategyFactory.java index 53866b7b9..157b192fe 100644 --- a/dc3-common/dc3-common-repository/src/main/java/io/github/pnoker/common/strategy/RepositoryStrategyFactory.java +++ b/dc3-common/dc3-common-repository/src/main/java/io/github/pnoker/common/strategy/RepositoryStrategyFactory.java @@ -38,7 +38,7 @@ import java.util.concurrent.ConcurrentHashMap; public class RepositoryStrategyFactory { // Map to store repository services with name as key - private static final Map repositoryServiceMap = new ConcurrentHashMap<>(); + private static final Map REPOSITORY_SERVICE_MAP = new ConcurrentHashMap<>(); // Private constructor to prevent instantiation private RepositoryStrategyFactory() { @@ -51,7 +51,7 @@ public class RepositoryStrategyFactory { * @return List of all repository services */ public static List get() { - return new ArrayList<>(repositoryServiceMap.values()); + return new ArrayList<>(REPOSITORY_SERVICE_MAP.values()); } /** @@ -65,7 +65,7 @@ public class RepositoryStrategyFactory { if (Objects.isNull(key)) { return null; } - return repositoryServiceMap.get(key); + return REPOSITORY_SERVICE_MAP.get(key); } /** @@ -90,7 +90,7 @@ public class RepositoryStrategyFactory { if (Objects.isNull(key)) { throw new IllegalArgumentException("Repository strategy name must not be blank"); } - repositoryServiceMap.put(key, service); + REPOSITORY_SERVICE_MAP.put(key, service); } /** @@ -103,14 +103,14 @@ public class RepositoryStrategyFactory { if (Objects.isNull(key)) { return; } - repositoryServiceMap.remove(key); + REPOSITORY_SERVICE_MAP.remove(key); } /** * Clear all registered repository services. */ public static void clear() { - repositoryServiceMap.clear(); + REPOSITORY_SERVICE_MAP.clear(); } private static String repositoryKey(String name) { diff --git a/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImpl.java b/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImpl.java index a2f9e2db1..78e7b8162 100644 --- a/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImpl.java +++ b/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImpl.java @@ -172,7 +172,7 @@ public class ListeningVirtualDriverCustomServiceImpl implements DriverCustomServ public Boolean write(Map driverConfig, Map pointConfig, DeviceBO device, PointBO point, WritePointValue writePointValue) { Long deviceId = device.getId(); - Channel channel = NettyTcpServer.deviceChannelMap.get(deviceId); + Channel channel = NettyTcpServer.DEVICE_CHANNEL_MAP.get(deviceId); if (Objects.nonNull(channel)) { log.debug("Driver point write requested, protocol=tcp, deviceId={}, pointId={}, valueLength={}", deviceId, point.getId(), Objects.toString(writePointValue.getValue(), "").length()); diff --git a/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/NettyServerHandler.java b/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/NettyServerHandler.java index 95d5f2079..c97a8238e 100644 --- a/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/NettyServerHandler.java +++ b/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/NettyServerHandler.java @@ -103,7 +103,7 @@ public class NettyServerHandler { } String hexKey = ByteBufUtil.hexDump(byteBuf, 22, 1); - NettyTcpServer.deviceChannelMap.put(deviceId, context.channel()); + NettyTcpServer.DEVICE_CHANNEL_MAP.put(deviceId, context.channel()); Map> pointConfigMap = deviceMetadata.getPointConfig(deviceId); if (Objects.isNull(pointConfigMap)) { diff --git a/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/tcp/NettyTcpServer.java b/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/tcp/NettyTcpServer.java index 8f95cbb90..f0697395f 100644 --- a/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/tcp/NettyTcpServer.java +++ b/dc3-driver/dc3-driver-listening-virtual/src/main/java/io/github/pnoker/driver/service/netty/tcp/NettyTcpServer.java @@ -59,7 +59,7 @@ public class NettyTcpServer { * sending data back to specific devices when needed. *

*/ - public static final Map deviceChannelMap = new ConcurrentHashMap<>(16); + public static final Map DEVICE_CHANNEL_MAP = new ConcurrentHashMap<>(16); /** * Starts the TCP server on the specified port. diff --git a/dc3-driver/dc3-driver-listening-virtual/src/test/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImplTest.java b/dc3-driver/dc3-driver-listening-virtual/src/test/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImplTest.java index 56975d0ec..55d6f9ee5 100644 --- a/dc3-driver/dc3-driver-listening-virtual/src/test/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImplTest.java +++ b/dc3-driver/dc3-driver-listening-virtual/src/test/java/io/github/pnoker/driver/service/impl/ListeningVirtualDriverCustomServiceImplTest.java @@ -106,7 +106,7 @@ class ListeningVirtualDriverCustomServiceImplTest { @AfterEach void cleanChannelMap() { - NettyTcpServer.deviceChannelMap.clear(); + NettyTcpServer.DEVICE_CHANNEL_MAP.clear(); } @Test @@ -159,7 +159,7 @@ class ListeningVirtualDriverCustomServiceImplTest { DeviceBO device = device(900L); PointBO point = point(2L); Channel channel = org.mockito.Mockito.mock(Channel.class); - NettyTcpServer.deviceChannelMap.put(900L, channel); + NettyTcpServer.DEVICE_CHANNEL_MAP.put(900L, channel); Boolean ok = service.write(null, null, device, point, WritePointValue.builder().value("hello").type(PointTypeFlagEnum.STRING).build());