MQTT协议框架MOP代码结构解析

一、MoP是啥

MoP 即MQTT on Pulsar ，是一个在Pulsar基础上实现的MQTT协议，git地址：https://github.com/streamnative/mop
通过MoP可以快速的搭建一个MQTT服务器，下面介绍一下MoP的主要的代码结构。

二、MQTT消息定义

MQTT的消息在MoP中的定义是MqttMesssage，其代码如下：

public class MqttMessage {private final MqttFixedHeader mqttFixedHeader;private final Object variableHeader;private final Object payload;private final DecoderResult decoderResult;public static final MqttMessage PINGREQ;public static final MqttMessage PINGRESP;public static final MqttMessage DISCONNECT;
}

（1）MqttFixedHeader（固定头）：存在于所有MQTT数据包中，表示数据包类型及数据包的分组类标识。
（2）variableHeader（可变头）：存在于部分MQTT数据包中，数据包类型决定了可变头是否存在及其具体内容。
（3）消息体（payload）。存在于部分MQTT数据包中，表示客户端收到的具体内容。

三、MoP主要代码模块

1、MQTTProtocolHandler:代码入口，继承自ProtocolHandler。
2、MQTTProxyService:采用netty的方式，实现网络通信
3、MQTTCommonInboundHandler:继承自ChannelInboundHandlerAdapter，实现ChannelInboundHandlerAdapter中的channelRead方法，获取客户端发送过来的消息，根据消息类型进入不同的处理流程。

public class MQTTCommonInboundHandler extends ChannelInboundHandlerAdapter {protected ProtocolMethodProcessor processor;  //协议处理@Overridepublic void channelRead(ChannelHandlerContext ctx, Object message) {checkArgument(message instanceof MqttMessage);checkNotNull(processor);MqttMessage msg = (MqttMessage) message;try {checkState(msg);MqttMessageType messageType = msg.fixedHeader().messageType();if (log.isDebugEnabled()) {log.debug("Processing MQTT Inbound handler message, type={}", messageType);}switch (messageType) {case CONNECT:checkArgument(msg instanceof MqttConnectMessage);processor.processConnect((MqttConnectMessage) msg);break;case SUBSCRIBE:checkArgument(msg instanceof MqttSubscribeMessage);processor.processSubscribe((MqttSubscribeMessage) msg);break;case UNSUBSCRIBE:checkArgument(msg instanceof MqttUnsubscribeMessage);processor.processUnSubscribe((MqttUnsubscribeMessage) msg);break;case PUBLISH:checkArgument(msg instanceof MqttPublishMessage);processor.processPublish((MqttPublishMessage) msg);break;case PUBREC:processor.processPubRec(msg);break;case PUBCOMP:processor.processPubComp(msg);break;case PUBREL:processor.processPubRel(msg);break;case DISCONNECT:processor.processDisconnect(msg);break;case PUBACK:checkArgument(msg instanceof MqttPubAckMessage);processor.processPubAck((MqttPubAckMessage) msg);break;case PINGREQ:processor.processPingReq();break;default:throw new UnsupportedOperationException("Unknown MessageType: " + messageType);}} catch (Throwable ex) {ReferenceCountUtil.safeRelease(msg);log.error("Exception was caught while processing MQTT message, ", ex);ctx.close();}}

如上，先将消息解析成MqttMessage类型，再从mqttFixedHeader中获取消息类型messageType，然后根据不同类型进行不同的处理流程。

4、MQTTProxyInboundHandler：继承自MQTTCommonInboundHandler，代码如下：

public class MQTTProxyInboundHandler extends MQTTCommonInboundHandler {private final MQTTProxyService proxyService;public MQTTProxyInboundHandler(MQTTProxyService proxyService) {this.proxyService = proxyService;}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {super.channelActive(ctx);this.processor = new MQTTProxyProtocolMethodProcessor(proxyService, ctx);}
}

MQTTProxyInboundHandler通过实现channelActive,初始化processor MQTTProxyProtocolMethodProcessor类型。即处理mqtt消息主要是在MQTTProxyProtocolMethodProcessor中进行。

5、MQTTProxyProtocolMethodProcessor：继承自AbstractCommonProtocolMethodProcessor，主要用来处理MQTT消息。下面以处理CONNECT消息和PUBLISH消息为例来说明一下主要的逻辑。

1）处理CONNECT消息

