vongosling commented on a change in pull request #687:
URL: https://github.com/apache/rocketmq-externals/pull/687#discussion_r595675827
##########
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:
Will the memory mode or rmq storage mode be the default? I don't think
the default mode is memory mode, which is not a production-available mode. If
we could reach a consensus, we should change the enable logic for the default
choice.
----------------------------------------------------------------
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]
