vongosling commented on a change in pull request #687:
URL: https://github.com/apache/rocketmq-externals/pull/687#discussion_r594798004
##########
File path: rocketmq-iot-bridge/pom.xml
##########
@@ -26,6 +26,16 @@ limitations under the License.
<name>rocketmq-iot-bridge ${project.version}</name>
<dependencies>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
Review comment:
tools has depends on client module.
##########
File path:
rocketmq-iot-bridge/src/main/java/org/apache/rocketmq/iot/MQTTBridge.java
##########
@@ -38,36 +38,57 @@
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttDisconnectMessageHandler;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttMessageForwarder;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttPingreqMessageHandler;
+import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttPublishMessageHandler;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttSubscribeMessageHandler;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttUnsubscribeMessagHandler;
+import org.apache.rocketmq.iot.storage.message.MessageStore;
+import org.apache.rocketmq.iot.storage.rocketmq.PublishProducer;
+import org.apache.rocketmq.iot.storage.rocketmq.RocketMQPublishProducer;
import org.apache.rocketmq.iot.storage.subscription.SubscriptionStore;
import
org.apache.rocketmq.iot.storage.subscription.impl.InMemorySubscriptionStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MQTTBridge {
+ private Logger logger = LoggerFactory.getLogger(MQTTBridge.class);
+
+ private MqttBridgeConfig bridgeConfig;
private ServerBootstrap serverBootstrap;
private NioEventLoopGroup bossGroup;
private NioEventLoopGroup workerGroup;
+
private MessageDispatcher messageDispatcher;
private SubscriptionStore subscriptionStore;
private ClientManager clientManager;
private MqttConnectionHandler connectionHandler;
- private Logger logger = LoggerFactory.getLogger(MQTTBridge.class);
+ private MessageStore messageStore;
+ private PublishProducer publishProducer;
public MQTTBridge() {
init();
}
private void init() {
- bossGroup = new
NioEventLoopGroup(MQTTBridgeConfiguration.threadNumOfBossGroup());
- workerGroup = new
NioEventLoopGroup(MQTTBridgeConfiguration.threadNumOfWorkerGroup());
+ this.bridgeConfig = new MqttBridgeConfig();
+
+ subscriptionStore = new InMemorySubscriptionStore();
+ if (bridgeConfig.isEnableRocketMQStore()) {
+ this.publishProducer = new RocketMQPublishProducer(bridgeConfig);
Review comment:
If disable, where init publishProducer?
##########
File path:
rocketmq-iot-bridge/src/main/java/org/apache/rocketmq/iot/MQTTBridge.java
##########
@@ -101,20 +124,22 @@ private void registerMessageHandlers() {
public void start() {
try {
+ publishProducer.start();
ChannelFuture channelFuture = serverBootstrap.bind().sync();
channelFuture.channel().closeFuture().sync();
+ logger.info("start the MQTTServer success.");
Review comment:
It would be better print all necesarry configuration when you starting
server. Only success or ok is too little significant.
##########
File path:
rocketmq-iot-bridge/src/main/java/org/apache/rocketmq/iot/MQTTBridge.java
##########
@@ -78,17 +99,19 @@ private void init() {
pipeline.addLast("connection-manager", connectionHandler);
}
});
- subscriptionStore = new InMemorySubscriptionStore();
- clientManager = new ClientManagerImpl();
- messageDispatcher = new MessageDispatcher(clientManager);
- connectionHandler = new MqttConnectionHandler(clientManager,
subscriptionStore);
- registerMessageHandlers();
+
+ logger.info("Mqtt bridge config: " + bridgeConfig);
}
private void registerMessageHandlers() {
messageDispatcher.registerHandler(Message.Type.MQTT_CONNECT, new
MqttConnectMessageHandler(clientManager));
messageDispatcher.registerHandler(Message.Type.MQTT_DISCONNECT, new
MqttDisconnectMessageHandler(clientManager));
- messageDispatcher.registerHandler(Message.Type.MQTT_PUBLISH, new
MqttMessageForwarder(subscriptionStore));
+ if (bridgeConfig.isEnableRocketMQStore()) {
+ messageDispatcher.registerHandler(Message.Type.MQTT_PUBLISH, new
MqttPublishMessageHandler(messageStore, publishProducer));
+ } else {
+ // TODO: mqtt cluster inner forwarder
Review comment:
I'd like to see implementation while not too many todo :)
##########
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:
?
##########
File path:
rocketmq-iot-bridge/src/main/java/org/apache/rocketmq/iot/MQTTBridge.java
##########
@@ -38,36 +38,57 @@
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttDisconnectMessageHandler;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttMessageForwarder;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttPingreqMessageHandler;
+import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttPublishMessageHandler;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttSubscribeMessageHandler;
import
org.apache.rocketmq.iot.protocol.mqtt.handler.downstream.impl.MqttUnsubscribeMessagHandler;
+import org.apache.rocketmq.iot.storage.message.MessageStore;
+import org.apache.rocketmq.iot.storage.rocketmq.PublishProducer;
+import org.apache.rocketmq.iot.storage.rocketmq.RocketMQPublishProducer;
import org.apache.rocketmq.iot.storage.subscription.SubscriptionStore;
import
org.apache.rocketmq.iot.storage.subscription.impl.InMemorySubscriptionStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MQTTBridge {
+ private Logger logger = LoggerFactory.getLogger(MQTTBridge.class);
+
+ private MqttBridgeConfig bridgeConfig;
private ServerBootstrap serverBootstrap;
private NioEventLoopGroup bossGroup;
private NioEventLoopGroup workerGroup;
+
private MessageDispatcher messageDispatcher;
private SubscriptionStore subscriptionStore;
private ClientManager clientManager;
private MqttConnectionHandler connectionHandler;
- private Logger logger = LoggerFactory.getLogger(MQTTBridge.class);
+ private MessageStore messageStore;
+ private PublishProducer publishProducer;
public MQTTBridge() {
init();
}
private void init() {
- bossGroup = new
NioEventLoopGroup(MQTTBridgeConfiguration.threadNumOfBossGroup());
- workerGroup = new
NioEventLoopGroup(MQTTBridgeConfiguration.threadNumOfWorkerGroup());
+ this.bridgeConfig = new MqttBridgeConfig();
+
+ subscriptionStore = new InMemorySubscriptionStore();
Review comment:
default InMemory subscription?
----------------------------------------------------------------
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]
