io.netty.handler.codec.mqtt.MqttMessageType # PINGREQ 源码实例Demo

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (!(msg instanceof MqttMessage)) {
        System.out.println("=============收到非Mqtt===========");
        ctx.fireChannelRead(msg);
        ctx.close();
        return;
    }
    MqttMessage message = (MqttMessage) msg;
    if(message.fixedHeader().messageType() == MqttMessageType.PINGREQ){
        System.out.println("===========服务端收到了ping请求==========");
        this.handlePingReq(ctx.channel());
    } else if(message.fixedHeader().messageType() == MqttMessageType.PINGRESP){
        System.out.println("===========服务端收到了ping响应==========");
        this.handlePingResp();
    }else{
        ctx.fireChannelRead(ReferenceCountUtil.retain(msg));
    }
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (!(msg instanceof MqttMessage)) {
        ctx.fireChannelRead(msg);
        return;
    }
    MqttMessage message = (MqttMessage) msg;
    if(message.fixedHeader().messageType() == MqttMessageType.PINGREQ){
        this.handlePingReq(ctx.channel());
    } else if(message.fixedHeader().messageType() == MqttMessageType.PINGRESP){
        this.handlePingResp();
    }else{
        ctx.fireChannelRead(ReferenceCountUtil.retain(msg));
    }
}
 

private void sendPingReq(Channel channel){
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0);
    channel.writeAndFlush(new MqttMessage(fixedHeader));

    if(this.pingRespTimeout != null){
        this.pingRespTimeout = channel.eventLoop().schedule(() -> {
            MqttFixedHeader fixedHeader2 = new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
            channel.writeAndFlush(new MqttMessage(fixedHeader2)).addListener(ChannelFutureListener.CLOSE);
            //TODO: what do when the connection is closed ?
        }, this.keepaliveSeconds, TimeUnit.SECONDS);
    }
}
 

/**
 * 5秒钟收不到来自对方的ping，就关闭连接
 * @param channel
 */
private void sendPingReq(Channel channel){
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0);
    channel.writeAndFlush(new MqttMessage(fixedHeader));

   /* if(this.pingRespTimeout == null){
        this.pingRespTimeout = channel.eventLoop().schedule(() -> {
            MqttFixedHeader fixedHeader2 = new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
            channel.writeAndFlush(new MqttMessage(fixedHeader2)).addListener(ChannelFutureListener.CLOSE);
            //TODO: what do when the connection is closed ?
        }, this.keepaliveSeconds, TimeUnit.SECONDS);
    }*/
}
 

@Override
public MqttMessageType type() {
    return MqttMessageType.PINGREQ;
}
 

/**
 * See {@link MqttClient#ping()} for more details
 */
@Override
public MqttClient ping() {

  MqttFixedHeader fixedHeader =
    new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0);

  io.netty.handler.codec.mqtt.MqttMessage pingreq = MqttMessageFactory.newMessage(fixedHeader, null, null);

  this.write(pingreq);

  return this;
}