基于netty, spring boot等框架实现的物联网中间件, 支持常用物联网协议支持快速接入数据库或消息队列

package com.iteaj.iot.client.mqtt; import cn.hutool.core.collection.CollectionUtil; import com.iteaj.iot.*; import com.iteaj.iot.client.*; import com.iteaj.iot.client.mqtt.message.MqttClientMessage; import com.iteaj.iot.client.protocol.ClientSocketProtocol; import com.iteaj.iot.client.protocol.ServerInitiativeProtocol; import com.iteaj.iot.config.ConnectProperties; import com.iteaj.iot.event.ClientStatus; import com.iteaj.iot.event.StatusEvent; import com.iteaj.iot.utils.ByteUtil; import io.netty.buffer.Unpooled; import io.netty.channel.*; import io.netty.handler.codec.DecoderResult; import io.netty.handler.codec.mqtt.*; import java.io.IOException; import java.net.SocketAddress; import java.nio.channels.ClosedChannelException; import java.util.List; import java.util.function.Consumer; import static com.iteaj.iot.CoreConst.CLIENT_DECODER_HANDLER; /** * 基于mqtt协议的客户端 * @see MqttConnectProperties#getAllIdleTime() 作为keepAlive字段 单位毫秒 * 详情资料{@code https://blog.csdn.net/weixin_40973138/article/details/90036953} * @see MqttEncoder * @see MqttDecoder */ public class MqttClient extends TcpSocketClient implements ChannelInboundHandler, ChannelOutboundHandler, MultiStageConnect { /** * 用来标识mqtt客户端是否已经得到服务端的connAck */ private ChannelPromise connAckFinished; private MessageIdManager messageIdManager; public MqttClient(MqttClientComponent clientComponent) { this(clientComponent, clientComponent.getConfig()); } public MqttClient(MqttClientComponent clientComponent, MqttConnectProperties config) { super(clientComponent, config); this.messageIdManager = new SimpleMessageIdManager(this, clientComponent); } public MqttClient(MqttClientComponent clientComponent, MqttConnectProperties config, MessageIdManager messageIdManager) { super(clientComponent, config); this.messageIdManager = messageIdManager; } @Override protected ChannelInboundHandler createProtocolDecoder() { // https://gitee.com/iteaj/iot/issues/I5NLVV return new MqttDecoder(getConfig().getMaxBytesInMessage()); } @Override protected ChannelOutboundHandlerAdapter createProtocolEncoder() { return MqttEncoder.INSTANCE; } /** * 新增MqttMessage到MqttClientMessage的处理器 包括心跳处理器和解码处理器 * @param channel */ @Override protected void doInitChannel(Channel channel) { super.doInitChannel(channel); channel.pipeline().addAfter(CLIENT_DECODER_HANDLER, "MqttMessageToClientMessageHandler", this); } /** * 连接成功后发起 connect 请求 * @param future */ @Override protected void successCallback(ChannelFuture future) { boolean willFlag = getConfig().getWillMessage() != null && getConfig().getWillTopic() != null; MqttConnectMessage connectMessage = MqttMessageBuilders.connect() .willFlag(willFlag) .willQoS(getConfig().getWillQos()) .clientId(getConfig().getClientId()) .username(getConfig().getUsername()) .willTopic(getConfig().getWillTopic()) .keepAlive(getConfig().getKeepAlive()) .willRetain(getConfig().isWillRetain()) .protocolVersion(getConfig().getVersion()) .cleanSession(getConfig().isCleanSession()) .password(getConfig().getPasswordByCharset()) .willMessage(getConfig().getWillMessageByCharset()) .build(); this.getChannel().writeAndFlush(connectMessage).addListener(future1 -> { if(future1.isSuccess()) { if(logger.isTraceEnabled()) { logger.trace("mqtt({}) {} - 远程主机：{} - 状态：成功 - 报文：{}" , getName(), MqttMessageType.CONNECT, remoteKey(), connectMessage); } } else { logger.warn("mqtt({}) {} - 远程主机：{} - 状态：失败 - 报文：{}" , getName(), MqttMessageType.CONNECT, remoteKey(), connectMessage, future1.cause()); } }); } @Override public ChannelFuture close() { MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.DISCONNECT ,false, MqttQoS.AT_MOST_ONCE, false, 0); MqttMessage disconnectMqttMessage = new MqttMessage(header); // 发送断开连接报文 return this.getChannel().writeAndFlush(disconnectMqttMessage).addListener(future -> { if(future.isSuccess()) { getChannel().close().addListener(disFuture -> { if(disFuture.isSuccess()) { disconnectSuccessCall(); if(logger.isTraceEnabled()) { logger.trace("mqtt({}) 关闭客户端({}) - 远程主机：{} - 状态：成功" , getName(), MqttMessageType.DISCONNECT, remoteKey()); } } else { logger.warn("mqtt({}) 关闭客户端({}) - 远程主机：{} - 状态：失败" , getName(), MqttMessageType.DISCONNECT, remoteKey(), disFuture.cause()); } }); if(logger.isTraceEnabled()) { logger.trace("mqtt({}) {} - 远程主机：{} - 状态：成功" , getName(), MqttMessageType.DISCONNECT, remoteKey()); } } else if(future.cause() instanceof ClosedChannelException) { // 服务端已经关闭此连接, 也属于完成 disconnectSuccessCall(); if(logger.isTraceEnabled()) { logger.trace("mqtt({}) 关闭客户端({}) - 远程主机：{} - 状态：成功" , getName(), MqttMessageType.DISCONNECT, remoteKey()); } } else { logger.warn("mqtt({}) {} - 远程主机：{} - 状态：失败" , getName(), MqttMessageType.DISCONNECT, remoteKey(), future.cause()); } }); } @Override public ChannelFuture writeAndFlush(ClientSocketProtocol clientProtocol) { ClientMessage clientMessage = clientProtocol.requestMessage(); if(!(clientMessage instanceof MqttClientMessage)) { throw new ProtocolException("Mqtt报文必须使用类型[MqttClientMessage]"); } return super.writeAndFlush(clientProtocol); } private Object debugMessage(ClientMessage message, MqttPublishMessage mqttMessage) { final byte[] binMsg = message.getMessage(); final MqttFixedHeader fixedHeader = mqttMessage.fixedHeader(); final MqttPublishVariableHeader variableHeader = mqttMessage.variableHeader(); return new StringBuilder("MqttPublishMessage") .append('[') .append(fixedHeader != null ? fixedHeader.toString() : "").append(',') .append(variableHeader != null ? variableHeader.toString() : "") .append(", payload=[").append(binMsg != null ? ByteUtil.bytesToHexByFormat(binMsg) : "") .append("]]") .toString(); } /** * 构建要发布的报文 * @param message * @return * @throws IOException */ public MqttPublishMessage buildPublishMqttMessage(MqttClientMessage message) { final int nextId = getMessageIdManager().nextId(); return this.buildPublishMqttMessage(message, nextId); } protected MqttPublishMessage buildPublishMqttMessage(MqttClientMessage message, Integer packetId) { if(message.getMessage() == null) { message.writeBuild(); } return MqttMessageBuilders.publish() .messageId(pack