maixiaohai commented on a change in pull request #687:
URL: https://github.com/apache/rocketmq-externals/pull/687#discussion_r594986415
##########
File path:
rocketmq-iot-bridge/src/main/java/org/apache/rocketmq/iot/protocol/mqtt/handler/downstream/impl/MqttPublishMessageHandler.java
##########
@@ -26,26 +27,52 @@
import org.apache.rocketmq.iot.protocol.mqtt.handler.MessageHandler;
import org.apache.rocketmq.iot.storage.message.MessageStore;
import org.apache.rocketmq.iot.common.util.MessageUtil;
+import org.apache.rocketmq.iot.storage.rocketmq.PublishProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class MqttPublishMessageHandler implements MessageHandler {
+ private Logger log =
LoggerFactory.getLogger(MqttSubscribeMessageHandler.class);
private MessageStore messageStore;
+ private PublishProducer publishProducer;
- public MqttPublishMessageHandler(MessageStore messageStore) {
+ public MqttPublishMessageHandler(MessageStore messageStore,
PublishProducer publishProducer) {
this.messageStore = messageStore;
+ this.publishProducer = publishProducer;
}
@Override public void handleMessage(Message message) {
Client client = message.getClient();
MqttPublishMessage publishMessage = (MqttPublishMessage)
message.getPayload();
- messageStore.put(message);
- int qos =
MessageUtil.actualQos(publishMessage.fixedHeader().qosLevel().value());
- if (qos == MqttQoS.AT_LEAST_ONCE.value()) {
- MqttPubAckMessage pubAckMessage =
MessageUtil.getMqttPubackMessage(publishMessage);
- client.getCtx().writeAndFlush(pubAckMessage);
- } else if (qos == MqttQoS.EXACTLY_ONCE.value()) {
- MqttMessage pubrecMessage =
MessageUtil.getMqttPubrecMessage(publishMessage);
- client.getCtx().writeAndFlush(pubrecMessage);
+
+ try {
+ publishProducer.send(publishMessage, client);
+ } catch (Exception e) {
+ log.error("send msg to rocketMQ failed. clientId:" +
client.getId(), e);
+ }
+
+ // TODO: qos1, qos2
Review comment:
qos1 need consumer response.
qos2 means Exactly once, it's difficult to implemet.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
