io.netty.handler.codec.mqtt.MqttMessageType # PINGRESP 源码实例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));
    }
}
 

@Override
public void handleRequest(Channel client, MqttMessage message) throws Exception {
    String clientId = NettyAttrManager.getAttrClientId(client);
    if (logger.isDebugEnabled()) {
        logger.debug(String.format("PingRequest clientId:%s", clientId));
    }
    MqttFixedHeader pingHeader = new MqttFixedHeader(
            MqttMessageType.PINGRESP,
            false,
            AT_MOST_ONCE,
            false,
            0);
    MqttMessage pingResp = new MqttMessage(pingHeader);
    client.writeAndFlush(pingResp);
}
 

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

private void handlePingReq(Channel channel){
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
    channel.writeAndFlush(new MqttMessage(fixedHeader));
    System.out.println("================服务端发送ping给客户端============");
    System.out.println(fixedHeader.toString());
}
 

public static MqttMessage getPingRespMessage(){
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP,false,MqttQoS.AT_MOST_ONCE,false,0);
    MqttMessage mqttMessage = new MqttMessage(fixedHeader);
    return mqttMessage;
}
 

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

public static MqttMessage pingresp() {
	MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false,
			0);

	return new MqttMessage(fixedHeader);
}
 

void handlePingreq() {
   MqttMessage pingResp = new MqttMessage(new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0));
   sendToClient(pingResp);
}
 

public MqttEndpointImpl pong() {

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

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

    this.write(pingresp);

    return this;
  }