Java 类io.netty.handler.codec.mqtt.MqttFixedHeader 实例源码

public static MqttConnectMessage connect(ConnectOptions options) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
            10);
    MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(options.version().protocolName(),
            options.version().protocolLevel(), options.userName() != null, options.password() != null,
            options.will() == null ? false : options.will().isRetain(),
            options.will() == null ? 0 : options.will().qos().value(), options.will() != null,
            options.cleanSession(), options.keepAliveTimeSeconds());

    MqttConnectPayload payload = new MqttConnectPayload(Strings.nullToEmpty(options.clientId()),
            options.will() == null ? "" : options.will().topicName(),
            options.will() == null ? "" : new String(options.will().message(), CharsetUtil.UTF_8),
            Strings.nullToEmpty(options.userName()), Strings.nullToEmpty(options.password()));

    return new MqttConnectMessage(fixedHeader, variableHeader, payload);
}

private MqttConnAckMessage executeNormalChannelRead0(String clientId, boolean cleanSession, ChannelId channelId)
        throws Exception {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
            10);
    MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader("MQTT", 4, true, true, true, 0, true,
            cleanSession, 60);
    MqttConnectPayload payload = new MqttConnectPayload(clientId, "willtopic", "willmessage", "username",
            "password");

    MqttConnectMessage msg = new MqttConnectMessage(fixedHeader, variableHeader, payload);

    ChannelId cid = channelId == null ? TestUtil.newChannelId(clientId, false) : channelId;

    EmbeddedChannel channel = new EmbeddedChannel(cid, new ConnectReceiver());

    channel.writeInbound(msg);

    return channel.readOutbound();
}

private void sendPubAck(String clientId, int messageID) {
    LOG.trace("sendPubAck invoked");
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE, false, 0);
    MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(fixedHeader, from(messageID));

    try {
        if (connectionDescriptors == null) {
            throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should " +
                "be initialized, somewhere it's overwritten!!");
        }
        LOG.trace("connected clientIDs are {}", connectionDescriptors.getConnectedClientIds());
        if (!connectionDescriptors.isConnected(clientId)) {
            throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s",
                clientId, connectionDescriptors));
        }
        connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId);
    } catch (Throwable t) {
        LOG.error(null, t);
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
  log.info("[{}] Processing msg: {}", sessionId, msg);
  if (msg instanceof MqttMessage) {
    MqttMessage mqttMessage = (MqttMessage) msg;
    MqttFixedHeader fixedHeader = mqttMessage.fixedHeader();
    if (fixedHeader != null) {
      processMqttMsg(ctx, (MqttMessage) msg);
    } else {
      //xtx
    }
  }

}

private void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {
  // deviceSessionCtx.setChannel(ctx);
  // assetSessionCtx.setChannel(ctx);

  switch (msg.fixedHeader().messageType()) {
  case CONNECT:
    processConnect(ctx, (MqttConnectMessage) msg);
    break;
  case PUBLISH:
    processPublish(ctx, (MqttPublishMessage) msg);
    // System.out.println("write...");
    // ctx.write("just for test");
    break;
  case SUBSCRIBE:
    processSubscribe(ctx, (MqttSubscribeMessage) msg);
    break;
  case UNSUBSCRIBE:
    processUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
    break;
  case PINGREQ:
    if (checkConnected(ctx)) {
      ctx.writeAndFlush(new MqttMessage(new MqttFixedHeader(PINGRESP, false, AT_MOST_ONCE, false, 0)));
    }
    break;
  case DISCONNECT:
    if (checkConnected(ctx)) {
      processDisconnect(ctx);
    }
    break;
  }
}

public MqttMessage doMessage(Channel channel, MqttMessage msg) {
    String channelId = channel.id().asLongText();
    logger.debug("MQTT PINGREQ " + channelId);

    // 更新最新连接时间
    ApplicationContext.updateChannelConTime(channelId);

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

    return message;
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
  log.info("[{}] Processing msg: {}", sessionId, msg);
  if (msg instanceof MqttMessage) {
    MqttMessage mqttMessage = (MqttMessage) msg;
    MqttFixedHeader fixedHeader = mqttMessage.fixedHeader();
    if (fixedHeader != null) {
      processMqttMsg(ctx, (MqttMessage) msg);
    } else {
      //xtx
    }
  }

}

