1、报税口加密数据，解密机处理

ruoyi-admin/src/main/resources/application-druid.yml

@@ -106,4 +106,16 @@ emqx:
           listener: clientLineStatusListener
         - topic: $share/wbjw/cpyypt/logup/#
           qos: 2
-          listener: deviceLogListener
+          listener: deviceLogListener
+
+# 报税口密文解密设备配置
+decrypt:
+    device:
+        config:
+            defaultFailTimes: 3
+            defaultSendSize: 20
+            timeout: 120000
+            decryptDeviceList:
+                - deviceSn: 1
+                  deviceType: '0102'
+            outDecrypt: false

ruoyi-device/src/main/java/com/ruoyi/device/config/DecryptDevice.java

@@ -0,0 +1,31 @@
+package com.ruoyi.device.config;
+
+/**
+ * 解密设备配置项
+ *
+ * @author lwm
+ */
+public class DecryptDevice
+{
+    /** 设备 SN */
+    private Long deviceSn;
+
+    /** 设备类型 */
+    private String deviceType;
+
+    public Long getDeviceSn() {
+        return deviceSn;
+    }
+
+    public void setDeviceSn(Long deviceSn) {
+        this.deviceSn = deviceSn;
+    }
+
+    public String getDeviceType() {
+        return deviceType;
+    }
+
+    public void setDeviceType(String deviceType) {
+        this.deviceType = deviceType;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/config/DecryptDeviceConfig.java

@@ -0,0 +1,95 @@
+package com.ruoyi.device.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 解密设备调度配置
+ *
+ * @author lwm
+ */
+@Configuration
+@ConfigurationProperties(prefix = "decrypt.device.config")
+public class DecryptDeviceConfig
+{
+    /** 最大失败重试次数 */
+    private Integer defaultFailTimes = 3;
+
+    /** 每次发送密文数据条数 */
+    private Integer defaultSendSize = 20;
+
+    /** 解密超时时间（毫秒） */
+    private Integer timeout = 120000;
+
+    /** 解密设备列表 */
+    private List<DecryptDevice> decryptDeviceList = new ArrayList<>();
+
+    /** 是否走外部解密 */
+    private Boolean outDecrypt = false;
+
+    /** 外部解密接口地址 */
+    private String outDecryptUrl;
+
+    public Integer getDefaultFailTimes()
+    {
+        return defaultFailTimes;
+    }
+
+    public void setDefaultFailTimes(Integer defaultFailTimes)
+    {
+        this.defaultFailTimes = defaultFailTimes;
+    }
+
+    public Integer getDefaultSendSize()
+    {
+        return defaultSendSize;
+    }
+
+    public void setDefaultSendSize(Integer defaultSendSize)
+    {
+        this.defaultSendSize = defaultSendSize;
+    }
+
+    public Integer getTimeout()
+    {
+        return timeout;
+    }
+
+    public void setTimeout(Integer timeout)
+    {
+        this.timeout = timeout;
+    }
+
+    public List<DecryptDevice> getDecryptDeviceList()
+    {
+        return decryptDeviceList;
+    }
+
+    public void setDecryptDeviceList(List<DecryptDevice> decryptDeviceList)
+    {
+        this.decryptDeviceList = decryptDeviceList;
+    }
+
+    public Boolean getOutDecrypt()
+    {
+        return outDecrypt;
+    }
+
+    public void setOutDecrypt(Boolean outDecrypt)
+    {
+        this.outDecrypt = outDecrypt;
+    }
+
+    public String getOutDecryptUrl()
+    {
+        return outDecryptUrl;
+    }
+
+    public void setOutDecryptUrl(String outDecryptUrl)
+    {
+        this.outDecryptUrl = outDecryptUrl;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/api/EmqxApiUtil.java

@@ -0,0 +1,74 @@
+package com.ruoyi.device.mqtt.api;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.TypeReference;
+import com.ruoyi.device.config.MqttConfig;
+import com.ruoyi.device.mqtt.api.constants.ApiPathConstants;
+import com.ruoyi.device.mqtt.api.item.ClientInfo;
+import com.ruoyi.device.mqtt.api.item.EmqxResult;
+import jakarta.annotation.Resource;
+import org.springframework.http.*;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.List;
+
+/**
+ * EMQX HTTP API 工具
+ *
+ * @author lwm
+ */
+@Component
+public class EmqxApiUtil
+{
+    @Resource
+    private MqttConfig mqttConfig;
+
+    /**
+     * 获取客户端信息
+     *
+     * @param deviceType 设备类型
+     * @param deviceSn 设备 SN
+     * @return 客户端信息
+     */
+    public EmqxResult<List<ClientInfo>> getClientInfo(String deviceType, Long deviceSn)
+    {
+        String url = mqttConfig.getApiPath()
+            + String.format(ApiPathConstants.GET_CLIENT_INFO, getClientId(deviceType, deviceSn));
+        return get(url, new TypeReference<>() {
+        });
+    }
+
+    /**
+     * 获取客户端ID
+     *
+     * @param deviceType 设备类型
+     * @param deviceSn 设备 SN
+     * @return 客户端ID
+     */
+    protected String getClientId(String deviceType, Long deviceSn)
+    {
+        String deviceSnFormat = String.format("%010d", deviceSn);
+        return deviceType + ApiPathConstants.MIDDLELINE + deviceSnFormat;
+    }
+
+    /**
+     * GET 请求
+     *
+     * @param url 请求地址
+     * @param typeReference 返回数据类型
+     * @return 返回数据
+     */
+    private <T> T get(String url, TypeReference<T> typeReference)
+    {
+        HttpHeaders headers = new HttpHeaders();
+        headers.setBasicAuth(mqttConfig.getApiUserName(), mqttConfig.getApiPassword());
+        headers.setContentType(MediaType.APPLICATION_JSON);
+        HttpEntity<Void> entity = new HttpEntity<>(headers);
+
+        RestTemplate restTemplate = new RestTemplate();
+        ResponseEntity<String> responseEntity = restTemplate.exchange(
+                url, HttpMethod.GET, entity, String.class);
+        return JSON.parseObject(responseEntity.getBody(), typeReference);
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/api/constants/ApiPathConstants.java

@@ -0,0 +1,32 @@
+package com.ruoyi.device.mqtt.api.constants;
+
+/**
+ * EMQX API 路径常量
+ *
+ * @author lwm
+ */
+public interface ApiPathConstants
+{
+    /**
+     * 中线
+     */
+    String MIDDLELINE = "-";
+
+    /**
+     * 四个下划线
+     */
+    String FOUR_UNDER_LINE = "%s_%s_%s_%s_%s";
+
+    /**
+     * 下划线
+     */
+    String UNDERLINE = "_";
+
+    /**
+     * EMQX API 路径：获取客户端信息
+     */
+    String GET_CLIENT_INFO = "/api/v4/clients/%s";
+    String GET_SUB_INFO = "/api/v4/subscriptions/%s";
+    String DELETE_CLIENT = "/api/v4/clients/%s";
+    String GET_ALL_CLIENT = "/api/v4/clients?_page=1&_limit=50000";
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/api/item/ClientInfo.java

@@ -0,0 +1,145 @@
+package com.ruoyi.device.mqtt.api.item;
+
+/**
+ * EMQX 客户端信息
+ *
+ * @author lwm
+ */
+public class ClientInfo
+{
+    /**
+     * 客户端所属区域/分区名称
+     */
+    private String zone;
+
+    /**
+     * 客户端接收的消息总数（累计计数）
+     */
+    private String recv_cnt;
+
+    /**
+     * 消息队列最大长度限制
+     */
+    private String max_mqueue;
+
+    /**
+     * EMQX 集群节点名称，标识客户端连接到的具体节点
+     */
+    private String node;
+
+    /**
+     * 当前消息队列中等待处理的消息数量
+     */
+    private String mqueue_len;
+
+    /**
+     * 最大飞行窗口大小（同时处理的未确认消息数）
+     */
+    private String max_inflight;
+
+    /**
+     * 是否为桥接客户端（true/false）
+     * 桥接用于连接多个 MQTT Broker
+     */
+    private String is_bridge;
+
+    /**
+     * 因消息队列满而被丢弃的消息数量
+     */
+    private String mqueue_dropped;
+
+    /**
+     * 客户端连接状态（connected/disconnected）
+     */
+    private String connected;
+
+    public String getZone()
+    {
+        return zone;
+    }
+
+    public void setZone(String zone)
+    {
+        this.zone = zone;
+    }
+
+    public String getRecv_cnt()
+    {
+        return recv_cnt;
+    }
+
+    public void setRecv_cnt(String recv_cnt)
+    {
+        this.recv_cnt = recv_cnt;
+    }
+
+    public String getMax_mqueue()
+    {
+        return max_mqueue;
+    }
+
+    public void setMax_mqueue(String max_mqueue)
+    {
+        this.max_mqueue = max_mqueue;
+    }
+
+    public String getNode()
+    {
+        return node;
+    }
+
+    public void setNode(String node)
+    {
+        this.node = node;
+    }
+
+    public String getMqueue_len()
+    {
+        return mqueue_len;
+    }
+
+    public void setMqueue_len(String mqueue_len)
+    {
+        this.mqueue_len = mqueue_len;
+    }
+
+    public String getMax_inflight()
+    {
+        return max_inflight;
+    }
+
+    public void setMax_inflight(String max_inflight)
+    {
+        this.max_inflight = max_inflight;
+    }
+
+    public String getIs_bridge()
+    {
+        return is_bridge;
+    }
+
+    public void setIs_bridge(String is_bridge)
+    {
+        this.is_bridge = is_bridge;
+    }
+
+    public String getMqueue_dropped()
+    {
+        return mqueue_dropped;
+    }
+
+    public void setMqueue_dropped(String mqueue_dropped)
+    {
+        this.mqueue_dropped = mqueue_dropped;
+    }
+
+    public String getConnected()
+    {
+        return connected;
+    }
+
+    public void setConnected(String connected)
+    {
+        this.connected = connected;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/api/item/EmqxResult.java

@@ -0,0 +1,32 @@
+package com.ruoyi.device.mqtt.api.item;
+
+/**
+ * EMQX API 响应
+ *
+ * @author lwm
+ */
+public class EmqxResult<T>
+{
+    private Integer code;
+    private T data;
+
+    public Integer getCode()
+    {
+        return code;
+    }
+
+    public void setCode(Integer code)
+    {
+        this.code = code;
+    }
+
+    public T getData()
+    {
+        return data;
+    }
+
+    public void setData(T data)
+    {
+        this.data = data;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/domain/decoder/tax/TaxMonthDataMessage.java

@@ -0,0 +1,38 @@
+package com.ruoyi.device.mqtt.domain.decoder.tax;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+
+/**
+ * 报税口月报数据（密文透传解密结果）
+ */
+public class TaxMonthDataMessage implements Serializable
+{
+    private static final long serialVersionUID = 1L;
+
+    /** 油量（L） */
+    private BigDecimal volume;
+
+    /** 总价（元） */
+    private BigDecimal amount;
+
+    public BigDecimal getVolume()
+    {
+        return volume;
+    }
+
+    public void setVolume(BigDecimal volume)
+    {
+        this.volume = volume;
+    }
+
+    public BigDecimal getAmount()
+    {
+        return amount;
+    }
+
+    public void setAmount(BigDecimal amount)
+    {
+        this.amount = amount;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/domain/decoder/tax/TaxTransferDataMessage.java

@@ -0,0 +1,208 @@
+package com.ruoyi.device.mqtt.domain.decoder.tax;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.util.Date;
+
+/**
+ * 报税口交易数据（密文解密结果）
+ */
+public class TaxTransferDataMessage implements Serializable
+{
+    private static final long serialVersionUID = 1L;
+
+    /** 设备类型 */
+    private Integer deviceType;
+
+    /** 网关编号 */
+    private Long gatewayNo;
+
+    /** 采集器编号 */
+    private Long collectorNo;
+
+    /** 报税口编号 */
+    private Integer taxNo;
+
+    /** 枪编号 */
+    private Integer gunNo;
+
+    /** 报税口交易流水号 */
+    private Long serialNo;
+
+    /** 单价（元） */
+    private BigDecimal unitPrice;
+
+    /** 油量（L） */
+    private BigDecimal volume;
+
+    /** 总价（元） */
+    private BigDecimal amount;
+
+    /** 总油量 */
+    private BigDecimal totalVolume;
+
+    /** 总金额 */
+    private BigDecimal totalAmount;
+
+    /** 数据上报时间 */
+    private Date dataTime;
+
+    /** 累计上报笔数 */
+    private Long totalTimes;
+
+    /** 提枪时间 */
+    private Date raiseTime;
+
+    /** 挂枪时间 */
+    private Date hangTime;
+
+    public Integer getDeviceType()
+    {
+        return deviceType;
+    }
+
+    public void setDeviceType(Integer deviceType)
+    {
+        this.deviceType = deviceType;
+    }
+
+    public Long getGatewayNo()
+    {
+        return gatewayNo;
+    }
+
+    public void setGatewayNo(Long gatewayNo)
+    {
+        this.gatewayNo = gatewayNo;
+    }
+
+    public Long getCollectorNo()
+    {
+        return collectorNo;
+    }
+
+    public void setCollectorNo(Long collectorNo)
+    {
+        this.collectorNo = collectorNo;
+    }
+
+    public Integer getTaxNo()
+    {
+        return taxNo;
+    }
+
+    public void setTaxNo(Integer taxNo)
+    {
+        this.taxNo = taxNo;
+    }
+
+    public Integer getGunNo()
+    {
+        return gunNo;
+    }
+
+    public void setGunNo(Integer gunNo)
+    {
+        this.gunNo = gunNo;
+    }
+
+    public Long getSerialNo()
+    {
+        return serialNo;
+    }
+
+    public void setSerialNo(Long serialNo)
+    {
+        this.serialNo = serialNo;
+    }
+
+    public BigDecimal getUnitPrice()
+    {
+        return unitPrice;
+    }
+
+    public void setUnitPrice(BigDecimal unitPrice)
+    {
+        this.unitPrice = unitPrice;
+    }
+
+    public BigDecimal getVolume()
+    {
+        return volume;
+    }
+
+    public void setVolume(BigDecimal volume)
+    {
+        this.volume = volume;
+    }
+
+    public BigDecimal getAmount()
+    {
+        return amount;
+    }
+
+    public void setAmount(BigDecimal amount)
+    {
+        this.amount = amount;
+    }
+
+    public BigDecimal getTotalVolume()
+    {
+        return totalVolume;
+    }
+
+    public void setTotalVolume(BigDecimal totalVolume)
+    {
+        this.totalVolume = totalVolume;
+    }
+
+    public BigDecimal getTotalAmount()
+    {
+        return totalAmount;
+    }
+
+    public void setTotalAmount(BigDecimal totalAmount)
+    {
+        this.totalAmount = totalAmount;
+    }
+
+    public Date getDataTime()
+    {
+        return dataTime;
+    }
+
+    public void setDataTime(Date dataTime)
+    {
+        this.dataTime = dataTime;
+    }
+
+    public Long getTotalTimes()
+    {
+        return totalTimes;
+    }
+
+    public void setTotalTimes(Long totalTimes)
+    {
+        this.totalTimes = totalTimes;
+    }
+
+    public Date getRaiseTime()
+    {
+        return raiseTime;
+    }
+
+    public void setRaiseTime(Date raiseTime)
+    {
+        this.raiseTime = raiseTime;
+    }
+
+    public Date getHangTime()
+    {
+        return hangTime;
+    }
+
+    public void setHangTime(Date hangTime)
+    {
+        this.hangTime = hangTime;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/domain/encoder/tax/CiphertextDataDown.java

@@ -0,0 +1,26 @@
+package com.ruoyi.device.mqtt.domain.encoder.tax;
+
+import com.ruoyi.device.mqtt.domain.BaseBody;
+
+/**
+ * 报税口密文 解密下发请求
+ */
+public class CiphertextDataDown extends BaseBody
+{
+    private byte[] ciphertextData;
+
+    public CiphertextDataDown()
+    {
+        super();
+    }
+
+    public byte[] getCiphertextData()
+    {
+        return ciphertextData;
+    }
+
+    public void setCiphertextData(byte[] ciphertextData)
+    {
+        this.ciphertextData = ciphertextData;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/enums/DeviceModelEnum.java

@@ -0,0 +1,104 @@
+package com.ruoyi.device.mqtt.enums;
+
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * 设备型号枚举
+ */
+public enum DeviceModelEnum
+{
+    GATEWAY_DEVICE_TYPE(1, "0101", "加油机路由器A"),
+    GATEWAY_DEVICE_TYPE_0102(1, "0102", "加油机路由器B"),
+    GATEWAY_DEVICE_TYPE_0103(1, "0103", "加油机路由器B"),
+    GATEWAY_DEVICE_TYPE_0104(1, "0104", "加油机路由器B"),
+    INVENT_GATEWAY_DEVICE(1, "0109", "定制加油机路由器"),
+
+    COLLECTOR_DEVICE_TYPE(2, "0201", "加油机采集器A"),
+    COLLECTOR_DEVICE_TYPE_0202(2, "0202", "加油机采集器B"),
+    INVENT_COLLECTOR_DEVICE(2, "0209", "定制加油机采集器"),
+
+    OPW_GATEWAY(3, "0301", "液位仪网关A"),
+    OPW_GATEWAY_0302(3, "0302", "液位仪网关B"),
+
+//    SCREEN(4, "1XXX", "屏采集器"),
+
+    CODER(5,"0501", "编码器"),
+
+//    ENCRYPTION_MODULE(6,"0601", "加密模块"),
+
+    GUN_BLOCKER(7, "0701", "加油枪阻断器"),
+
+    TAX(8, "0801", "报税口"),
+
+    PROBE_0901(9, "0901", "探针板A"),
+    PROBE_0902(9, "0902", "探针板B"),
+    PROBE_0904(9, "0904", "探针板D"),
+    PROBE_0906(9, "0906", "探针板E");
+
+    private final Integer type;
+    private final String model;
+    private final String message;
+
+    DeviceModelEnum(Integer type, String model, String message)
+    {
+        this.type = type;
+        this.model = model;
+        this.message = message;
+    }
+
+    public Integer getType()
+    {
+        return type;
+    }
+
+    public String getModel()
+    {
+        return model;
+    }
+
+    public String getMessage()
+    {
+        return message;
+    }
+
+    public static Integer getTypeByModel(String model)
+    {
+        if (StringUtils.isBlank(model))
+        {
+            return null;
+        }
+
+        if (model.startsWith("10"))
+        {
+            return 10;
+        }
+
+        for (DeviceModelEnum value : DeviceModelEnum.values())
+        {
+            if (value.getModel().equals(model))
+            {
+                return value.getType();
+            }
+        }
+
+        return null;
+    }
+
+    public static String getMessageByModel(String model)
+    {
+        if (StringUtils.isBlank(model))
+        {
+            return null;
+        }
+
+        for (DeviceModelEnum value : DeviceModelEnum.values())
+        {
+            if (value.getModel().equals(model))
+            {
+                return value.getMessage();
+            }
+        }
+
+        return null;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/enums/DeviceRedisEnum.java

@@ -0,0 +1,41 @@
+package com.ruoyi.device.mqtt.enums;
+
+/**
+ * 设备模块 Redis 键枚举（密文解密相关）
+ */
+public enum DeviceRedisEnum
+{
+    DEVICE_FREE_SPACE("CPYYPT:DEVICE_FREE_SPACE:%s", -1, "设备剩余空间"),
+    PASS_TAX_DATA_GUN_NO("%s_%s_%s", 600, "加油机报税口月报数据请求中枪号"),
+    PASS_TAX_DATA("%s_%s_%s_%s_%s", 600, "加油机报税口月报数据"),
+    TAX_CIPHERTEXT_DATA_ORIGNAL("CPYYPT:TAX_CIPHERTEXT_DATA_ORIGNAL", -1, "报税口密文原始队列"),
+    TAX_CIPHERTEXT_DATA_SEND("CPYYPT:TAX_CIPHERTEXT_DATA_SEND", -1, "报税口密文发送队列"),
+    TAX_CIPHERTEXT_DATA_FAIL("CPYYPT:TAX_CIPHERTEXT_DATA_FAIL", -1, "报税口密文失败队列"),
+    TAX_CIPHERTEXT_OUT_RESULT("CPYYPT:TCOR:%s", 30, "外部报税口密文数据结果");
+
+    private final String key;
+    private final int expireTime;
+    private final String message;
+
+    DeviceRedisEnum(String key, int expireTime, String message)
+    {
+        this.key = key;
+        this.expireTime = expireTime;
+        this.message = message;
+    }
+
+    public String getKey()
+    {
+        return key;
+    }
+
+    public int getExpireTime()
+    {
+        return expireTime;
+    }
+
+    public String getMessage()
+    {
+        return message;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/decoder/MessageHandler.java

@@ -161,7 +161,7 @@ public class MessageHandler
      * @param bytes bytes
      * @return 16进制字符串 打印日志
      */
-    private String bytesToHexLog(byte[] bytes)
+    public static String bytesToHexLog(byte[] bytes)
     {
         try
         {

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/decoder/status/DecryptStatusDecoder.java

@@ -0,0 +1,31 @@
+package com.ruoyi.device.mqtt.handler.decoder.status;
+
+import com.ruoyi.common.core.redis.RedisCache;
+import com.ruoyi.device.mqtt.enums.DeviceRedisEnum;
+import com.ruoyi.device.mqtt.annotation.ConsumerHandler;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.handler.decoder.AbstractDecoder;
+import com.ruoyi.device.mqtt.vo.CommonHeader;
+import com.ruoyi.device.mqtt.vo.CommonTopic;
+import io.netty.buffer.ByteBuf;
+import jakarta.annotation.Resource;
+
+/**
+ * 解密板设备状态上行
+ */
+@ConsumerHandler(msgType = MsgTypeEnum.DECRYPT_DEVICE_UP)
+public class DecryptStatusDecoder extends AbstractDecoder<Void>
+{
+    @Resource
+    private RedisCache redisCache;
+
+    @Override
+    protected Void decode(CommonTopic commonTopic, CommonHeader header, ByteBuf body)
+    {
+        body.readBytes(48);
+        int freeSpace = body.readUnsignedByte();
+        redisCache.setCacheObject(
+            String.format(DeviceRedisEnum.DEVICE_FREE_SPACE.getKey(), commonTopic.getDeviceSn()), freeSpace);
+        return null;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/decoder/tax/CiphertextDataDecoder.java

@@ -0,0 +1,28 @@
+package com.ruoyi.device.mqtt.handler.decoder.tax;
+
+import com.ruoyi.device.mqtt.annotation.ConsumerHandler;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.handler.decoder.AbstractDecoder;
+import com.ruoyi.device.mqtt.vo.CommonHeader;
+import com.ruoyi.device.mqtt.vo.CommonTopic;
+import io.netty.buffer.ByteBuf;
+import jakarta.annotation.Resource;
+
+/**
+ * 报税口密文数据上行解码
+ *
+ * @author lwm
+ */
+@ConsumerHandler(msgType = MsgTypeEnum.CIPHERTEXT_DATA_UP)
+public class CiphertextDataDecoder extends AbstractDecoder<Void>
+{
+    @Resource
+    private CiphertextDataService ciphertextDataService;
+
+    @Override
+    protected Void decode(CommonTopic topic, CommonHeader header, ByteBuf body)
+    {
+        ciphertextDataService.decode(topic, header, body, false);
+        return null;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/decoder/tax/CiphertextDataDecryptDecoder.java

@@ -0,0 +1,228 @@
+package com.ruoyi.device.mqtt.handler.decoder.tax;
+
+import com.alibaba.fastjson2.JSON;
+import com.ruoyi.common.core.redis.RedisCache;
+import com.ruoyi.device.config.DecryptDeviceConfig;
+import com.ruoyi.device.mqtt.enums.DeviceRedisEnum;
+import com.ruoyi.device.mqtt.annotation.ConsumerHandler;
+import com.ruoyi.device.mqtt.api.constants.ApiPathConstants;
+import com.ruoyi.device.mqtt.domain.decoder.tax.TaxMonthDataMessage;
+import com.ruoyi.device.mqtt.domain.decoder.tax.TaxTransferDataMessage;
+import com.ruoyi.device.mqtt.domain.encoder.tax.CiphertextDataDown;
+import com.ruoyi.device.mqtt.enums.DeviceModelEnum;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.enums.VersionEnum;
+import com.ruoyi.device.mqtt.handler.HandlerManager;
+import com.ruoyi.device.mqtt.handler.decoder.AbstractDecoder;
+import com.ruoyi.device.mqtt.handler.decoder.MessageHandler;
+import com.ruoyi.device.mqtt.handler.encoder.IEncoder;
+import com.ruoyi.device.mqtt.util.MsgHandlerUtil;
+import com.ruoyi.device.mqtt.util.TargetTypeConvertUtil;
+import com.ruoyi.device.mqtt.vo.CiphertextData;
+import com.ruoyi.device.mqtt.vo.CommonHeader;
+import com.ruoyi.device.mqtt.vo.CommonTopic;
+import com.ruoyi.device.service.TaxTransferDataService;
+import io.netty.buffer.ByteBuf;
+import jakarta.annotation.Resource;
+import org.apache.commons.lang3.StringUtils;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 报税口密文 解密结果上行解码
+ */
+@ConsumerHandler(msgType = MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_UP)
+public class CiphertextDataDecryptDecoder extends AbstractDecoder<Void>
+{
+    @Resource
+    private HandlerManager handlerManager;
+
+    @Resource
+    private DecryptDeviceConfig decryptDeviceConfig;
+
+    @Resource
+    private RedisCache redisCache;
+
+    @Resource
+    private TaxTransferDataService taxTransferDataService;
+
+    @Override
+    protected Void decode(CommonTopic commonTopic, CommonHeader header, ByteBuf body)
+    {
+        ByteBuf copyBuf = body.copy();
+        Long gatewayNo = null;
+        Long collectorNo = null;
+        if (header.getProtoVer() == VersionEnum.THIRD.getValue())
+        {
+            TargetTypeConvertUtil.convert(body.readUnsignedShortLE());
+            gatewayNo = body.readUnsignedIntLE();
+            TargetTypeConvertUtil.convert(body.readUnsignedShortLE());
+            collectorNo = body.readUnsignedIntLE();
+        }
+        else
+        {
+            gatewayNo = body.readUnsignedIntLE();
+            collectorNo = body.readUnsignedIntLE();
+        }
+
+        int taxNo = body.readByte();
+        int gunNo = body.readByte();
+        Long serialNo = body.readUnsignedIntLE();
+        // 环境 0：解码数据返回结服务器  1：解码数据返回给数据源设备
+        int env = body.readByte();
+        // 剩余空间
+        int freeSpace = body.readByte();
+        // 1：解码成功；2：解码超时；3：报税口数据错误；4：子板没有空闲的sim卡；5：解码失败；6：子板超时；7：主板缓存已满
+        int result = body.readByte();
+
+        String key2 = String.format(ApiPathConstants.FOUR_UNDER_LINE, gatewayNo, collectorNo, taxNo, gunNo, serialNo);
+        redisCache.setCacheObject(
+            String.format(DeviceRedisEnum.DEVICE_FREE_SPACE.getKey(), commonTopic.getDeviceSn()), freeSpace);
+
+        if (result == 1 && env == 9)
+        {
+            CiphertextData ciphertextData = redisCache.getCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+            // 偷传返回
+            BigDecimal hand = new BigDecimal("100");
+            body.readBytes(19);
+            // 开始解析数据 累计值
+            body.readBytes(7);
+            String totalVolume = readASCII(body, 12);
+            String totalAmount = readASCII(body, 12);
+
+            redisCache.deleteCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+            if (StringUtils.isNotBlank(ciphertextData.getPlatform()))
+            {
+                TaxTransferDataMessage taxTransferDataMessage = new TaxTransferDataMessage();
+                taxTransferDataMessage.setGatewayNo(gatewayNo);
+                taxTransferDataMessage.setCollectorNo(collectorNo);
+                taxTransferDataMessage.setTaxNo(taxNo);
+                taxTransferDataMessage.setGunNo(gunNo);
+                taxTransferDataMessage.setSerialNo(serialNo);
+                taxTransferDataMessage.setRaiseTime(ciphertextData.getRaiseTime());
+                taxTransferDataMessage.setHangTime(ciphertextData.getHangTime());
+                taxTransferDataMessage.setTotalVolume(new BigDecimal(totalVolume).divide(hand, RoundingMode.HALF_UP));
+                taxTransferDataMessage.setTotalAmount(new BigDecimal(totalAmount).divide(hand, RoundingMode.HALF_UP));
+                // 如果说是外部平台的话 加入到一个新的缓存
+                log.info("外部平台密文透传8c13 数据解密成功,解密透传数据存入缓存：{}", taxTransferDataMessage);
+                redisCache.setCacheObject(
+                    String.format(DeviceRedisEnum.TAX_CIPHERTEXT_OUT_RESULT.getKey(), ciphertextData.getPlatform()),
+                        JSON.toJSONString(taxTransferDataMessage), DeviceRedisEnum.TAX_CIPHERTEXT_OUT_RESULT.getExpireTime(), TimeUnit.SECONDS);
+            }
+            else
+            {
+                String param = redisCache.getCacheObject(
+                    String.format(DeviceRedisEnum.PASS_TAX_DATA_GUN_NO.getKey(), gatewayNo, collectorNo, taxNo));
+                String[] split = param.split(ApiPathConstants.UNDERLINE);
+
+                TaxMonthDataMessage taxMonthDataMessage = new TaxMonthDataMessage();
+                taxMonthDataMessage.setVolume(new BigDecimal(totalVolume).divide(hand, RoundingMode.HALF_UP));
+                taxMonthDataMessage.setAmount(new BigDecimal(totalAmount).divide(hand, RoundingMode.HALF_UP));
+                log.info("密文透传8c13 数据解密成功,解密透传数据存入缓存：{}", taxMonthDataMessage);
+                redisCache.setCacheObject(
+                    String.format(DeviceRedisEnum.PASS_TAX_DATA.getKey(), gatewayNo, collectorNo, taxNo, split[0], split[1]),
+                        JSON.toJSONString(taxMonthDataMessage), DeviceRedisEnum.PASS_TAX_DATA.getExpireTime(), TimeUnit.SECONDS);
+            }
+        }
+        else if (result == 1 && env == 0)
+        {
+            // 服务器保存数据
+            CiphertextData ciphertextData = redisCache.getCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+            BigDecimal hand = new BigDecimal("100");
+            body.readBytes(19);
+            // 开始解析数据 累计值
+            body.readBytes(7);
+            String totalVolume = readASCII(body, 12);
+            String totalAmount = readASCII(body, 12);
+            body.readByte();
+            // 开始解析交易值
+            body.readBytes(7);
+            body.readBytes(6);
+            String volume = readASCII(body, 9);
+            String amount = readASCII(body, 9);
+            String price = readASCII(body, 4);
+
+            TaxTransferDataMessage taxTransferDataMessage = new TaxTransferDataMessage();
+            taxTransferDataMessage.setDeviceType(DeviceModelEnum.getTypeByModel(ciphertextData.getDeviceType()));
+            taxTransferDataMessage.setGatewayNo(gatewayNo);
+            taxTransferDataMessage.setCollectorNo(collectorNo);
+            taxTransferDataMessage.setTaxNo(taxNo);
+            taxTransferDataMessage.setGunNo(gunNo);
+            taxTransferDataMessage.setSerialNo(serialNo);
+            taxTransferDataMessage.setRaiseTime(ciphertextData.getRaiseTime());
+            taxTransferDataMessage.setHangTime(ciphertextData.getHangTime());
+            taxTransferDataMessage.setUnitPrice(new BigDecimal(price).divide(hand, RoundingMode.HALF_UP));
+            taxTransferDataMessage.setVolume(new BigDecimal(volume).divide(hand, RoundingMode.HALF_UP));
+            taxTransferDataMessage.setAmount(new BigDecimal(amount).divide(hand, RoundingMode.HALF_UP));
+            taxTransferDataMessage.setTotalVolume(new BigDecimal(totalVolume).divide(hand, RoundingMode.HALF_UP));
+            taxTransferDataMessage.setTotalAmount(new BigDecimal(totalAmount).divide(hand, RoundingMode.HALF_UP));
+            taxTransferDataMessage.setDataTime(ciphertextData.getDateTime());
+            //删除队列
+            log.info("数据解密成功,调用业务服务保存数据");
+            redisCache.deleteCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+            if (StringUtils.isNotBlank(ciphertextData.getPlatform()))
+            {
+                // 如果说是外部平台的话 加入到一个新的缓存
+                redisCache.setCacheObject(
+                    String.format(DeviceRedisEnum.TAX_CIPHERTEXT_OUT_RESULT.getKey(), ciphertextData.getPlatform()),
+                        JSON.toJSONString(taxTransferDataMessage), DeviceRedisEnum.TAX_CIPHERTEXT_OUT_RESULT.getExpireTime(), TimeUnit.SECONDS);
+            }
+            else
+            {
+                taxTransferDataService.saveTaxTransferData(taxTransferDataMessage);
+            }
+        }
+        else if (result == 1 && env == 1)
+        {
+            // 返回给客户端测试板
+            byte[] bytes = new byte[copyBuf.readableBytes() - 2];
+            copyBuf.getBytes(copyBuf.readerIndex(), bytes);
+
+            CiphertextData ciphertextData = redisCache.getCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+
+            CiphertextDataDown request = new CiphertextDataDown();
+            request.setCiphertextData(bytes);
+            request.setDeviceSn(ciphertextData.getDeviceSn());
+            request.setDeviceType(ciphertextData.getDeviceType());
+            String key = MsgHandlerUtil.getEncoderKey(MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_DOWN);
+            IEncoder<CiphertextDataDown> encoder = (IEncoder<CiphertextDataDown>) handlerManager.getEncoder(key);
+            encoder.encode(request);
+
+            log.info("数据解密成功,返回客户端解密后数据");
+            // 删除队列
+            redisCache.deleteCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+        }
+        else if (result == 2 || result == 4 || result == 6 || result == 7)
+        {
+            // 超时
+            // 删除发送队列
+            CiphertextData ciphertextData = redisCache.getCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+            redisCache.deleteCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+            if (decryptDeviceConfig.getDefaultFailTimes() > ciphertextData.getSendTimes())
+            {
+                // 插入失败队列
+                ciphertextData.setKey1(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_FAIL.getKey());
+                redisCache.setCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_FAIL.getKey(), key2, JSON.toJSONString(ciphertextData));
+                log.info("解密设备返回数据解密超时，加入到失败队列等待下次发送，数据标志:{}", key2);
+            }
+            else
+            {
+                log.info("解密设备返回数据解密超时，已过最大失败次数，数据删除，数据标志:{}", key2);
+            }
+        }
+        else
+        {
+            // 解密失败
+            redisCache.deleteCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), key2);
+
+            byte[] bytes = new byte[copyBuf.readableBytes() - 2];
+            copyBuf.getBytes(copyBuf.readerIndex(), bytes);
+            String payloadHex = MessageHandler.bytesToHexLog(bytes);
+            log.info("数据解密失败：{}", payloadHex);
+        }
+
+        return null;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/decoder/tax/CiphertextDataService.java

@@ -0,0 +1,238 @@
+package com.ruoyi.device.mqtt.handler.decoder.tax;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.ruoyi.common.core.domain.AjaxResult;
+import com.ruoyi.common.core.redis.RedisCache;
+import com.ruoyi.common.utils.http.HttpUtils;
+import com.ruoyi.device.config.DecryptDevice;
+import com.ruoyi.device.config.DecryptDeviceConfig;
+import com.ruoyi.device.mqtt.api.EmqxApiUtil;
+import com.ruoyi.device.mqtt.api.constants.ApiPathConstants;
+import com.ruoyi.device.mqtt.api.item.ClientInfo;
+import com.ruoyi.device.mqtt.api.item.EmqxResult;
+import com.ruoyi.device.mqtt.domain.decoder.tax.TaxTransferDataMessage;
+import com.ruoyi.device.mqtt.domain.encoder.tax.CiphertextDataDown;
+import com.ruoyi.device.mqtt.enums.DeviceRedisEnum;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.enums.VersionEnum;
+import com.ruoyi.device.mqtt.handler.HandlerManager;
+import com.ruoyi.device.mqtt.handler.decoder.MessageHandler;
+import com.ruoyi.device.mqtt.handler.encoder.IEncoder;
+import com.ruoyi.device.mqtt.util.MsgHandlerUtil;
+import com.ruoyi.device.mqtt.util.TargetTypeConvertUtil;
+import com.ruoyi.device.mqtt.vo.CiphertextData;
+import com.ruoyi.device.mqtt.vo.CommonHeader;
+import com.ruoyi.device.mqtt.vo.CommonTopic;
+import com.ruoyi.device.service.TaxTransferDataService;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import jakarta.annotation.Resource;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+
+import javax.xml.bind.DatatypeConverter;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 报税口密文数据解析与下发服务
+ *
+ * @author lwm
+ */
+@Component
+public class CiphertextDataService
+{
+    private static final Logger log = LoggerFactory.getLogger(CiphertextDataService.class);
+
+    @Resource
+    private HandlerManager handlerManager;
+
+    @Resource
+    private EmqxApiUtil emqxApiUtil;
+
+    @Resource
+    private DecryptDeviceConfig decryptDeviceConfig;
+
+    @Resource
+    private RedisCache redisCache;
+
+    @Resource
+    private TaxTransferDataService taxTransferDataService;
+
+    /**
+     * 组装数据 然后调用外部解密服务 或者 下发解密机
+     * @param topic topic
+     * @param header header
+     * @param body body
+     * @param raiseHangFlag 是否有抬挂枪
+     */
+    public Void decode(CommonTopic topic, CommonHeader header, ByteBuf body, boolean raiseHangFlag)
+    {
+        ByteBuf device = Unpooled.buffer(256);
+        CiphertextData ciphertextData = new CiphertextData();
+        Long gatewayNo;
+        Long collectorNo;
+        if (header.getProtoVer() == VersionEnum.THIRD.getValue())
+        {
+            TargetTypeConvertUtil.convert(body.readUnsignedShortLE());
+            gatewayNo = body.readUnsignedIntLE();
+            TargetTypeConvertUtil.convert(body.readUnsignedShortLE());
+            collectorNo = body.readUnsignedIntLE();
+        }
+        else
+        {
+            gatewayNo = body.readUnsignedIntLE();
+            collectorNo = body.readUnsignedIntLE();
+        }
+        device.writeIntLE(gatewayNo.intValue());
+        device.writeIntLE(collectorNo.intValue());
+        if (raiseHangFlag)
+        {
+            // 有抬挂枪
+            int gunStatus = body.readByte();
+            if (gunStatus == 1)
+            {
+                int batteryTime = body.readUnsignedShortLE();
+                int saveTime = body.readUnsignedShortLE();
+                long startTime = System.currentTimeMillis() - (batteryTime + saveTime) * 1000L;
+                long endTime = startTime + batteryTime * 1000L;
+                ciphertextData.setRaiseTime(new Date(startTime));
+                ciphertextData.setHangTime(new Date(endTime));
+            }
+            else
+            {
+                body.readUnsignedIntLE();
+            }
+        }
+
+        ByteBuf copyBuf = body.copy();
+
+        int taxNo = body.readByte();
+        int gunNo = body.readByte();
+        Long serialNo = body.readUnsignedIntLE();
+
+        body.readBytes(20);
+        int data14Len = body.readByte();
+        int data11Len = body.readByte();
+        if (data14Len != 0x20 || data11Len != 0x24)
+        {
+            byte[] bytes = new byte[copyBuf.readableBytes()];
+            copyBuf.getBytes(copyBuf.readerIndex(), bytes);
+            String payloadHex = MessageHandler.bytesToHexLog(bytes);
+            log.info("网关:{},采集器:{},密文数据包异常,忽略当前数据包：{}", gatewayNo, collectorNo, payloadHex);
+            return null;
+        }
+
+        byte[] deviceBytes = new byte[device.readableBytes()];
+        device.getBytes(device.readerIndex(), deviceBytes);
+
+        // 去掉crc16
+        byte[] bytes = new byte[copyBuf.readableBytes() - 2];
+        copyBuf.getBytes(copyBuf.readerIndex(), bytes);
+
+        byte[] dataBytes = ArrayUtils.addAll(deviceBytes, bytes);
+
+        // hash 存储
+        String key2 = String.format(ApiPathConstants.FOUR_UNDER_LINE, gatewayNo, collectorNo, taxNo, gunNo, serialNo);
+
+        ciphertextData.setDeviceSn(gatewayNo);
+        ciphertextData.setDeviceType(topic.getDeviceType());
+        ciphertextData.setSendTimes(0);
+        ciphertextData.setData(DatatypeConverter.printHexBinary(dataBytes));
+        ciphertextData.setDateTime(new Date());
+        ciphertextData.setKey2(key2);
+
+        if (Boolean.TRUE.equals(decryptDeviceConfig.getOutDecrypt()))
+        {
+            // 调用远程接口解析
+            String post = HttpUtils.sendPost(decryptDeviceConfig.getOutDecryptUrl(), JSON.toJSONString(ciphertextData), MediaType.APPLICATION_JSON_VALUE);
+            log.info("调用外部解密结果:{}", post);
+
+            if (StringUtils.isNotBlank(post))
+            {
+                AjaxResult ajaxResult = JSON.parseObject(post, AjaxResult.class);
+                if (ajaxResult != null &&
+                        ajaxResult.get(AjaxResult.DATA_TAG) != null && ajaxResult.get(AjaxResult.DATA_TAG) instanceof JSONObject data)
+                {
+                    log.info("外部解密成功,调用业务服务保存数据:{}", data);
+                    TaxTransferDataMessage taxTransferDataMessage = data.to(TaxTransferDataMessage.class);
+                    taxTransferDataService.saveTaxTransferData(taxTransferDataMessage);
+                }
+            }
+        }
+        else
+        {
+            if (!send(ciphertextData))
+            {
+                ciphertextData.setKey1(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_ORIGNAL.getKey());
+                redisCache.setCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_ORIGNAL.getKey(), key2, JSON.toJSONString(ciphertextData));
+                log.info("收到设备：{} 发送的待解密数据，已存入原始队列，等待任务执行！", key2);
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 发送待解密数据
+     *
+     * @param ciphertextData 待解密数据
+     */
+    private boolean send(CiphertextData ciphertextData)
+    {
+        for (DecryptDevice decryptDevice : decryptDeviceConfig.getDecryptDeviceList())
+        {
+            EmqxResult<List<ClientInfo>> clientInfo = emqxApiUtil.getClientInfo(
+                decryptDevice.getDeviceType(), decryptDevice.getDeviceSn());
+            if (clientInfo.getData() != null && !clientInfo.getData().isEmpty())
+            {
+                log.info("解密设备：{} 在线", decryptDevice.getDeviceSn());
+
+                // 默认发送条数
+                int freeSpace = decryptDeviceConfig.getDefaultSendSize();
+                // 获取剩余空间
+                String obj = redisCache.getCacheObject(
+                    String.format(DeviceRedisEnum.DEVICE_FREE_SPACE.getKey(), decryptDevice.getDeviceSn()));
+                if (StringUtils.isNotBlank(obj))
+                {
+                    freeSpace = Integer.parseInt(obj);
+                }
+
+                if (freeSpace > 1)
+                {
+                    send(ciphertextData, decryptDevice);
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    /**
+     * 发送数据到解密版
+     * 并移除队列加入到发送队列
+     * @param ciphertextData 待解密数据
+     * @param decryptDevice 解密设备
+     */
+    @SuppressWarnings("unchecked")
+    private void send(CiphertextData ciphertextData, DecryptDevice decryptDevice)
+    {
+        CiphertextDataDown request = new CiphertextDataDown();
+        request.setCiphertextData(DatatypeConverter.parseHexBinary(ciphertextData.getData()));
+        request.setDeviceSn(decryptDevice.getDeviceSn());
+        request.setDeviceType(decryptDevice.getDeviceType());
+        String key = MsgHandlerUtil.getEncoderKey(MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_DOWN);
+        IEncoder<CiphertextDataDown> encoder = (IEncoder<CiphertextDataDown>) handlerManager.getEncoder(key);
+        encoder.encode(request);
+
+        // 添加发送队列 发送次数+1
+        ciphertextData.setSendTimes(ciphertextData.getSendTimes() + 1);
+        ciphertextData.setSendTime(System.currentTimeMillis());
+        ciphertextData.setKey1(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_ORIGNAL.getKey());
+        redisCache.setCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), ciphertextData.getKey2(), JSON.toJSONString(ciphertextData));
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/decoder/tax/CiphertextRaiseHangDataDecoder.java

@@ -0,0 +1,28 @@
+package com.ruoyi.device.mqtt.handler.decoder.tax;
+
+import com.ruoyi.device.mqtt.annotation.ConsumerHandler;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.handler.decoder.AbstractDecoder;
+import com.ruoyi.device.mqtt.vo.CommonHeader;
+import com.ruoyi.device.mqtt.vo.CommonTopic;
+import io.netty.buffer.ByteBuf;
+import jakarta.annotation.Resource;
+
+/**
+ * 报税口密文数据上行解码（带抬挂枪）
+ *
+ * @author lwm
+ */
+@ConsumerHandler(msgType = MsgTypeEnum.CIPHERTEXT_RAISE_HANG_DATA_UP)
+public class CiphertextRaiseHangDataDecoder extends AbstractDecoder<Void>
+{
+    @Resource
+    private CiphertextDataService ciphertextDataService;
+
+    @Override
+    protected Void decode(CommonTopic topic, CommonHeader header, ByteBuf body)
+    {
+        ciphertextDataService.decode(topic, header, body, true);
+        return null;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/encoder/tax/CiphertextDataDecryptEncoder.java

@@ -0,0 +1,45 @@
+package com.ruoyi.device.mqtt.handler.encoder.tax;
+
+import com.ruoyi.device.mqtt.annotation.ProducerHandler;
+import com.ruoyi.device.mqtt.domain.encoder.tax.CiphertextDataDown;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.enums.QosEnum;
+import com.ruoyi.device.mqtt.handler.encoder.AbstractEncoder;
+import io.netty.buffer.ByteBuf;
+
+/**
+ * 报税口密文数据 解密下发编码
+ */
+@ProducerHandler(msgType = MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_DOWN)
+public class CiphertextDataDecryptEncoder extends AbstractEncoder<CiphertextDataDown>
+{
+    @Override
+    protected void encode(CiphertextDataDown request, ByteBuf body)
+    {
+        body.writeBytes(request.getCiphertextData());
+    }
+
+    @Override
+    protected String topic(CiphertextDataDown request)
+    {
+        return generateTopic(request);
+    }
+
+    @Override
+    protected MsgTypeEnum msgType()
+    {
+        return MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_DOWN;
+    }
+
+    @Override
+    protected QosEnum qos()
+    {
+        return QosEnum.QoS2;
+    }
+
+    @Override
+    protected boolean retain()
+    {
+        return false;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/handler/encoder/tax/CiphertextDataDecryptReturnEncoder.java

@@ -0,0 +1,45 @@
+package com.ruoyi.device.mqtt.handler.encoder.tax;
+
+import com.ruoyi.device.mqtt.annotation.ProducerHandler;
+import com.ruoyi.device.mqtt.domain.encoder.tax.CiphertextDataDown;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.enums.QosEnum;
+import com.ruoyi.device.mqtt.handler.encoder.AbstractEncoder;
+import io.netty.buffer.ByteBuf;
+
+/**
+ * 报税口密文数据返回测试板下发编码
+ */
+@ProducerHandler(msgType = MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_RETURN_DOWN)
+public class CiphertextDataDecryptReturnEncoder extends AbstractEncoder<CiphertextDataDown>
+{
+    @Override
+    protected void encode(CiphertextDataDown request, ByteBuf body)
+    {
+        body.writeBytes(request.getCiphertextData());
+    }
+
+    @Override
+    protected String topic(CiphertextDataDown request)
+    {
+        return generateTopic(request);
+    }
+
+    @Override
+    protected MsgTypeEnum msgType()
+    {
+        return MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_RETURN_DOWN;
+    }
+
+    @Override
+    protected QosEnum qos()
+    {
+        return QosEnum.QoS2;
+    }
+
+    @Override
+    protected boolean retain()
+    {
+        return false;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/util/TargetTypeConvertUtil.java

@@ -0,0 +1,28 @@
+package com.ruoyi.device.mqtt.util;
+
+/**
+ * 设备类型转换工具
+ *
+ * @author lwm
+ */
+public final class TargetTypeConvertUtil
+{
+    private TargetTypeConvertUtil()
+    {
+    }
+
+    /**
+     * 设备类型转换
+     * @param targetType 设备类型（上报的初始数据）
+     * @return 设备类型（转换后的数据）
+     */
+    public static String convert(Integer targetType)
+    {
+        String deviceType = Integer.toHexString(targetType);
+        if (deviceType.length() == 3)
+        {
+            return "0" + deviceType;
+        }
+        return deviceType;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/mqtt/vo/CiphertextData.java

@@ -0,0 +1,175 @@
+package com.ruoyi.device.mqtt.vo;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * 报税口密文数据
+ *
+ * @author lwm
+ */
+public class CiphertextData implements Serializable
+{
+    private static final long serialVersionUID = 1L;
+ 
+    /**
+     * 设备 SN码
+     */
+    private Long deviceSn;
+
+    /**
+     * 设备类型
+     */
+    private String deviceType;
+
+    /**
+     * 密文数据（十六进制字符串）
+     */
+    private String data;
+
+    /**
+     * 发送次数（重试计数）
+     */
+    private Integer sendTimes;
+
+    /**
+     * 发送时间戳（毫秒）
+     */
+    private Long sendTime;
+
+    /**
+     * 数据接收时间
+     */
+    private Date dateTime;
+
+    /**
+     * Redis Hash 一级键（数据类型标识）
+     */
+    private String key1;
+
+    /**
+     * Redis Hash 二级键（唯一标识：网关_采集器_税号_枪号_流水号）
+     */
+    private String key2;
+
+    /**
+     * 提枪时间
+     */
+    private Date raiseTime;
+
+    /**
+     * 挂枪时间
+     */
+    private Date hangTime;
+
+    /**
+     * 平台标识（用于区分不同的业务平台）
+     */
+    private String platform;
+
+    public Long getDeviceSn() {
+        return deviceSn;
+    }
+
+    public void setDeviceSn(Long deviceSn) {
+        this.deviceSn = deviceSn;
+    }
+
+    public String getDeviceType() {
+        return deviceType;
+    }
+
+    public void setDeviceType(String deviceType) {
+        this.deviceType = deviceType;
+    }
+
+    public String getData()
+    {
+        return data;
+    }
+
+    public void setData(String data)
+    {
+        this.data = data;
+    }
+
+    public Integer getSendTimes()
+    {
+        return sendTimes;
+    }
+
+    public void setSendTimes(Integer sendTimes)
+    {
+        this.sendTimes = sendTimes;
+    }
+
+    public Long getSendTime()
+    {
+        return sendTime;
+    }
+
+    public void setSendTime(Long sendTime)
+    {
+        this.sendTime = sendTime;
+    }
+
+    public Date getDateTime()
+    {
+        return dateTime;
+    }
+
+    public void setDateTime(Date dateTime)
+    {
+        this.dateTime = dateTime;
+    }
+
+    public String getKey1()
+    {
+        return key1;
+    }
+
+    public void setKey1(String key1)
+    {
+        this.key1 = key1;
+    }
+
+    public String getKey2()
+    {
+        return key2;
+    }
+
+    public void setKey2(String key2)
+    {
+        this.key2 = key2;
+    }
+
+    public Date getRaiseTime()
+    {
+        return raiseTime;
+    }
+
+    public void setRaiseTime(Date raiseTime)
+    {
+        this.raiseTime = raiseTime;
+    }
+
+    public Date getHangTime()
+    {
+        return hangTime;
+    }
+
+    public void setHangTime(Date hangTime)
+    {
+        this.hangTime = hangTime;
+    }
+
+    public String getPlatform()
+    {
+        return platform;
+    }
+
+    public void setPlatform(String platform)
+    {
+        this.platform = platform;
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/service/TaxTransferDataService.java

@@ -0,0 +1,16 @@
+package com.ruoyi.device.service;
+
+import com.ruoyi.device.mqtt.domain.decoder.tax.TaxTransferDataMessage;
+
+/**
+ * 报税口密文解密结果处理（替代 Kafka 推送，直接方法调用）
+ */
+public interface TaxTransferDataService
+{
+    /**
+     * 处理解密成功的报税口交易数据
+     *
+     * @param message 解密结果
+     */
+    void saveTaxTransferData(TaxTransferDataMessage message);
+}

ruoyi-device/src/main/java/com/ruoyi/device/service/impl/TaxTransferDataServiceImpl.java

@@ -0,0 +1,23 @@
+package com.ruoyi.device.service.impl;
+
+import com.alibaba.fastjson2.JSON;
+import com.ruoyi.device.mqtt.domain.decoder.tax.TaxTransferDataMessage;
+import com.ruoyi.device.service.TaxTransferDataService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+/**
+ * 报税口密文解密结果默认处理
+ */
+@Service
+public class TaxTransferDataServiceImpl implements TaxTransferDataService
+{
+    private static final Logger log = LoggerFactory.getLogger(TaxTransferDataServiceImpl.class);
+
+    @Override
+    public void saveTaxTransferData(TaxTransferDataMessage message)
+    {
+        log.info("收到报税口密文解密数据：{}", JSON.toJSONString(message));
+    }
+}

ruoyi-device/src/main/java/com/ruoyi/device/task/CiphertextDataHandler.java

@@ -0,0 +1,215 @@
+package com.ruoyi.device.task;
+
+import com.alibaba.fastjson2.JSON;
+import com.ruoyi.common.core.redis.RedisCache;
+import com.ruoyi.device.config.DecryptDevice;
+import com.ruoyi.device.config.DecryptDeviceConfig;
+import com.ruoyi.device.mqtt.enums.DeviceRedisEnum;
+import com.ruoyi.device.mqtt.api.EmqxApiUtil;
+import com.ruoyi.device.mqtt.api.item.ClientInfo;
+import com.ruoyi.device.mqtt.api.item.EmqxResult;
+import com.ruoyi.device.mqtt.domain.encoder.tax.CiphertextDataDown;
+import com.ruoyi.device.mqtt.enums.MsgTypeEnum;
+import com.ruoyi.device.mqtt.handler.HandlerManager;
+import com.ruoyi.device.mqtt.handler.encoder.IEncoder;
+import com.ruoyi.device.mqtt.util.MsgHandlerUtil;
+import com.ruoyi.device.mqtt.vo.CiphertextData;
+import jakarta.annotation.Resource;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import javax.xml.bind.DatatypeConverter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 密文解密定时调度：失败重试与待解密队列下发
+ */
+@Component("ciphertextDataHandler")
+public class CiphertextDataHandler
+{
+    private static final Logger log = LoggerFactory.getLogger(CiphertextDataHandler.class);
+
+    @Resource
+    private DecryptDeviceConfig decryptDeviceConfig;
+
+    @Resource
+    private HandlerManager handlerManager;
+
+    @Resource
+    private EmqxApiUtil emqxApiUtil;
+
+    @Resource
+    private RedisCache redisCache;
+
+    /**
+     * 密文解密定时调度：失败重试与待解密队列下发
+     */
+    public void sendCiphertextDataDecrypt()
+    {
+        // 检查是否有超时未返回的数据
+        Map<String, CiphertextData> timeOutQueue = redisCache.getCacheMap(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_FAIL.getKey());
+        if (timeOutQueue != null && !timeOutQueue.isEmpty())
+        {
+            Long currentTime = System.currentTimeMillis();
+
+            for (CiphertextData ciphertextData : timeOutQueue.values())
+            {
+                if (currentTime - ciphertextData.getSendTime() > decryptDeviceConfig.getTimeout())
+                {
+                    redisCache.deleteCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), ciphertextData.getKey2());
+                    if (decryptDeviceConfig.getDefaultFailTimes() > ciphertextData.getSendTimes())
+                    {
+                        // 插入失败队列
+                        ciphertextData.setKey1(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_FAIL.getKey());
+                        redisCache.setCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_FAIL.getKey(), ciphertextData.getKey2(), JSON.toJSONString(ciphertextData));
+                        log.info("已发送数据解密超时，加入到失败队列等待下次发送");
+                    }
+                    else
+                    {
+                        log.info("已发送数据解密超时，已过最大失败次数，数据删除");
+                    }
+                }
+            }
+        }
+
+        List<DecryptDevice> conversion = new ArrayList<>();
+        // 先判断谁在线
+        for (DecryptDevice decryptDevice : decryptDeviceConfig.getDecryptDeviceList())
+        {
+            EmqxResult<List<ClientInfo>> clientInfo = emqxApiUtil.getClientInfo(decryptDevice.getDeviceType(), decryptDevice.getDeviceSn());
+            if (clientInfo.getData() != null && !clientInfo.getData().isEmpty())
+            {
+                conversion.add(decryptDevice);
+                log.info("解密设备：{} 在线", decryptDevice.getDeviceSn());
+            }
+        }
+
+        List<CiphertextData> waitQueue = getWaitQueue();
+        if (waitQueue.isEmpty())
+        {
+            log.info("待解密队列为空");
+            return;
+        }
+
+        for (DecryptDevice decryptDevice : conversion)
+        {
+            // 默认发送条数
+            int freeSpace = decryptDeviceConfig.getDefaultSendSize();
+            // 获取剩余空间
+            String obj = redisCache.getCacheObject(
+                String.format(DeviceRedisEnum.DEVICE_FREE_SPACE.getKey(), decryptDevice.getDeviceSn()));
+            if (StringUtils.isNotBlank(obj))
+            {
+                freeSpace = Integer.parseInt(obj);
+            }
+
+            List<List<CiphertextData>> list = taskSplit(waitQueue, freeSpace);
+            if (list.isEmpty())
+            {
+                log.info("解密设备:{},未分配到解密任务，结束！", decryptDevice.getDeviceSn());
+                break;
+            }
+
+            if (list.size() > 1)
+            {
+                waitQueue = list.get(1);
+                log.info("解密设备:{},分配到解密任务数量:{}，剩余待解密队列数量:{}！",
+                    decryptDevice.getDeviceSn(), list.get(0).size(), list.get(1).size());
+            }
+            else
+            {
+                log.info("解密设备:{},分配到解密任务数量:{}，剩余待解密队列数量:0,开始发送数据到解密板！",
+                    decryptDevice.getDeviceSn(), list.get(0).size());
+                waitQueue = null;
+            }
+
+            for (CiphertextData ciphertextData : list.get(0))
+            {
+                send(ciphertextData, decryptDevice);
+            }
+        }
+    }
+
+    /**
+     * 待解密队列任务分配 获取指定数量合集 和剩余合集
+     *
+     * @param waitQueue 待解密队列
+     * @param size 分配数量
+     */
+    private List<List<CiphertextData>> taskSplit(List<CiphertextData> waitQueue, int size)
+    {
+        // 存放结果的集合
+        List<List<CiphertextData>> result = new ArrayList<>();
+        if (waitQueue != null && !waitQueue.isEmpty())
+        {
+            if (waitQueue.size() >= size)
+            {
+                List<CiphertextData> sublist = waitQueue.subList(0, size);
+//                for (int startIndex = 0; startIndex < waitQueue.size(); startIndex += size)
+//                {
+//                    List<CiphertextData> sublist = waitQueue.subList(startIndex, Math.min(startIndex + size, waitQueue.size()));
+                    result.add(sublist);
+//                }
+            }
+            else
+            {
+                result.add(waitQueue);
+            }
+        }
+        return result;
+    }
+
+    /**
+     * 获取待解密队列
+     */
+    private List<CiphertextData> getWaitQueue()
+    {
+        Map<String, CiphertextData> failQueue = redisCache.getCacheMap(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_FAIL.getKey());
+        Map<String, CiphertextData> originalQueue = redisCache.getCacheMap(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_ORIGNAL.getKey());
+
+        List<CiphertextData> result = new ArrayList<>();
+        if (failQueue != null && !failQueue.isEmpty())
+        {
+            result.addAll(failQueue.values());
+            log.info("失败队列数据条数：{}", failQueue.size());
+        }
+
+        if (originalQueue != null && !originalQueue.isEmpty())
+        {
+            result.addAll(originalQueue.values());
+            log.info("原始队列数据条数：{}", originalQueue.size());
+        }
+
+        return result;
+    }
+
+    /**
+     * 发送数据到解密版
+     * 并移除队列加入到发送队列
+     *
+     * @param ciphertextData 待解密数据
+     * @param decryptDevice 解密设备
+     */
+    @SuppressWarnings("unchecked")
+    private void send(CiphertextData ciphertextData, DecryptDevice decryptDevice)
+    {
+        CiphertextDataDown request = new CiphertextDataDown();
+        request.setCiphertextData(DatatypeConverter.parseHexBinary(ciphertextData.getData()));
+        request.setDeviceSn(decryptDevice.getDeviceSn());
+        request.setDeviceType(decryptDevice.getDeviceType());
+        String key = MsgHandlerUtil.getEncoderKey(MsgTypeEnum.CIPHERTEXT_DATA_DECRYPT_DOWN);
+        IEncoder<CiphertextDataDown> encoder = (IEncoder<CiphertextDataDown>) handlerManager.getEncoder(key);
+        encoder.encode(request);
+        // 队列中移除
+        redisCache.deleteCacheMapValue(ciphertextData.getKey1(), ciphertextData.getKey2());
+
+        // 添加 发送队列 发送次数+1
+        ciphertextData.setSendTimes(ciphertextData.getSendTimes() + 1);
+        ciphertextData.setSendTime(System.currentTimeMillis());
+        redisCache.setCacheMapValue(DeviceRedisEnum.TAX_CIPHERTEXT_DATA_SEND.getKey(), ciphertextData.getKey2(), ciphertextData);
+    }
+}