查看本类的 API文档回源码主页

   /*
    * Copyright 2019 The Netty Project
    *
    * The Netty Project licenses this file to you under the Apache License,
    * version 2.0 (the "License"); you may not use this file except in compliance
    * with the License. You may obtain a copy of the License at:
    *
    *   https://www.apache.org/licenses/LICENSE-2.0
    *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   * License for the specific language governing permissions and limitations
   * under the License.
   */
  package io.netty.example.mqtt.heartBeat;
  
  import io.netty.channel.ChannelHandlerContext;
  import io.netty.channel.ChannelInboundHandlerAdapter;
  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.MqttMessage;
  import io.netty.handler.codec.mqtt.MqttMessageType;
  import io.netty.handler.codec.mqtt.MqttQoS;
  import io.netty.handler.codec.mqtt.MqttProperties;
  import io.netty.handler.timeout.IdleStateEvent;
  import io.netty.util.ReferenceCountUtil;
  
  public class MqttHeartBeatClientHandler extends ChannelInboundHandlerAdapter {
  
      private static final String PROTOCOL_NAME_MQTT_3_1_1 = "MQTT";
      private static final int PROTOCOL_VERSION_MQTT_3_1_1 = 4;
  
      private final String clientId;
      private final String userName;
      private final byte[] password;
  
      public MqttHeartBeatClientHandler(String clientId, String userName, String password) {
          this.clientId = clientId;
          this.userName = userName;
          this.password = password.getBytes();
      }
  
      @Override
      public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
          // discard all messages
          ReferenceCountUtil.release(msg);
      }
  
      @Override
      public void channelActive(ChannelHandlerContext ctx) throws Exception {
          MqttFixedHeader connectFixedHeader =
                  new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
          MqttConnectVariableHeader connectVariableHeader =
                  new MqttConnectVariableHeader(PROTOCOL_NAME_MQTT_3_1_1, PROTOCOL_VERSION_MQTT_3_1_1, true, true, false,
                                                0, false, false, 20, MqttProperties.NO_PROPERTIES);
          MqttConnectPayload connectPayload = new MqttConnectPayload(clientId,
                  MqttProperties.NO_PROPERTIES,
                  null,
                  null,
                  userName,
                  password);
          MqttConnectMessage connectMessage =
                  new MqttConnectMessage(connectFixedHeader, connectVariableHeader, connectPayload);
          ctx.writeAndFlush(connectMessage);
          System.out.println("Sent CONNECT");
      }
  
      @Override
      public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
          if (evt instanceof IdleStateEvent) {
              MqttFixedHeader pingreqFixedHeader =
                      new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0);
              MqttMessage pingreqMessage = new MqttMessage(pingreqFixedHeader);
              ctx.writeAndFlush(pingreqMessage);
              System.out.println("Sent PINGREQ");
          } else {
              super.userEventTriggered(ctx, evt);
          }
      }
  
      @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
          cause.printStackTrace();
          ctx.close();
      }
  }