[ISSUE #305]feat: Introduce Standalone Mode by Allowing Meta Module Disabling

README.md

@@ -23,6 +23,14 @@ For example, set the following parameters to true in broker.conf
 enableLmq = true
 enableMultiDispatch = true
 ```
+
+### Deployment Modes
+RocketMQ-MQTT supports two deployment modes:
+
++ Cluster-Ready Mode (Meta Enabled, Default): This is the default and full-featured mode. It relies on a meta service (based on RocketMQ's KV storage) for dynamic configuration, cluster node discovery, and advanced features like Retain Messages and Will Messages. This mode is suitable for production and high-availability environments.
+
++ Standalone Mode (Meta Disabled): This is a simplified mode designed for single-node deployments or scenarios where advanced features are not required. It operates without any dependency on the meta service, making configuration and deployment much simpler. In this mode, features like Retain Messages and Will Messages are disabled.
+
 ### Build Requirements
 The current project requires JDK 1.8.x.  When building on MAC arm64 the recommended JDK 1.8 must be based on 386 architecture or use Maven flag `-Dos.arch=x86_64` when building with Maven.
 
@@ -43,15 +51,17 @@ cd conf
 ```
 Some important configuration items in the **service.conf** configuration file 
 
-| **Config Key**        | **Instruction**                                               |
-|-----------------------|---------------------------------------------------------------|
-| username              | used for auth                                                 |
-| secretKey             | used for auth                                                 |
-| NAMESRV_ADDR          | specify namesrv address                                       |
-| eventNotifyRetryTopic | notify event retry topic                                      |
-| clientRetryTopic      | client retry topic                                            |
-| metaAddr              | meta all nodes ip:port. Same as membersAddress in meta.config |
-
+| **Config Key**        | **Instruction**                                                                                                                  |
+|-----------------------|----------------------------------------------------------------------------------------------------------------------------------|
+| username              | used for auth                                                                                                                    |
+| secretKey             | used for auth                                                                                                                    |
+| NAMESRV_ADDR          | specify namesrv address                                                                                                          |
+| eventNotifyRetryTopic | notify event retry topic                                                                                                         |
+| clientRetryTopic      | client retry topic                                                                                                               |
+| enableMetaModule   	| (Optional) Switch for meta module. Defaults to true. Set to `false` to enable Standalone Mode.                                   |
+| metaAddr              | meta all nodes ip:port. Same as membersAddress in meta.config. **Required when `enableMetaModule` is `true`.**                   |
+| staticFirstTopics     | A comma-separated list of all first-level MQTT topics. E.g., `chat,iot`. **Required when `enableMetaModule` is `false`.**        |
+| localAddress          | The address for the server's internal RPC loopback. `127.0.0.1` is recommended. **Required when `enableMetaModule` is `false`.** |
 
 And some configuration items in the **meta.conf** configuration file
 
@@ -80,6 +90,7 @@ sh mqadmin updateKvConfig -s LMQ -k ALL_FIRST_TOPICS -v {topic1,topic2} -n {name
 sh mqadmin updateKvConfig  -s LMQ -k {topic} -v {topic/+}  -n {namesrv}
 ```
 6. Start Process
+Note: In standalone mode, Only mqtt.sh is needed to start.
 ```shell
 cd bin
 sh meta.sh start

mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/MqttRetainException.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.mqtt.common.model;
+
+public class MqttRetainException extends RuntimeException {
+    public MqttRetainException() { }
+
+    public MqttRetainException(String message) {
+        super(message);
+    }
+}

mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/mqtt/MqttPacketDispatcher.java

@@ -23,6 +23,7 @@
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.mqtt.MqttConnectMessage;
 import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageType;
 import io.netty.handler.codec.mqtt.MqttPubAckMessage;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
 import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
@@ -32,9 +33,11 @@
 import org.apache.rocketmq.mqtt.common.hook.UpstreamHookManager;
 import org.apache.rocketmq.mqtt.common.model.MqttMessageUpContext;
 import org.apache.rocketmq.mqtt.common.util.HostInfo;
+import org.apache.rocketmq.mqtt.cs.channel.ChannelCloseFrom;
 import org.apache.rocketmq.mqtt.cs.channel.ChannelDecodeException;
 import org.apache.rocketmq.mqtt.cs.channel.ChannelException;
 import org.apache.rocketmq.mqtt.cs.channel.ChannelInfo;
+import org.apache.rocketmq.mqtt.cs.channel.ChannelManager;
 import org.apache.rocketmq.mqtt.cs.protocol.mqtt.handler.MqttConnectHandler;
 import org.apache.rocketmq.mqtt.cs.protocol.mqtt.handler.MqttDisconnectHandler;
 import org.apache.rocketmq.mqtt.cs.protocol.mqtt.handler.MqttPingHandler;
@@ -90,6 +93,9 @@ public class MqttPacketDispatcher extends SimpleChannelInboundHandler<MqttMessag
 
     @Resource
     private UpstreamHookManager upstreamHookManager;
+
+    @Resource
+    private ChannelManager channelManager;
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
@@ -134,6 +140,14 @@ protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws E
                 ctx.fireExceptionCaught(new ChannelException("UpstreamHook Result Unknown"));
                 return;
             }
+
+            if (!hookResult.isSuccess() && msg.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
+                logger.warn("Request (PUBLISH) from client {} was rejected by hook chain. Reason: {}. Closing connection.",
+                        ChannelInfo.getClientId(ctx.channel()), hookResult.getRemark());
+
+                channelManager.closeConnect(ctx.channel(), ChannelCloseFrom.SERVER, hookResult.getRemark());
+                return; 
+            }
             try {
                 _channelRead0(ctx, msg, hookResult);
             } catch (Throwable t) {

mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/mqtt/handler/MqttConnectHandler.java

@@ -35,6 +35,7 @@
 import org.apache.rocketmq.mqtt.cs.protocol.mqtt.facotry.MqttMessageFactory;
 import org.apache.rocketmq.mqtt.cs.session.loop.SessionLoop;
 import org.apache.rocketmq.mqtt.cs.session.loop.WillLoop;
+import org.apache.rocketmq.mqtt.ds.config.ServiceConf;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
@@ -52,6 +53,9 @@ public class MqttConnectHandler implements MqttPacketHandler<MqttConnectMessage>
     @Resource
     private ChannelManager channelManager;
 
+    @Resource
+    private ServiceConf serviceConf;
+
     @Resource
     private SessionLoop sessionLoop;
 
@@ -111,15 +115,21 @@ public void doHandler(ChannelHandlerContext ctx, MqttConnectMessage connectMessa
             sessionLoop.loadSession(ChannelInfo.getClientId(channel), channel);
 
             // save will message
-            WillMessage willMessage = null;
             if (variableHeader.isWillFlag()) {
+                if (!serviceConf.isEnableMetaModule()) {
+                    String clientId = ChannelInfo.getClientId(channel);
+                    logger.error("Client [{}] trying to set a Will Message, but the meta module is disabled. Connection refused.", clientId);
+                    channelManager.closeConnect(channel, ChannelCloseFrom.SERVER, "Will Message feature is disabled");
+                    return; 
+                }
+
                 if (payload.willTopic() == null || payload.willMessageInBytes() == null) {
                     logger.error("Will message and will topic can not be empty");
                     channelManager.closeConnect(channel, ChannelCloseFrom.SERVER, "Will message and will topic can not be empty");
                     return;
                 }
 
-                willMessage = new WillMessage(payload.willTopic(), payload.willMessageInBytes(), variableHeader.isWillRetain(), variableHeader.willQos());
+                WillMessage willMessage = new WillMessage(payload.willTopic(), payload.willMessageInBytes(), variableHeader.isWillRetain(), variableHeader.willQos());
                 willLoop.addWillMessage(channel, willMessage);
             }
 

mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/WillLoop.java

@@ -31,6 +31,7 @@
 import org.apache.rocketmq.mqtt.cs.channel.ChannelInfo;
 import org.apache.rocketmq.mqtt.cs.config.WillLoopConf;
 import org.apache.rocketmq.mqtt.cs.session.infly.MqttMsgId;
+import org.apache.rocketmq.mqtt.ds.config.ServiceConf;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
@@ -63,12 +64,21 @@ public class WillLoop {
     @Resource
     private WillMsgSender willMsgSender;
 
+    @Resource
+    private ServiceConf serviceConf;
+
     public void setWillLoopConf(WillLoopConf willLoopConf) {
         this.willLoopConf = willLoopConf;
     }
 
     @PostConstruct
     public void init() {
+
+        if (!serviceConf.isEnableMetaModule()) {
+            logger.info("Meta module is disabled, WillLoop will not be initialized.");
+            return; 
+        }
+
         aliveService.scheduleWithFixedDelay(() -> csLoop(), 15 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
         aliveService.scheduleWithFixedDelay(() -> masterLoop(), 10 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
 
@@ -239,6 +249,11 @@ private void _handleShutDownCS(String ip, AtomicInteger loopNum) {
     }
 
     public void closeConnect(Channel channel, String clientId, String reason) {
+
+        if (!serviceConf.isEnableMetaModule()) {     
+            return;
+        }
+
         String ip = IpUtil.getLocalAddressCompatible();
         String willKey = ip + Constants.CTRL_1 + clientId;
         CompletableFuture<byte[]> willMessageFuture = willMsgPersistManager.get(willKey);

mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/config/ServiceConf.java

@@ -46,6 +46,10 @@ public class ServiceConf {
     private String metaAddr;
     private long retainMsgExpire = 3 * 24 * 60 * 60 * 1000L;
 
+    private boolean enableMetaModule = true;
+    private String staticFirstTopics;
+    private String localAddress;
+
     @PostConstruct
     public void init() throws IOException {
         ClassPathResource classPathResource = new ClassPathResource(CONF_FILE_NAME);
@@ -62,8 +66,17 @@ public void init() throws IOException {
         if (StringUtils.isBlank(eventNotifyRetryTopic)) {
             throw new RemoteException("eventNotifyRetryTopic is blank");
         }
-        if (StringUtils.isBlank(metaAddr)) {
-            throw new RemoteException("metaAddr is blank");
+        if (this.enableMetaModule) {
+            if (StringUtils.isBlank(metaAddr)) {
+                throw new IOException("metaAddr is blank when meta module is enabled");
+            }
+        } else {
+            if (StringUtils.isBlank(staticFirstTopics)) {
+                throw new IOException("When meta module is disabled, staticFirstTopics cannot be blank.");
+            }
+            if (StringUtils.isBlank(localAddress)) {
+                throw new IOException("When meta module is disabled, localAddress cannnot be blank.");
+            }
         }
     }
 
@@ -158,4 +171,28 @@ public long getRetainMsgExpire() {
     public void setRetainMsgExpire(long retainMsgExpire) {
         this.retainMsgExpire = retainMsgExpire;
     }
+
+    public String getStaticFirstTopics() {
+        return staticFirstTopics;
+    }
+
+    public void setStaticFirstTopics(String staticFirstTopics) {
+        this.staticFirstTopics = staticFirstTopics;
+    }
+
+    public boolean isEnableMetaModule() {
+        return enableMetaModule;
+    }
+
+    public void setEnableMetaModule(boolean enableMetaModule) {
+        this.enableMetaModule = enableMetaModule;
+    }
+
+    public String getLocalAddress() {
+        return localAddress;
+    }
+
+    public void setLocalAddress(String localAddress) {
+        this.localAddress = localAddress;
+    }
 }

mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/meta/MetaRpcClient.java

@@ -59,6 +59,12 @@ public class MetaRpcClient {
 
     @PostConstruct
     public void init() throws InterruptedException, TimeoutException {
+
+        if (!serviceConf.isEnableMetaModule()) {
+            logger.info("Meta module is disabled, MetaRpcClient will not be initialized.");
+            return; 
+        }
+
         initRpcServer();
         cliClientService = new CliClientServiceImpl();
         cliClientService.init(new CliOptions());