io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader # messageId ( ) 源码实例Demo

/**
 *  P - B (Qos2)
 * @param channel
 * @param variableHeader
 */
public void processPubRel(Channel channel, MqttMessageIdVariableHeader variableHeader) {
	int messageId = variableHeader.messageId();
	ProcedureMessage info = this.procedureProcess.processPubRel(channel, messageId);

	if (info != null) {
		NettyLog.debug("relInfo:" + info);

		BorkerMessage bMsgInfo = BorkerMessage.builder().sourceClientId(info.getSourceClientId())
				.sourceMsgId(info.getSourceMsgId()).topicName(info.getTopicName()).iQosLevel(info.getIQosLevel())
				.msgBytes(info.getMsgBytes()).retain(false).build();

		List<SubscribeTopicInfo> subscribeClientList = this.topicProcess.search(bMsgInfo.getTopicName());
		this.consumerProcess.sendSubscribMessage(bMsgInfo, subscribeClientList);
	}

	this.sendProcess.sendPubCompMessage(channel, messageId);
}
 

/**
 * B - P (Qos1)
 * @param channel
 * @param mqttMessage
 */
public void processPubAck(Channel channel, MqttMessage mqttMessage) {
	MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
			.variableHeader();
	int messageId = messageIdVariableHeader.messageId();

	producerProcess.processPubAck(messageId);
	producerProcess.delMessage(messageId);
}
 

/**
 * B- P(Qos2)
 * @param channel
 * @param mqttMessage
 */
public void processPubRec(Channel channel, MqttMessage mqttMessage) {
	MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
			.variableHeader();
	int messageId = messageIdVariableHeader.messageId();

	producerProcess.processPubRec(messageId);
	producerProcess.sendPubRel(messageId);
}
 

/**
 * B - P (Qos2)
 * @param channel
 * @param mqttMessage
 */
public void processPubComp(Channel channel, MqttMessage mqttMessage) {
	MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
			.variableHeader();
	int messageId = messageIdVariableHeader.messageId();

	producerProcess.processPubComp(messageId);
	producerProcess.delMessage(messageId);
}
 

/**
 * B - S(Qos2)
 * @param channel
 * @param mqttMessage
 */
public void processPubRel(Channel channel, MqttMessage mqttMessage) {
	MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
			.variableHeader();
	int messageId = messageIdVariableHeader.messageId();
	this.consumerProcess.processPubRel(messageId);
	this.consumerProcess.sendPubCompMessage(messageId);
	this.consumerProcess.delMesage(messageId);
}
 

/**
 * B - S
 * @param channel
 * @param mqttMessage
 */
public void processUnSubBack(Channel channel, MqttMessage mqttMessage) {
	int messageId;
	if (mqttMessage instanceof MqttUnsubAckMessage) {
		MqttUnsubAckMessage mqttUnsubAckMessage = (MqttUnsubAckMessage) mqttMessage;
		messageId = mqttUnsubAckMessage.variableHeader().messageId();
	} else {
		MqttMessageIdVariableHeader o = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
		messageId = o.messageId();
		NettyLog.error("not UnsubAckMessage:{}", messageId);
	}
	this.consumerProcess.processUnSubBack(messageId);
}
 

public void processPubAck(Channel client, MqttPubAckMessage pubAckMessage) {
    MqttMessageIdVariableHeader pubAckVariableMessage = pubAckMessage.variableHeader();
    short packageId = (short) pubAckVariableMessage.messageId();
    String clientId = NettyAttrManager.getAttrClientId(client);

    consumerManager.acknowledge(clientId, packageId);

    if (LOG.isDebugEnabled()) {
        LOG.debug("Received PubAck packageID: {}" + packageId);
    }
}
 

/**
 *  S - B (Qos2)
 * @param channel
 * @param variableHeader
 */
public void processPubRec(Channel channel, MqttMessageIdVariableHeader variableHeader) {
	int messageId = variableHeader.messageId();
	this.consumerProcess.processPubRec(channel, messageId);
	this.sendProcess.sendPubRelMessage(channel, messageId, false);
}
 

public static int getMessageId(MqttMessage mqttMessage){
    MqttMessageIdVariableHeader idVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
    return idVariableHeader.messageId();
}
 

public void processPubComp(Channel channel, MqttMessageIdVariableHeader variableHeader){
    int messageId = variableHeader.messageId();
    log.info("PUBCOMP - clientId: {}, messageId: {}", (String) channel.attr(AttributeKey.valueOf("clientId")).get(), messageId);
    grozaDupPubRelMessageStoreService.remove((String)channel.attr(AttributeKey.valueOf("clientId")).get(), variableHeader.messageId());
}
 

public void processPubAck(Channel channel, MqttMessageIdVariableHeader variableHeader){
    int messageId = variableHeader.messageId();
    log.info("PUBACK - clientId: {}, messageId: {}", (String) channel.attr(AttributeKey.valueOf("clientId")).get(), messageId);
    grozaDupPublishMessageStoreService.remove((String) channel.attr(AttributeKey.valueOf("clientId")).get(), messageId);

}
 

/**
 *  S - B (Qos1)
 * @param channel
 * @param variableHeader
 */
public void processPubAck(Channel channel, MqttMessageIdVariableHeader variableHeader) {
	int messageId = variableHeader.messageId();
	this.consumerProcess.processPubAck(channel, messageId);
}
 

/**
 *  S - B (Qos2)
 * @param channel
 * @param variableHeader
 */
public void processPubComp(Channel channel, MqttMessageIdVariableHeader variableHeader) {
	int messageId = variableHeader.messageId();
	this.consumerProcess.processPubComp(channel, messageId);
}