maixiaohai commented on a change in pull request #687:
URL: https://github.com/apache/rocketmq-externals/pull/687#discussion_r594970260
##########
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:
cluster mode need management of consumer offset and client, I'll finish
it at consumer part.
----------------------------------------------------------------
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]
