物联网实战：Spring Boot + Netty 搭建 MQTT

// 创建设备身份
DeviceIdentity identity = new DeviceIdentity("product001", "device001");

// 创建内部消息
InternalMessage message = InternalMessage.builder()
    .deviceId(identity)
    .topic("sensors/temperature")
    .payload("{\"temp\": 25.5}")
    .qos(1)
    .timestamp(System.currentTimeMillis())
    .build();

// 1. 实现协议适配器
@Component
public class MyProtocolAdapter implements ProtocolAdapter {
    
    @Override
    public ProtocolType supportProtocol() {
        return ProtocolType.MY_CUSTOM;
    }
    
    @Override
    public InternalMessage decode(ByteBuf buffer) {
        // 解析自定义协议字节流
        // ...
        return message;
    }
    
    @Override
    public ByteBuf encode(InternalMessage message) {
        // 编码为自定义协议字节流
        // ...
        return buffer;
    }
}

// 2. 在 ProtocolDetector 中添加识别规则
private ProtocolType detectProtocol(ByteBuf buffer) {
    byte firstByte = buffer.getByte(0);
    if (firstByte == 0xAA) {  // 自定义协议魔数
        return ProtocolType.MY_CUSTOM;
    }
    // ... 其他协议判断
}

// 设备订阅主题
Subscription sub = new Subscription(deviceId, "sensors/+/temperature", QoS.AT_LEAST_ONCE);
subscriptionManager.subscribe(sub);

// 发布消息时匹配订阅者
List<Subscription> matchedSubs = topicMatcher.match("sensors/room1/temperature");
// 返回所有匹配的订阅关系

// 上行消息分发（设备 → 服务端）
upstreamDispatcher.dispatch(internalMessage);
// 1. 匹配主题订阅者
// 2. 通过事件总线推送到Core层
// 3. 在线设备直接推送

// 下行消息分发（服务端 → 设备）
downstreamDispatcher.dispatch(targetDeviceId, message);
// 1. 查询ConnectionManager获取Channel
// 2. 在线则直接发送
// 3. 离线则触发持久化暂存

// 注册新设备
DeviceInfo device = DeviceInfo.builder()
    .productKey("sensor_product")
    .deviceKey("temp_sensor_001")
    .deviceName("温度传感器001")
    .protocolType(ProtocolType.MQTT)
    .authType(AuthType.JWT)
    .build();

deviceRegistry.register(device);

// 定义规则
Rule rule = Rule.builder()
    .ruleId("alert_high_temp")
    .name("高温告警")
    .condition("temperature > 80")
    .action(ActionType.NOTIFY)
    .actionConfig("{\"target\": \"admin@example.com\", \"method\": \"email\"}")
    .enabled(true)
    .build();

ruleEngine.addRule(rule);

// 规则触发（自动）
// 当设备上报 temperature > 80 时，自动发送邮件通知

// 更新期望状态（云端下发）
shadowManager.updateDesired(deviceId, "{\"targetTemp\": 25}");

// 更新上报状态（设备上报）
shadowManager.updateReported(deviceId, "{\"currentTemp\": 23}");

// 获取差异
ShadowDiff diff = shadowManager.getDiff(deviceId);
// diff = {desired: {targetTemp: 25}, reported: {currentTemp: 23}}

// 设备上线时同步期望状态
shadowSyncService.syncOnConnect(deviceId);

// 钉钉告警
AlertService.sendDingTalk(webhookUrl, alertMessage);

// 邮件告警
AlertService.sendEmail(toAddress, subject, content);

@EnableAsync          // 启用异步支持
@EnableScheduling     // 启用定时任务
@SpringBootApplication(scanBasePackages = "com.jysemel.iot")
public class IotPlatformApplication {
    public static void main(String[] args) {
        SpringApplication.run(IotPlatformApplication.class, args);
        System.out.println("IoT物联网平台启动成功！");
    }
}

@Component
public class CustomTcpProtocolAdapter implements ProtocolAdapter {
    
    @Override
    public ProtocolType supportProtocol() {
        return ProtocolType.TCP_CUSTOM;
    }
    
    @Override
    public InternalMessage decode(ByteBuf buffer) {
        // 读取魔数
        byte magic = buffer.readByte();
        if (magic != 0xAA) {
            throw new ProtocolNotSupportException("Invalid magic number");
        }
        
        // 读取版本号、命令字、数据长度
        byte version = buffer.readByte();
        short command = buffer.readShort();
        int length = buffer.readInt();
        
        // 读取数据
        byte[] data = new byte[length];
        buffer.readBytes(data);
        
        // 读取CRC
        short crc = buffer.readShort();
        
        // 转换为InternalMessage
        return InternalMessage.builder()
            .payload(new String(data, StandardCharsets.UTF_8))
            .timestamp(System.currentTimeMillis())
            .build();
    }
}

// 生成Token
String token = jwtTokenProvider.generateToken(deviceIdentity);

// 验证Token
boolean valid = jwtTokenProvider.validateToken(token);
DeviceIdentity identity = jwtTokenProvider.getIdentityFromToken(token);

// 设置ACL规则
authorizationManager.setAcl(deviceId, Arrays.asList(
    new AclRule("sensors/+/temperature", Permission.SUBSCRIBE),
    new AclRule("devices/" + deviceId + "/commands", Permission.PUBLISH)
));

