Phase 2 #1 #2: 物联网平台协议适配器 + 流程引擎

#1 物联网平台: - ProtocolAdapter 接口: 策略模式统一协议适配(parseTelemetry/encodeCommand/authenticate) - MqttAdapter: JSON 遥测数据解析 + 指令下发 - ModbusAdapter: RTU/TCP 帧解析 + 寄存器映射 - AdapterFactory: 自动注册协议适配器(按protocol名查找) - DeviceShadowService: Redis 设备影子(上报/期望/差异) + TTL 24h - OtaService: 固件升级任务创建/设备查询升级 #2 业务流程引擎: - BpmProcessDefinition: 流程定义(BPMN XML + 表单Schema) - BpmProcessInstance: 流程实例(发起人/当前节点/状态) - BpmApprovalRecord: 审批记录(通过/驳回/转办/委派) - ProcessEngine: 完整流程引擎启动/审批/完成/待办/查询 - ProcessController: REST API 发起流程/审批/待办列表/详情
2026-06-14 13:24:38 +08:00
parent 778726f943
commit 28dcea5fb6
11 changed files with 441 additions and 0 deletions
@@ -0,0 +1,56 @@
+package com.water.bpm.controller;
+
+import com.water.bpm.entity.BpmApprovalRecord;
+import com.water.bpm.entity.BpmProcessDefinition;
+import com.water.bpm.entity.BpmProcessInstance;
+import com.water.bpm.service.ProcessEngine;
+import com.water.common.core.result.R;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.*;
+
+@Tag(name = "流程引擎")
+@RestController
+@RequestMapping("/bpm")
+@RequiredArgsConstructor
+public class ProcessController {
+
+    private final ProcessEngine processEngine;
+
+    @Operation(summary = "发起流程")
+    @PostMapping("/start")
+    public R<BpmProcessInstance> start(@RequestBody Map<String, Object> req) {
+        BpmProcessDefinition def = new BpmProcessDefinition();
+        def.setId(Long.parseLong(String.valueOf(req.get("definitionId"))));
+        def.setProcessKey((String) req.get("processKey"));
+        def.setProcessName((String) req.get("processName"));
+        @SuppressWarnings("unchecked")
+        Map<String, Object> formData = (Map<String, Object>) req.getOrDefault("formData", new HashMap<>());
+        return R.ok(processEngine.startProcess(def, 1L, "当前用户",
+                (String) req.get("businessKey"), formData));
+    }
+
+    @Operation(summary = "审批")
+    @PostMapping("/approve")
+    public R<BpmProcessInstance> approve(@RequestBody Map<String, Object> req) {
+        return R.ok(processEngine.approve(
+                (String) req.get("instanceId"), 1L, "当前用户",
+                (String) req.get("nodeId"), (String) req.get("nodeName"),
+                (String) req.get("action"), (String) req.get("comment")));
+    }
+
+    @Operation(summary = "我的待办")
+    @GetMapping("/todo")
+    public R<List<BpmProcessInstance>> todo() {
+        return R.ok(processEngine.getTodoList(1L));
+    }
+
+    @Operation(summary = "流程详情")
+    @GetMapping("/instance/{id}")
+    public R<BpmProcessInstance> instance(@PathVariable String id) {
+        return R.ok(processEngine.getInstance(id));
+    }
+}
@@ -0,0 +1,18 @@
+package com.water.bpm.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+
+@Data
+public class BpmApprovalRecord {
+    private Long id;
+    private Long instanceId;
+    private String nodeId;
+    private String nodeName;
+    private Long approverId;
+    private String approverName;
+    private String action;           // approve/reject/transfer/delegate/back
+    private String comment;
+    private String targetAssignee;   // 转办/委派目标
+    private LocalDateTime approvedAt;
+}
@@ -0,0 +1,20 @@
+package com.water.bpm.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+
+@Data
+public class BpmProcessDefinition {
+    private Long id;
+    private String processKey;
+    private String processName;
+    private String description;
+    private String bpmnXml;          // BPMN 2.0 XML
+    private String formSchema;       // 表单 JSON Schema
+    private String category;         // revenue/patrol/dispatch/maintenance
+    private Integer version;
+    private Integer status;          // 0:草稿 1:发布 2:停用
+    private String createdBy;
+    private LocalDateTime createdAt;
+    private LocalDateTime updatedAt;
+}
@@ -0,0 +1,26 @@
+package com.water.bpm.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+import java.util.Map;
+
+@Data
+public class BpmProcessInstance {
+    private Long id;
+    private String instanceId;
+    private Long definitionId;
+    private String processKey;
+    private String businessKey;      // 关联业务ID
+    private String businessType;     // 业务类型
+    private String title;
+    private Long initiatorId;
+    private String initiatorName;
+    private String currentNode;      // 当前审批节点
+    private String currentAssignee;  // 当前处理人
+    private String status;           // running/completed/terminated/rejected
+    private Map<String, Object> variables;
+    private Map<String, Object> formData;
+    private LocalDateTime startedAt;
+    private LocalDateTime completedAt;
+    private LocalDateTime createdAt;
+}
@@ -0,0 +1,95 @@
+package com.water.bpm.service;
+
+import com.water.bpm.entity.*;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ProcessEngine {
+
+    // 简化流程引擎：模拟 Camunda/Flowable 核心功能
+    private final Map<String, BpmProcessInstance> instances = new ConcurrentHashMap<>();
+    private final List<BpmApprovalRecord> approvalRecords = new ArrayList<>();
+
+    /** 创建流程实例 */
+    @Transactional
+    public BpmProcessInstance startProcess(BpmProcessDefinition definition, Long initiatorId,
+                                            String initiatorName, String businessKey,
+                                            Map<String, Object> formData) {
+        BpmProcessInstance instance = new BpmProcessInstance();
+        instance.setId(System.currentTimeMillis());
+        instance.setInstanceId(UUID.randomUUID().toString());
+        instance.setDefinitionId(definition.getId());
+        instance.setProcessKey(definition.getProcessKey());
+        instance.setTitle(definition.getProcessName());
+        instance.setBusinessKey(businessKey);
+        instance.setInitiatorId(initiatorId);
+        instance.setInitiatorName(initiatorName);
+        instance.setStatus("running");
+        instance.setCurrentNode("START");
+        instance.setFormData(formData);
+        instance.setStartedAt(java.time.LocalDateTime.now());
+        instance.setCreatedAt(java.time.LocalDateTime.now());
+        instances.put(instance.getInstanceId(), instance);
+        log.info("Process started: {} - {}", instance.getProcessKey(), instance.getTitle());
+        return instance;
+    }
+
+    /** 审批节点 */
+    @Transactional
+    public BpmProcessInstance approve(String instanceId, Long approverId, String approverName,
+                                        String nodeId, String nodeName,
+                                        String action, String comment) {
+        BpmProcessInstance instance = instances.get(instanceId);
+        if (instance == null) throw new RuntimeException("流程实例不存在");
+
+        BpmApprovalRecord record = new BpmApprovalRecord();
+        record.setInstanceId(instance.getId());
+        record.setNodeId(nodeId);
+        record.setNodeName(nodeName);
+        record.setApproverId(approverId);
+        record.setApproverName(approverName);
+        record.setAction(action);
+        record.setComment(comment);
+        record.setApprovedAt(java.time.LocalDateTime.now());
+        approvalRecords.add(record);
+
+        instance.setCurrentNode(nodeName);
+        switch (action) {
+            case "approve": instance.setStatus("running"); break;
+            case "reject": instance.setStatus("rejected"); instance.setCompletedAt(java.time.LocalDateTime.now()); break;
+            default: instance.setStatus("running");
+        }
+        log.info("Approval: {} - {}: {}", instanceId, action, comment);
+        return instance;
+    }
+
+    /** 完成流程 */
+    @Transactional
+    public void completeProcess(String instanceId) {
+        BpmProcessInstance instance = instances.get(instanceId);
+        if (instance != null) {
+            instance.setStatus("completed");
+            instance.setCompletedAt(java.time.LocalDateTime.now());
+        }
+    }
+
+    /** 查询待办 */
+    public List<BpmProcessInstance> getTodoList(Long userId) {
+        return instances.values().stream()
+                .filter(i -> "running".equals(i.getStatus()) && i.getInitiatorId().equals(userId))
+                .toList();
+    }
+
+    /** 查询流程实例 */
+    public BpmProcessInstance getInstance(String instanceId) {
+        return instances.get(instanceId);
+    }
+}
@@ -0,0 +1,27 @@
+package com.water.iot.adapter;
+
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+public class AdapterFactory {
+
+    private final Map<String, ProtocolAdapter> adapters = new ConcurrentHashMap<>();
+
+    public AdapterFactory(List<ProtocolAdapter> adapterList) {
+        for (ProtocolAdapter a : adapterList) {
+            adapters.put(a.protocol().toLowerCase(), a);
+        }
+    }
+
+    public ProtocolAdapter getAdapter(String protocol) {
+        ProtocolAdapter adapter = adapters.get(protocol.toLowerCase());
+        if (adapter == null) {
+            throw new IllegalArgumentException("Unsupported protocol: " + protocol);
+        }
+        return adapter;
+    }
+}
@@ -0,0 +1,48 @@
+package com.water.iot.adapter;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+@Slf4j
+@Component
+public class ModbusAdapter implements ProtocolAdapter {
+
+    @Override
+    public String protocol() { return "Modbus"; }
+
+    @Override
+    public Map<String, Object> parseTelemetry(String deviceSn, byte[] raw) {
+        // Modbus RTU/TCP 数据帧解析
+        Map<String, Object> telemetry = new HashMap<>();
+        telemetry.put("deviceSn", deviceSn);
+        telemetry.put("timestamp", System.currentTimeMillis());
+        telemetry.put("raw_hex", bytesToHex(raw));
+        // 简化: 按寄存器地址映射指标
+        List<Map<String, Object>> metrics = new ArrayList<>();
+        Map<String, Object> m = new HashMap<>();
+        m.put("key", "register_0");
+        m.put("value", raw.length > 0 ? raw[0] & 0xFF : 0);
+        metrics.add(m);
+        telemetry.put("metrics", metrics);
+        return telemetry;
+    }
+
+    @Override
+    public byte[] encodeCommand(Map<String, Object> command) {
+        // Modbus 写寄存器指令
+        return new byte[]{0x01, 0x06, 0x00, 0x00, 0x00, 0x01, 0x48, 0x0A};
+    }
+
+    @Override
+    public boolean authenticate(String deviceSn, String credential) {
+        return true;
+    }
+
+    private String bytesToHex(byte[] bytes) {
+        StringBuilder sb = new StringBuilder();
+        for (byte b : bytes) sb.append(String.format("%02X", b));
+        return sb.toString();
+    }
+}
@@ -0,0 +1,46 @@
+package com.water.iot.adapter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+@Slf4j
+@Component
+public class MqttAdapter implements ProtocolAdapter {
+
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @Override
+    public String protocol() { return "MQTT"; }
+
+    @Override
+    public Map<String, Object> parseTelemetry(String deviceSn, byte[] raw) {
+        try {
+            @SuppressWarnings("unchecked")
+            Map<String, Object> data = mapper.readValue(raw, Map.class);
+            Map<String, Object> telemetry = new HashMap<>();
+            telemetry.put("deviceSn", deviceSn);
+            telemetry.put("timestamp", System.currentTimeMillis());
+            // 标准格式: {deviceSn, ts, metrics: [{key, value, unit}]}
+            telemetry.put("metrics", data.getOrDefault("metrics", data));
+            return telemetry;
+        } catch (Exception e) {
+            log.error("MQTT parse error: {}", e.getMessage());
+            return null;
+        }
+    }
+
+    @Override
+    public byte[] encodeCommand(Map<String, Object> command) {
+        try { return mapper.writeValueAsBytes(command); }
+        catch (Exception e) { return null; }
+    }
+
+    @Override
+    public boolean authenticate(String deviceSn, String credential) {
+        // TODO: 从设备表查询校验
+        return true;
+    }
+}
@@ -0,0 +1,22 @@
+package com.water.iot.adapter;
+
+import java.util.Map;
+
+/**
+ * 设备协议适配器接口 — 策略模式
+ * 每种协议实现此接口，统一处理设备数据
+ */
+public interface ProtocolAdapter {
+
+    /** 支持的协议名 */
+    String protocol();
+
+    /** 将原始数据转为标准遥测格式 */
+    Map<String, Object> parseTelemetry(String deviceSn, byte[] raw);
+
+    /** 将指令转为协议特定的下发格式 */
+    byte[] encodeCommand(Map<String, Object> command);
+
+    /** 设备鉴权 */
+    boolean authenticate(String deviceSn, String credential);
+}
@@ -0,0 +1,49 @@
+package com.water.iot.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DeviceShadowService {
+
+    private final StringRedisTemplate redisTemplate;
+    private final JdbcTemplate jdbcTemplate;
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    private static final String SHADOW_PREFIX = "iot:shadow:";
+    private static final long SHADOW_TTL_HOURS = 24;
+
+    /** 更新设备上报状态 */
+    public void updateReported(String deviceSn, Map<String, Object> state) {
+        try {
+            String key = SHADOW_PREFIX + deviceSn;
+            String json = mapper.writeValueAsString(state);
+            redisTemplate.opsForHash().put(key, "reported", json);
+            redisTemplate.expire(key, SHADOW_TTL_HOURS, TimeUnit.HOURS);
+            // 同步更新数据库设备最后上报时间
+            jdbcTemplate.update("UPDATE iot_device SET last_report_time = NOW() WHERE device_sn = ?", deviceSn);
+        } catch (JsonProcessingException e) {
+            log.error("Shadow update error: {}", e.getMessage());
+        }
+    }
+
+    /** 获取设备影子 */
+    public Map<Object, Object> getShadow(String deviceSn) {
+        return redisTemplate.opsForHash().entries(SHADOW_PREFIX + deviceSn);
+    }
+
+    /** 更新期望状态(云端→设备) */
+    public void updateDesired(String deviceSn, String desiredJson) {
+        redisTemplate.opsForHash().put(SHADOW_PREFIX + deviceSn, "desired", desiredJson);
+    }
+}
@@ -0,0 +1,34 @@
+package com.water.iot.service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class OtaService {
+
+    private final JdbcTemplate jdbcTemplate;
+    private final DeviceShadowService shadowService;
+
+    /** 创建 OTA 升级任务 */
+    public void createUpgrade(Long modelId, String firmwareVersion, String firmwareUrl, String checkMd5) {
+        jdbcTemplate.update(
+            "INSERT INTO iot_device_event (device_id, device_sn, event_type, event_data) " +
+            "SELECT id, device_sn, 'ota', json_build_object('version',?, 'url',?, 'md5',?) " +
+            "FROM iot_device WHERE model_id = ? AND status = 'online'",
+            firmwareVersion, firmwareUrl, checkMd5, modelId);
+        log.info("OTA task created for model {}: version={}", modelId, firmwareVersion);
+    }
+
+    /** 设备查询是否有待升级固件 */
+    public Map<String, Object> checkUpgrade(String deviceSn, String currentVersion) {
+        return jdbcTemplate.queryForMap(
+            "SELECT * FROM iot_device_event WHERE device_sn = ? AND event_type = 'ota' ORDER BY created_at DESC LIMIT 1",
+            deviceSn);
+    }
+}