 @Overridepublic void processConnect(MqttConnectMessage msg) {MqttConnectPayload payload = msg.payload();MqttConnectMessage connectMessage = msg;final int protocolVersion = msg.variableHeader().version();final String username = payload.userName();String clientId = payload.clientIdentifier();if (log.isDebugEnabled()) {log.debug("process CONNECT message. CId={}, username={}", clientId, username);}// Check MQTT protocol version.if (!MqttUtils.isSupportedVersion(protocolVersion)) {log.error("MQTT protocol version is not valid. CId={}", clientId);channel.writeAndFlush(MqttConnectAckHelper.errorBuilder().unsupportedVersion());channel.close();return;}if (!MqttUtils.isQosSupported(msg)) {channel.writeAndFlush(MqttConnectAckHelper.errorBuilder().willQosNotSupport(protocolVersion));channel.close();return;}// Client must specify the client ID except enable clean session on the connection.if (StringUtils.isEmpty(clientId)) {if (!msg.variableHeader().isCleanSession()) {channel.writeAndFlush(MqttConnectAckHelper.errorBuilder().identifierInvalid(protocolVersion));channel.close();log.error("The MQTT client ID cannot be empty. Username={}", username);return;}clientId = MqttMessageUtils.createClientIdentifier(channel);connectMessage = MqttMessageUtils.stuffClientIdToConnectMessage(msg, clientId);if (log.isDebugEnabled()) {log.debug("Client has connected with generated identifier. CId={}", clientId);}}String userRole = null;if (!authenticationEnabled) {if (log.isDebugEnabled()) {log.debug("Authentication is disabled, allowing client. CId={}, username={}", clientId, username);}} else {MQTTAuthenticationService.AuthenticationResult authResult = authenticationService.authenticate(payload);if (authResult.isFailed()) {channel.writeAndFlush(MqttConnectAckHelper.errorBuilder().authFail(protocolVersion));channel.close();log.error("Invalid or incorrect authentication. CId={}, username={}", clientId, username);return;}userRole = authResult.getUserRole();}doProcessConnect(connectMessage, userRole);}

当MoP接收到connect类型消息时，调用processConnect方法，进行一系列的消息校验，然后调用 doProcessConnect(connectMessage, userRole)来创建连接

 public void doProcessConnect(MqttConnectMessage msg, String userRole) {connection = Connection.builder().protocolVersion(msg.variableHeader().version()).clientId(msg.payload().clientIdentifier()).userRole(userRole).cleanSession(msg.variableHeader().isCleanSession()).connectMessage(msg).keepAliveTime(msg.variableHeader().keepAliveTimeSeconds()).channel(channel).connectionManager(connectionManager).serverReceivePubMaximum(proxyConfig.getReceiveMaximum()).build();connection.sendConnAck();}

通过创建好的connection对象来和客户端通信，并发送消息ack通知客户端建立连接成功。

2）处理PUBLISH消息

//处理PUBLISH消息
public void processPublish(MqttPublishMessage msg) {if (log.isDebugEnabled()) {log.debug("[Proxy Publish] publish to topic = {}, CId={}",msg.variableHeader().topicName(), connection.getClientId());}final int packetId = msg.variableHeader().packetId();final String pulsarTopicName = PulsarTopicUtils.getEncodedPulsarTopicName(msg.variableHeader().topicName(),proxyConfig.getDefaultTenant(), proxyConfig.getDefaultNamespace(),TopicDomain.getEnum(proxyConfig.getDefaultTopicDomain()));CompletableFuture<InetSocketAddress> lookupResult = lookupHandler.findBroker(TopicName.get(pulsarTopicName));lookupResult.thenCompose(brokerAddress -> writeToBroker(brokerAddress, pulsarTopicName, msg)).exceptionally(ex -> {msg.release();log.error("[Proxy Publish] Failed to publish for topic : {}, CId : {}",msg.variableHeader().topicName(), connection.getClientId(), ex);MopExceptionHelper.handle(MqttMessageType.PUBLISH, packetId, channel, ex);return null;});}
//写入消息到broker中private CompletableFuture<Void> writeToBroker(InetSocketAddress mqttBroker, String topic, MqttMessage msg) {CompletableFuture<MQTTProxyExchanger> proxyExchanger = connectToBroker(mqttBroker, topic);return proxyExchanger.thenCompose(exchanger -> writeToBroker(exchanger, msg));}

从msg中获取topicName，再根据topicName获取broker地址，再根据broker地址、topicName将消息写入broker

MQTT协议框架MOP代码结构解析相关推荐

Android蓝牙系统框架和代码结构
Android蓝牙系统框架和代码结构概述在 Android 4.2版本中,谷歌公司和博通合作,引入了博通的 BTE/BTA 协议栈,重构了蓝牙子系统.新的蓝牙协议栈被命名为 BlueDroid.它 ...
smali代码结构解析
smali代码结构解析一.开头(类的声明) 1.声明一个类继承自某个类 java代码如下 public class MainActivity extends AppCompatActivity 对应 ...
Android之高仿雅虎天气(二)---代码结构解析
版本已升级至1.0.1 源码地址: GitHub:https://github.com/way1989/WayHoo OsChina:http://git.oschina.net/way/WayHoo ...
xfire框架内部基本结构解析
1 概述 xfire是webservice的一个实现框架,是apache旗下CXF的前身,是一个比较被广泛使用的webservice框架,网上有很多关于如何使用xfire或cxf的hello worl ...
合泰32-Onenet-WiFi模块-合泰单片机通过MQTT协议数据上云（二）
本篇目标一.学习MQTT协议二.发送温度湿度数据到OneNET 准备材料感觉麻烦的小伙伴请移步到文末,那里有整个工程的链接. (关于串口和WiFi部分的内容,可以查看前一篇博文,链接:这个是传送 ...
物联网平台通信协议之 MQTT 协议
物联网平台通信协议之 MQTT 协议文章目录物联网平台通信协议之 MQTT 协议 MQTT 概述 MQTT 数据格式 MQTT 概述 MQTT( Message Queuing Telemetry ...
DDD 实战 (2)：看看代码结构长啥样（值得收藏）
Talking is easy, show me your code 真正开始 DDD 旅程前,我想让您看到经过 DDD 设计之后的代码长啥样.我想,这是所有本着"talking is ea ...
DDD实战(2)：看看代码结构长啥样
Talking is easy, show me your code 真正开始 DDD 旅程前,我想让您看到经过 DDD 设计之后的代码长啥样.我想,这是所有本着"talking is ea ...
MQTT协议详解及开发教程（一）MQTT协议概述
推荐一款稳定的基于C编写的MQTT Client开源库<cMQTT> 一概述 MQTT协议目前在物联网技术中应用非常广泛,各种公有云的IOT平台通信基本上都是按照该协议来实现的,这里先简 ...

MQTT协议框架MOP代码结构解析

MQTT协议框架MOP代码结构解析相关推荐

最新文章

热门文章