// 检查权限
boolean allowed = authorizationManager.checkPermission(deviceId, 
    "sensors/room1/temperature", Permission.PUBLISH);

// ProtocolType.java
public enum ProtocolType {
    MQTT, COAP, HTTP, TCP_CUSTOM, MY_NEW_PROTOCOL
}

@Component
public class MyNewProtocolAdapter implements ProtocolAdapter {
    
    @Override
    public ProtocolType supportProtocol() {
        return ProtocolType.MY_NEW_PROTOCOL;
    }
    
    @Override
    public InternalMessage decode(ByteBuf buffer) {
        // 实现解码逻辑
    }
    
    @Override
    public ByteBuf encode(InternalMessage message) {
        // 实现编码逻辑
    }
}

private ProtocolType detectProtocol(ByteBuf buffer) {
    byte firstByte = buffer.getByte(0);
    
    // 添加新协议识别
    if (firstByte == 0xBB) {
        return ProtocolType.MY_NEW_PROTOCOL;
    }
    
    // ... 其他协议判断
}

@Override
protected void initChannel(Channel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    
    // 添加协议识别
    pipeline.addLast("protocolDetector", new ProtocolDetector());
    
    // 根据协议添加对应Handler
    pipeline.addLast("myProtocolHandler", new MyProtocolHandler());
    
    // ... 其他Handler
}

// ActionType.java
public enum ActionType {
    NOTIFY, INVOKE_API, STORE_DATA, MY_CUSTOM_ACTION
}

@Component
public class MyCustomActionExecutor implements ActionExecutor {
    
    @Override
    public boolean support(ActionType type) {
        return type == ActionType.MY_CUSTOM_ACTION;
    }
    
    @Override
    public void execute(Rule rule, InternalMessage message) {
        // 实现自定义动作逻辑
        log.info("执行自定义动作: {}", rule.getRuleId());
        
        // 例如：调用第三方服务
        String apiUrl = rule.getActionConfig().get("url");
        httpClient.post(apiUrl, buildPayload(message));
    }
}

@RestController
@RequestMapping("/api/custom")
public class CustomController {
    
    @Autowired
    private CustomService customService;
    
    @GetMapping("/data")
    public ApiResponse<List<DataVO>> getData(@RequestParam String deviceId) {
        List<DataVO> data = customService.queryData(deviceId);
        return ApiResponse.success(data);
    }
    
    @PostMapping("/command")
    public ApiResponse<Void> sendCommand(@RequestBody CommandDTO command) {
        customService.sendCommand(command);
        return ApiResponse.success();
    }
}

@Service
public class CustomService {
    
    @Autowired
    private DownstreamDispatcher downstreamDispatcher;
    
    public void sendCommand(CommandDTO command) {
        InternalMessage message = InternalMessage.builder()
            .deviceId(new DeviceIdentity(command.getProductKey(), command.getDeviceKey()))
            .topic("devices/" + command.getDeviceKey() + "/commands")
            .payload(JsonUtil.toJson(command.getData()))
            .build();
        
        downstreamDispatcher.dispatch(message.getDeviceId(), message);
    }
}

@SpringBootTest
class MqttProtocolAdapterTest {
    
    @Autowired
    private MqttProtocolAdapter adapter;
    
    @Test
    void testDecodeConnectPacket() {
        ByteBuf buffer = Unpooled.buffer();
        // 构造MQTT CONNECT报文
        buffer.writeBytes(mqttConnectBytes);
        
        InternalMessage message = adapter.decode(buffer);
        
        assertNotNull(message);
        assertEquals(ProtocolType.MQTT, message.getProtocolType());
    }
    
    @Test
    void testEncodePublishPacket() {
        InternalMessage message = InternalMessage.builder()
            .topic("test/topic")
            .payload("Hello")
            .build();
        
        ByteBuf buffer = adapter.encode(message);
        
        assertNotNull(buffer);
        assertTrue(buffer.readableBytes() > 0);
    }
}

@SpringBootTest
class TrieTopicMatcherTest {
    
    @Autowired
    private TopicMatcher topicMatcher;
    
    @Test
    void testSingleWildcard() {
        topicMatcher.subscribe("sensors/+/temperature", "device1");
        
        List<String> matched = topicMatcher.match("sensors/room1/temperature");
        
        assertTrue(matched.contains("device1"));
    }
    
    @Test
    void testMultiWildcard() {
        topicMatcher.subscribe("sensors/#", "device1");
        
        List<String> matched = topicMatcher.match("sensors/room1/floor1/temperature");
        
        assertTrue(matched.contains("device1"));
    }
}

// 检查设备连接状态
boolean online = connectionManager.isOnline(deviceId);
log.info("Device {} online: {}", deviceId, online);

// 检查订阅关系
List<Subscription> subs = subscriptionManager.getSubscriptionsByTopic(topic);
log.info("Subscriptions for {}: {}", topic, subs.size());

// 检查Channel状态
Channel channel = connectionManager.getChannel(deviceId);
if (channel != null) {
    log.info("Channel active: {}", channel.isActive());
}

// 启用规则引擎调试日志
logging:
  level:
    com.jysemel.iot.core.rule: debug

// 手动触发规则测试
ruleEngine.testRule(ruleId, testMessage);