Merge branch 'v1.0' into v1.2

Author: Amazingwujun

src/main/java/com/jun/mqttx/broker/handler/PublishHandler.java

@@ -404,7 +404,6 @@ private void publish0(ClientSub clientSub, PubMsg pubMsg, boolean isClusterMessa
     public void handleRetainMsg(PubMsg pubMsg) {
         byte[] payload = pubMsg.getPayload();
         String topic = pubMsg.getTopic();
-        int qos = pubMsg.getQoS();
 
         // with issue https://github.com/Amazingwujun/mqttx/issues/14 And PR
         // https://github.com/Amazingwujun/mqttx/pull/15
@@ -415,15 +414,16 @@ public void handleRetainMsg(PubMsg pubMsg) {
         // no retained message for that topic [MQTT-3.3.1-7].
         // [MQTT-3.3.1-7] 当 broker 收到 qos 为 0 并且 RETAIN = 1 的消息 必须抛弃该主题保留
         // 之前的消息（注意：是 retained 消息）, 同时 broker 可以选择保留或抛弃当前的消息，MQTTX
-        // 的选择是抛弃.
+        // 的选择是保留.
+
         // A PUBLISH Packet with a RETAIN flag set to 1 and a payload containing zero
         // bytes will be processed as normal by the Server and sent to Clients with a
         // subscription matching the topic name. Additionally any existing retained
         // message with the same topic name MUST be removed and any future subscribers
         // for the topic will not receive a retained message [MQTT-3.3.1-10].
         // [MQTT-3.3.1-10] 注意 [Additionally] 内容， broker 收到 retain 消息载荷（payload）
         // 为空时，broker 必须移除 topic 关联的 retained 消息.
-        if (MqttQoS.AT_MOST_ONCE.value() == qos || ObjectUtils.isEmpty(payload)) {
+        if (ObjectUtils.isEmpty(payload)) {
             retainMessageService.remove(topic);
             return;
         }

src/main/java/com/jun/mqttx/service/ISubscriptionService.java

@@ -63,13 +63,6 @@ public interface ISubscriptionService {
      */
     void clearClientSubscriptions(String clientId, boolean cleanSession);
 
-    /**
-     * 移除指定 topic
-     *
-     * @param topic 主题
-     */
-    void removeTopic(String topic);
-
     /**
      * 移除未包含在 authorizedSub 集合中的客户端订阅
      *
@@ -107,4 +100,4 @@ public interface ISubscriptionService {
      * @param clientId 客户 id
      */
     void clearClientSysSub(String clientId);
-}
+}

src/main/java/com/jun/mqttx/service/impl/DefaultSubscriptionServiceImpl.java

@@ -253,39 +253,6 @@ public void clearClientSubscriptions(String clientId, boolean cleanSession) {
         }
     }
 
-    /**
-     * 移除 topic
-     *
-     * @param topic 主题
-     */
-    @Override
-    public void removeTopic(String topic) {
-        inMemTopics.remove(topic);
-        Optional.ofNullable(inMemTopicClientsMap.remove(topic))
-                .ifPresent(e -> e.forEach(
-                        clientSub -> Optional.
-                                ofNullable(inMemClientTopicsMap.get(clientSub.getClientId()))
-                                .ifPresent(t -> t.remove(topic)))
-                );
-        stringRedisTemplate.opsForSet().remove(topicSetKey, topic);
-
-        // 抓取订阅主题的全部 clientId, 通过 clientId 移除所有其订阅的 topic
-        Set<Object> keys = stringRedisTemplate.opsForHash().keys(topicPrefix + topic);
-        keys.forEach(t -> stringRedisTemplate.opsForSet().remove(clientTopicsPrefix + t, topic));
-        stringRedisTemplate.delete(topicPrefix + topic);
-
-        if (enableInnerCache) {
-            removeTopicWithCache(topic);
-
-            // 集群广播
-            if (enableCluster) {
-                ClientSubOrUnsubMsg clientSubOrUnsubMsg = new ClientSubOrUnsubMsg(null, 0, topic, false, null, DEL_TOPIC);
-                InternalMessage<ClientSubOrUnsubMsg> im = new InternalMessage<>(clientSubOrUnsubMsg, System.currentTimeMillis(), brokerId);
-                internalMessagePublishService.publish(im, InternalMessageEnum.SUB_UNSUB.getChannel());
-            }
-        }
-    }
-
     @Override
     public void clearUnAuthorizedClientSub(String clientId, List<String> authorizedSub) {
         List<String> collect = inDiskTopics

src/main/resources/application.yml

@@ -2,7 +2,7 @@
 
 spring:
   application:
-    name: @broker.name@
+    name: @project.name@
   profiles:
     # 该配置决定使用 application-dev.yml 还是 application-prod.yml 作为附加配置文件
     active: @profile@