private void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {
  // deviceSessionCtx.setChannel(ctx);
  // assetSessionCtx.setChannel(ctx);

  switch (msg.fixedHeader().messageType()) {
  case CONNECT:
    processConnect(ctx, (MqttConnectMessage) msg);
    break;
  case PUBLISH:
    processPublish(ctx, (MqttPublishMessage) msg);
    // System.out.println("write...");
    // ctx.write("just for test");
    break;
  case SUBSCRIBE:
    processSubscribe(ctx, (MqttSubscribeMessage) msg);
    break;
  case UNSUBSCRIBE:
    processUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
    break;
  case PINGREQ:
    if (checkConnected(ctx)) {
      ctx.writeAndFlush(new MqttMessage(new MqttFixedHeader(PINGRESP, false, AT_MOST_ONCE, false, 0)));
    }
    break;
  case DISCONNECT:
    if (checkConnected(ctx)) {
      processDisconnect(ctx);
    }
    break;
  }
}

@Override
public void onMsg(SessionCtrlMsg msg) throws SessionException {
  if (msg instanceof SessionCloseMsg) {
    pushToNetwork(
        new MqttMessage(new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0)));
    channel.close();
  }
}

private MqttPublishMessage createMqttPublishMsg(DeviceSessionCtx ctx, String topic, JsonElement json) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false,
      0);
  MqttPublishVariableHeader header = new MqttPublishVariableHeader(topic, ctx.nextMsgId());
  ByteBuf payload = ALLOCATOR.buffer();
  payload.writeBytes(GSON.toJson(json).getBytes(UTF8));
  return new MqttPublishMessage(mqttFixedHeader, header, payload);
}

public static MqttConnAckMessage connack(MqttConnectReturnCode returnCode, boolean sessionPresent) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false,
            2);
    MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);

    return new MqttConnAckMessage(fixedHeader, variableHeader);
}

public static MqttPubAckMessage puback(int messageId) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false,
            2);
    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);

    return new MqttPubAckMessage(fixedHeader, variableHeader);
}

public static MqttMessage pubrec(int messageId) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false,
            2);
    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);

    return new MqttMessage(fixedHeader, variableHeader);
}

public static MqttMessage pubrel(int messageId) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false,
            2);
    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);

    return new MqttMessage(fixedHeader, variableHeader);
}

public static MqttMessage pubcomp(int messageId) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false,
            2);
    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);

    return new MqttMessage(fixedHeader, variableHeader);
}

public static MqttSubscribeMessage subscribe(int messageId, MqttTopicSubscription... topicSubscriptions) {
    int topicNameSize = 0;
    int topicCount = topicSubscriptions.length;

    for (MqttTopicSubscription item : topicSubscriptions) {
        topicNameSize += item.topicName().getBytes(CharsetUtil.UTF_8).length;
    }

    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE,
            false, 2 + topicNameSize + topicCount);
    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
    MqttSubscribePayload payload = new MqttSubscribePayload(Lists.newArrayList(topicSubscriptions));

    return new MqttSubscribeMessage(fixedHeader, variableHeader, payload);
}

public static MqttSubAckMessage suback(int messageId, List<Integer> grantedQoSLevels) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false,
            2 + grantedQoSLevels.size());
    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
    MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);

    return new MqttSubAckMessage(fixedHeader, variableHeader, payload);
}

public static MqttUnsubAckMessage unsuback(int messageId) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false,
            2);
    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);

    return new MqttUnsubAckMessage(fixedHeader, variableHeader);
}

/**
 * Sends PUBREC packet to server
 *
 * @param publishMessage a PUBLISH message to acknowledge
 */
void publishReceived(MqttPublishMessage publishMessage) {

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

  MqttMessageIdVariableHeader variableHeader =
    MqttMessageIdVariableHeader.from(publishMessage.messageId());

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

  qos2inbound.put(publishMessage.messageId(), publishMessage);
  this.write(pubrec);
}

