✨ 添加 mica-mqtt-broker，待完善。

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/cluster/RedisMqttMessageReceiver.java

@@ -16,9 +16,11 @@
 
 package net.dreamlu.iot.mqtt.broker.cluster;
 
+import net.dreamlu.iot.mqtt.codec.MqttMessageType;
 import net.dreamlu.iot.mqtt.codec.MqttQoS;
 import net.dreamlu.iot.mqtt.core.server.MqttServer;
 import net.dreamlu.iot.mqtt.core.server.model.Message;
+import net.dreamlu.iot.mqtt.core.server.session.IMqttSessionManager;
 import net.dreamlu.mica.core.utils.JsonUtil;
 import net.dreamlu.mica.core.utils.StringUtil;
 import net.dreamlu.mica.redis.cache.MicaRedisCache;
@@ -40,6 +42,7 @@ public class RedisMqttMessageReceiver implements MessageListener, InitializingBe
 	private final RedisTemplate<String, Object> redisTemplate;
 	private final String channel;
 	private final MqttServer mqttServer;
+	private final IMqttSessionManager sessionManager;
 
 	public RedisMqttMessageReceiver(MicaRedisCache redisCache,
 									String channel,
@@ -47,6 +50,7 @@ public class RedisMqttMessageReceiver implements MessageListener, InitializingBe
 		this.redisTemplate = redisCache.getRedisTemplate();
 		this.channel = Objects.requireNonNull(channel, "Redis pub/sub channel is null.");
 		this.mqttServer = mqttServer;
+		this.sessionManager = mqttServer.getServerCreator().getSessionManager();
 	}
 
 	@Override
@@ -57,14 +61,25 @@ public class RedisMqttMessageReceiver implements MessageListener, InitializingBe
 		if (mqttMessage == null) {
 			return;
 		}
-		String clientId = mqttMessage.getClientId();
-		String topic = mqttMessage.getTopic();
-		MqttQoS mqttQoS = MqttQoS.valueOf(mqttMessage.getQos());
-		boolean retain = mqttMessage.isRetain();
+		messageProcessing(mqttMessage);
+	}
+
+	public void messageProcessing(Message message) {
+		MqttMessageType messageType = MqttMessageType.valueOf(message.getMessageType());
+		if (MqttMessageType.PUBLISH == messageType) {
+			String clientId = message.getClientId();
+			String topic = message.getTopic();
+			MqttQoS mqttQoS = MqttQoS.valueOf(message.getQos());
+			boolean retain = message.isRetain();
 			if (StringUtil.isBlank(clientId)) {
-			mqttServer.publishAll(topic, ByteBuffer.wrap(mqttMessage.getPayload()), mqttQoS, retain);
+				mqttServer.publishAll(topic, ByteBuffer.wrap(message.getPayload()), mqttQoS, retain);
 			} else {
-			mqttServer.publish(clientId, topic, ByteBuffer.wrap(mqttMessage.getPayload()), mqttQoS, retain);
+				mqttServer.publish(clientId, topic, ByteBuffer.wrap(message.getPayload()), mqttQoS, retain);
+			}
+		} else if (MqttMessageType.SUBSCRIBE == messageType) {
+			sessionManager.addSubscribe(message.getTopic(), message.getClientId(), message.getQos());
+		} else if (MqttMessageType.UNSUBSCRIBE == messageType) {
+			sessionManager.removeSubscribe(message.getTopic(), message.getClientId());
 		}
 	}
 

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/cluster/RedisMqttMessageStore.java

@@ -61,7 +61,7 @@ public class RedisMqttMessageStore implements IMqttMessageStore {
 	}
 
 	@Override
-	public Message getRetainMessage(String topic) {
-		return null;
+	public Message getRetainMessage(String topicFilter) {
+		return redisCache.get(RedisKeys.MESSAGE_STORE_RETAIN.getKey(topicFilter));
 	}
 }

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/server/dispatcher/AbstractMqttMessageDispatcher.java

@@ -16,9 +16,11 @@
 
 package net.dreamlu.iot.mqtt.core.server.dispatcher;
 
+import net.dreamlu.iot.mqtt.codec.MqttMessageType;
 import net.dreamlu.iot.mqtt.codec.MqttQoS;
 import net.dreamlu.iot.mqtt.core.server.MqttServer;
 import net.dreamlu.iot.mqtt.core.server.model.Message;
+import net.dreamlu.iot.mqtt.core.server.session.IMqttSessionManager;
 import org.tio.core.ChannelContext;
 import org.tio.core.Tio;
 import org.tio.server.ServerTioConfig;
