java实现一个Mqtt broker02处理MqttConn请求 _java

这一节，我们来实现MqttConn请求的响应，返回ConnAck消息
我们先创建一个MqttHandler，让他继承
ChannelInboundHandlerAdapter, 用来接力MqttDecoder解码完成后的消息,这里要继承其中的channelRead方法
package com.github.shoothzj.mqtt;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import lombok.extern.slf4j.Slf4j;@Slf4jpublic class MqttHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {super.channelRead(ctx, msg);}}然后把这个handler加入到netty的职责链中，放到解码器的后面

文章插图

在mqtt handler中插入我们的代码

@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {super.channelRead(ctx, msg);if (msg instanceof MqttConnectMessage) {handleConnect(ctx, (MqttConnectMessage) msg);} else {log.error("Unsupported type msg [{}]", msg);}}private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage connectMessage) {log.info("connect msg is [{}]", connectMessage);}

打印出connectMessage如下

[MqttConnectMessage[fixedHeader=MqttFixedHeader[messageType=CONNECT, isDup=false, qosLevel=AT_MOST_ONCE, isRetain=false, remainingLength=22], variableHeader=MqttConnectVariableHeader[name=MQTT, version=4, hasUserName=false, hasPassword=false, isWillRetain=false, isWillFlag=false, isCleanSession=true, keepAliveTimeSeconds=60], payload=MqttConnectPayload[clientIdentifier=JAVASample, willTopic=null, willMessage=null, userName=null, password=null]]]

通常，mqtt connect message中会包含qos、用户名、密码等信息，由于我们启动客户端的时候也没有携带用户名和密码，这里获取到的都为null，我们先不校验这些消息，直接给客户端返回connack消息，代表连接成功

final MqttConnAckMessage ackMessage = MqttMessageBuilders.connAck().returnCode(CONNECTION_ACCEPTED).build();ctx.channel().writeAndFlush(ackMessage);

我们再运行起Server和Client，随后可以看到已经走过了Connect阶段，进入了publish message过程，接下来我们再实现更多的其他场景

文章插图

附上此阶段的MqttHandler代码

package com.github.shoothzj.mqtt;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.handler.codec.mqtt.MqttConnAckMessage;import io.netty.handler.codec.mqtt.MqttConnectMessage;import io.netty.handler.codec.mqtt.MqttConnectPayload;import io.netty.handler.codec.mqtt.MqttConnectVariableHeader;import io.netty.handler.codec.mqtt.MqttFixedHeader;import io.netty.handler.codec.mqtt.MqttMessageBuilders;import lombok.extern.slf4j.Slf4j;import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;@Slf4jpublic class MqttHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {super.channelRead(ctx, msg);if (msg instanceof MqttConnectMessage) {handleConnect(ctx, (MqttConnectMessage) msg);} else {log.error("Unsupported type msg [{}]", msg);}}private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage connectMessage) {log.info("connect msg is [{}]", connectMessage);final MqttFixedHeader fixedHeader = connectMessage.fixedHeader();final MqttConnectVariableHeader variableHeader = connectMessage.variableHeader();final MqttConnectPayload connectPayload = connectMessage.payload();final MqttConnAckMessage ackMessage = MqttMessageBuilders.connAck().returnCode(CONNECTION_ACCEPTED).build();ctx.channel().writeAndFlush(ackMessage);}}

【java实现一个Mqtt broker02处理MqttConn请求】