/**
 * Sends the PUBREL message to server
 *
 * @param publishMessageId  identifier of the PUBLISH message to acknowledge
 */
void publishRelease(int publishMessageId) {

  MqttFixedHeader fixedHeader =
    new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);

  MqttMessageIdVariableHeader variableHeader =
    MqttMessageIdVariableHeader.from(publishMessageId);

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

  qos2outbound.put(publishMessageId, pubrel);
  this.write(pubrel);
}

private MqttMessage createConnectPacket(MqttClientOptions options) {
  MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT,
    false,
    MqttQoS.AT_MOST_ONCE,
    false,
    0);

  MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(
    PROTOCOL_NAME,
    PROTOCOL_VERSION,
    options.hasUsername(),
    options.hasPassword(),
    options.isWillRetain(),
    options.getWillQoS(),
    options.isWillFlag(),
    options.isCleanSession(),
    options.getKeepAliveTimeSeconds()
  );

  MqttConnectPayload payload = new MqttConnectPayload(
    options.getClientId() == null ? "" : options.getClientId(),
    options.getWillTopic(),
    options.getWillMessage() != null ? options.getWillMessage().getBytes(StandardCharsets.UTF_8) : null,
    options.hasUsername() ? options.getUsername() : null,
    options.hasPassword() ? options.getPassword().getBytes(StandardCharsets.UTF_8) : null
  );

  return MqttMessageFactory.newMessage(fixedHeader, variableHeader, payload);
}

private boolean checkMessageProperties(MqttMessage message, Map<String, Object> expectedProperties) {
   System.out.println("Checking properties in interceptor");
   try {
      assertNotNull(message);
      assertNotNull(server.getNodeID());
      MqttFixedHeader header = message.fixedHeader();
      assertNotNull(header.messageType());
      assertEquals(header.qosLevel().value(), AT_MOST_ONCE);
      assertEquals(header.isRetain(), expectedProperties.get(RETAINED));
   } catch (Throwable t) {
      collector.addError(t);
   }
   return true;
}

void sendPublishProtocolControlMessage(int messageId, MqttMessageType messageType) {
   MqttQoS qos = (messageType == MqttMessageType.PUBREL) ? MqttQoS.AT_LEAST_ONCE : MqttQoS.AT_MOST_ONCE;
   MqttFixedHeader fixedHeader = new MqttFixedHeader(messageType, false, qos, // Spec requires 01 in header for rel
                                                     false, 0);
   MqttPubAckMessage rel = new MqttPubAckMessage(fixedHeader, MqttMessageIdVariableHeader.from(messageId));
   sendToClient(rel);
}

void handleSubscribe(MqttSubscribeMessage message) throws Exception {
   MQTTSubscriptionManager subscriptionManager = session.getSubscriptionManager();
   int[] qos = subscriptionManager.addSubscriptions(message.payload().topicSubscriptions());

   MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
   MqttSubAckMessage ack = new MqttSubAckMessage(header, message.variableHeader(), new MqttSubAckPayload(qos));
   sendToClient(ack);
}

protected void send(int messageId, String topicName, int qosLevel, boolean isRetain, ByteBuf payload, int deliveryCount) {
   boolean redelivery = qosLevel == 0 ? false : (deliveryCount > 0);
   MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.PUBLISH, redelivery, MqttQoS.valueOf(qosLevel), isRetain, 0);
   MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topicName, messageId);
   MqttMessage publish = new MqttPublishMessage(header, varHeader, payload);
   sendToClient(publish);
}

private MqttMessage createUnSubAckMessage(int msgId) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(UNSUBACK, false, AT_LEAST_ONCE, false, 0);
  MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader.from(msgId);
  return new MqttMessage(mqttFixedHeader, mqttMessageIdVariableHeader);
}

private MqttConnAckMessage createMqttConnAckMsg(MqttConnectReturnCode returnCode) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(CONNACK, false, AT_MOST_ONCE, false, 0);
  MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, true);
  return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}

