✨ 添加示例

example/mica-mqtt-example/src/main/java/net/dreamlu/iot/mqtt/server/MqttMessageInterceptor.java

+/*
+ * Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
+ *
+ * Licensed 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 net.dreamlu.iot.mqtt.server;
+
+import net.dreamlu.iot.mqtt.codec.MqttMessage;
+import net.dreamlu.iot.mqtt.core.server.interceptor.IMqttMessageInterceptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.tio.core.ChannelContext;
+
+/**
+ * mqtt 消息拦截器
+ *
+ * @author L.cm
+ */
+public class MqttMessageInterceptor implements IMqttMessageInterceptor {
+	private static final Logger logger = LoggerFactory.getLogger(MqttMessageInterceptor.class);
+
+	@Override
+	public void onAfterReceivedBytes(ChannelContext context, int receivedBytes) throws Exception {
+		// 注意：此时 clientId 可能为空
+		String clientId = context.getBsId();
+		// ChannelStat channelStat = context.stat;
+		// 自定义规则，超限是可用 Tio.remove(context, "xxx超限"); 断开连接。
+		logger.info("===接收 client:{} clientId:{} data:{}b", context.getClientNode(), clientId, receivedBytes);
+	}
+
+	@Override
+	public void onAfterDecoded(ChannelContext context, MqttMessage message, int packetSize) {
+		// 注意：此时 clientId 可能为空
+		String clientId = context.getBsId();
+		logger.info("===解码 client:{} clientId:{} message:{}", context.getClientNode(), clientId, message);
+	}
+
+	@Override
+	public void onAfterHandled(ChannelContext context, MqttMessage message, long cost) throws Exception {
+		String clientId = context.getBsId();
+		logger.info("===处理完成 ip:{} clientId:{} message:{} 耗时:{}", context.getClientNode(), clientId, message, cost);
+	}
+}

example/mica-mqtt-example/src/main/java/net/dreamlu/iot/mqtt/server/MqttServerTest.java

@@ -54,6 +54,8 @@ public class MqttServerTest {
 			})
 			// 客户端连接状态监听
 			.connectStatusListener(new MqttConnectStatusListener())
+			// 自定义消息拦截器
+//			.addInterceptor(new MqttMessageInterceptor())
 			// 开启 http
 			.httpEnable(true)
 			// http basic 认证，自定义认证，实现 HttpFilter， 注册到 MqttHttpRoutes 即可