@@ -33,9 +35,11 @@ import java.util.Objects;
  */
 public abstract class AbstractMqttMessageDispatcher implements IMqttMessageDispatcher {
 	protected MqttServer mqttServer;
+	protected IMqttSessionManager sessionManager;
 
 	public void config(MqttServer mqttServer) {
 		this.mqttServer = mqttServer;
+		this.sessionManager = mqttServer.getServerCreator().getSessionManager();
 	}
 
 	/**
@@ -59,9 +63,16 @@ public abstract class AbstractMqttMessageDispatcher implements IMqttMessageDispa
 	public boolean send(Message message) {
 		Objects.requireNonNull(mqttServer, "MqttServer require not Null.");
 		// 1. 先发送到本服务
+		MqttMessageType messageType = MqttMessageType.valueOf(message.getMessageType());
+		if (MqttMessageType.PUBLISH == messageType) {
 			ByteBuffer payload = ByteBuffer.wrap(message.getPayload());
 			MqttQoS qoS = MqttQoS.valueOf(message.getQos());
-		mqttServer.publishAll(message.getTopic(), payload, qoS);
+			mqttServer.publishAll(message.getTopic(), payload, qoS, message.isRetain());
+		} else if (MqttMessageType.SUBSCRIBE == messageType) {
+			sessionManager.addSubscribe(message.getTopic(), message.getClientId(), message.getQos());
+		} else if (MqttMessageType.UNSUBSCRIBE == messageType) {
+			sessionManager.removeSubscribe(message.getTopic(), message.getClientId());
+		}
 		return sendAll(message);
 	}
 

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/server/http/api/MqttHttpApi.java

@@ -18,7 +18,6 @@ package net.dreamlu.iot.mqtt.core.server.http.api;
 
 import com.alibaba.fastjson.JSON;
 import net.dreamlu.iot.mqtt.codec.MqttMessageType;
-import net.dreamlu.iot.mqtt.codec.MqttQoS;
 import net.dreamlu.iot.mqtt.core.server.dispatcher.IMqttMessageDispatcher;
 import net.dreamlu.iot.mqtt.core.server.http.api.code.ResultCode;
 import net.dreamlu.iot.mqtt.core.server.http.api.form.BaseForm;
@@ -27,7 +26,6 @@ import net.dreamlu.iot.mqtt.core.server.http.api.form.SubscribeForm;
 import net.dreamlu.iot.mqtt.core.server.http.api.result.Result;
 import net.dreamlu.iot.mqtt.core.server.http.handler.MqttHttpRoutes;
 import net.dreamlu.iot.mqtt.core.server.model.Message;
-import net.dreamlu.iot.mqtt.core.server.session.IMqttSessionManager;
 import net.dreamlu.iot.mqtt.core.util.PayloadEncode;
 import org.tio.http.common.HttpRequest;
 import org.tio.http.common.HttpResponse;
@@ -45,12 +43,9 @@ import java.util.function.Function;
  */
 public class MqttHttpApi {
 	private final IMqttMessageDispatcher messageDispatcher;
-	private final IMqttSessionManager sessionManager;
 
-	public MqttHttpApi(IMqttMessageDispatcher messageDispatcher,
-					   IMqttSessionManager sessionManager) {
+	public MqttHttpApi(IMqttMessageDispatcher messageDispatcher) {
 		this.messageDispatcher = messageDispatcher;
-		this.sessionManager = sessionManager;
 	}
 
 	/**
@@ -74,7 +69,7 @@ public class MqttHttpApi {
 		if (validResponse != null) {
 			return validResponse;
 		}
-		send(form);
+		sendPublish(form);
 		return Result.ok(response);
 	}
 
@@ -105,12 +100,12 @@ public class MqttHttpApi {
 		}
 		// 批量发送
 		for (PublishForm form : formList) {
-			send(form);
+			sendPublish(form);
 		}
 		return Result.ok(response);
 	}
 
-	private void send(PublishForm form) {
+	private void sendPublish(PublishForm form) {
 		String payload = form.getPayload();
 		Message message = new Message();
 		message.setMessageType(MqttMessageType.PUBLISH.value());
@@ -151,7 +146,7 @@ public class MqttHttpApi {
 			return Result.fail(response, ResultCode.E101);
 		}
 		// 接口手动添加的订阅关系，可用来调试，不建议其他场景使用
-		sessionManager.addSubscribe(form.getTopic(), form.getClientId(), MqttQoS.valueOf(qos));
+		sendSubscribe(form);
 		return Result.ok(response);
 	}
 
@@ -187,7 +182,7 @@ public class MqttHttpApi {
 		// 批量处理
 		for (SubscribeForm form : formList) {
 			// 接口手动添加的订阅关系，可用来调试，不建议其他场景使用
-			sessionManager.addSubscribe(form.getTopic(), form.getClientId(), MqttQoS.valueOf(form.getQos()));
+			sendSubscribe(form);
 		}
 		return Result.ok(response);
 	}
@@ -214,7 +209,7 @@ public class MqttHttpApi {
 			return validResponse;
 		}
 		// 接口手动取消的订阅关系，可用来调试，不建议其他场景使用
-		sessionManager.removeSubscribe(form.getTopic(), form.getClientId());
+		sendSubscribe(form);
 		return Result.ok(response);
 	}
 
@@ -246,11 +241,24 @@ public class MqttHttpApi {
 		// 批量处理
 		for (BaseForm form : formList) {
 			// 接口手动添加的订阅关系，可用来调试，不建议其他场景使用
-			sessionManager.removeSubscribe(form.getTopic(), form.getClientId());
+			sendSubscribe(form);
 		}
 		return Result.ok(response);
 	}
 
+	private void sendSubscribe(BaseForm form) {
+		Message message = new Message();
+		message.setClientId(form.getClientId());
+		message.setTopic(form.getTopic());
+		if (form instanceof SubscribeForm) {
+			message.setQos(((SubscribeForm) form).getQos());
+			message.setMessageType(MqttMessageType.SUBSCRIBE.value());
+		} else {
+			message.setMessageType(MqttMessageType.UNSUBSCRIBE.value());
+		}
+		messageDispatcher.send(message);
+	}
+
 	/**
 	 * 读取表单
 	 *

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/server/http/core/MqttWebServer.java

@@ -317,7 +317,7 @@ public class MqttWebServer {
 				System.setProperty(TIO_SYSTEM_TIMER_PERIOD, "50");
 			}
 			// 2.2 http 路由配置
-			MqttHttpApi httpApi = new MqttHttpApi(serverCreator.getMessageDispatcher(), serverCreator.getSessionManager());
+			MqttHttpApi httpApi = new MqttHttpApi(serverCreator.getMessageDispatcher());
 			httpApi.register();
 			// 2.3 认证配置
 			String username = serverCreator.getHttpBasicUsername();

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/server/session/IMqttSessionManager.java

@@ -37,7 +37,7 @@ public interface IMqttSessionManager {
 	 * @param clientId    客户端 Id
 	 * @param mqttQoS     MqttQoS
 	 */
-	void addSubscribe(String topicFilter, String clientId, MqttQoS mqttQoS);
+	void addSubscribe(String topicFilter, String clientId, int mqttQoS);
 
 	/**
 	 * 删除订阅

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/server/session/InMemoryMqttSessionManager.java

@@ -51,12 +51,12 @@ public class InMemoryMqttSessionManager implements IMqttSessionManager {
 	private final ConcurrentMap<String, Map<Integer, MqttPendingQos2Publish>> pendingQos2PublishStore = new ConcurrentHashMap<>();
 
 	@Override
-	public void addSubscribe(String topicFilter, String clientId, MqttQoS mqttQoS) {
+	public void addSubscribe(String topicFilter, String clientId, int mqttQoS) {
 		Map<String, Integer> data = subscribeStore.computeIfAbsent(topicFilter, (key) -> new ConcurrentHashMap<>(16));
 		// 如果不存在或者老的订阅 qos 比较小也重新设置
 		Integer existingQos = data.get(clientId);
-		if (existingQos == null || existingQos < mqttQoS.value()) {
-			data.put(clientId, mqttQoS.value());
+		if (existingQos == null || existingQos < mqttQoS) {
+			data.put(clientId, mqttQoS);
 		}
 	}
 

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/server/support/DefaultMqttServerProcessor.java

@@ -268,7 +268,7 @@ public class DefaultMqttServerProcessor implements MqttServerProcessor {
 			MqttQoS mqttQoS = subscription.qualityOfService();
 			mqttQosList.add(mqttQoS);
 			topicList.add(topicName);
-			sessionManager.addSubscribe(topicName, clientId, mqttQoS);
+			sessionManager.addSubscribe(topicName, clientId, mqttQoS.value());
 		}
 		logger.info("Subscribe - clientId:{} TopicFilters:{} mqttQoS:{} messageId:{}", clientId, topicList, mqttQosList, messageId);
 		// 3. 返回 ack