private static MqttSubAckMessage createSubAckMessage(Integer msgId, List<Integer> grantedQoSList) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(SUBACK, false, AT_LEAST_ONCE, false, 0);
  MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader.from(msgId);
  MqttSubAckPayload mqttSubAckPayload = new MqttSubAckPayload(grantedQoSList);
  return new MqttSubAckMessage(mqttFixedHeader, mqttMessageIdVariableHeader, mqttSubAckPayload);
}

public static MqttPubAckMessage createMqttPubAckMsg(int requestId) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(PUBACK, false, AT_LEAST_ONCE, false, 0);
  MqttMessageIdVariableHeader mqttMsgIdVariableHeader = MqttMessageIdVariableHeader.from(requestId);
  return new MqttPubAckMessage(mqttFixedHeader, mqttMsgIdVariableHeader);
}

private MqttMessage createUnSubAckMessage(int msgId) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(UNSUBACK, false, AT_LEAST_ONCE, false, 0);
  MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader.from(msgId);
  return new MqttMessage(mqttFixedHeader, mqttMessageIdVariableHeader);
}

private MqttConnAckMessage createMqttConnAckMsg(MqttConnectReturnCode returnCode) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(CONNACK, false, AT_MOST_ONCE, false, 0);
  MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, true);
  return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}

private static MqttSubAckMessage createSubAckMessage(Integer msgId, List<Integer> grantedQoSList) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(SUBACK, false, AT_LEAST_ONCE, false, 0);
  MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader.from(msgId);
  MqttSubAckPayload mqttSubAckPayload = new MqttSubAckPayload(grantedQoSList);
  return new MqttSubAckMessage(mqttFixedHeader, mqttMessageIdVariableHeader, mqttSubAckPayload);
}

public static MqttPubAckMessage createMqttPubAckMsg(int requestId) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(PUBACK, false, AT_LEAST_ONCE, false, 0);
  MqttMessageIdVariableHeader mqttMsgIdVariableHeader = MqttMessageIdVariableHeader.from(requestId);
  return new MqttPubAckMessage(mqttFixedHeader, mqttMsgIdVariableHeader);
}

private static MqttPubAckMessage createMqttPubAckMsg(int requestId) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(PUBACK, false, AT_LEAST_ONCE, false, 0);
  MqttMessageIdVariableHeader mqttMsgIdVariableHeader = MqttMessageIdVariableHeader.from(requestId);
  return new MqttPubAckMessage(mqttFixedHeader, mqttMsgIdVariableHeader);
}

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

    return new MqttMessage(fixedHeader);
}

public static MqttMessage disconnect() {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE,
            false, 2);

    return new MqttMessage(fixedHeader);
}

public MqttEndpointImpl subscribeAcknowledge(int subscribeMessageId, List<MqttQoS> grantedQoSLevels) {

    this.checkConnected();

    MqttFixedHeader fixedHeader =
      new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
    MqttMessageIdVariableHeader variableHeader =
      MqttMessageIdVariableHeader.from(subscribeMessageId);

    MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels.stream().mapToInt(MqttQoS::value).toArray());

    io.netty.handler.codec.mqtt.MqttMessage suback = MqttMessageFactory.newMessage(fixedHeader, variableHeader, payload);

    this.write(suback);

    return this;
  }

private static MqttConnAckMessage createConnAckMessage(MqttConnectReturnCode code) {
    MqttFixedHeader mqttFixedHeader =
            new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
    MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(code, true);
    return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}

void sendConnack(MqttConnectReturnCode returnCode) {
   MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
   MqttConnAckVariableHeader varHeader = new MqttConnAckVariableHeader(returnCode, true);
   MqttConnAckMessage message = new MqttConnAckMessage(fixedHeader, varHeader);
   sendToClient(message);
}

void handleUnsubscribe(MqttUnsubscribeMessage message) throws Exception {
   session.getSubscriptionManager().removeSubscriptions(message.payload().topics());
   MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
   MqttUnsubAckMessage m = new MqttUnsubAckMessage(header, message.variableHeader());
   sendToClient(m);
}