[broker-30] Avoid double message retaining

Author: crazyrokr

core-service/src/main/java/javasabr/mqtt/service/PublishDeliveringService.java

@@ -2,7 +2,6 @@
 
 import javasabr.mqtt.model.publishing.Publish;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
-import javasabr.mqtt.model.topic.TopicFilter;
 import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 import javasabr.rlib.collections.array.Array;
 

core-service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishDeliveringService.java

@@ -52,16 +52,10 @@ public DefaultPublishDeliveringService(
 
   @Override
   public PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber) {
-    try {
-      if (publish.retained()) {
-        retainedMessageTree.retainMessage(publish);
-      }
-      //noinspection DataFlowIssue
-      return publishOutMessageHandlers[subscriber.qos().level()].handle(publish, subscriber);
-    } catch (IndexOutOfBoundsException | NullPointerException ex) {
-      log.warning(publish, "Received not supported publish message:[%s]"::formatted);
-      return PublishHandlingResult.UNSPECIFIED_ERROR;
+    if (publish.retained()) {
+      retainedMessageTree.retainMessage(publish);
     }
+    return startDeliveringWithoutRetain(publish, subscriber);
   }
 
   @Override
@@ -72,11 +66,21 @@ public Array<PublishHandlingResult> deliverRetainedMessages(SingleSubscriber sub
     Array<Publish> retainedMessages = retainedMessageTree.getRetainedMessage(subscription.topicFilter(), transformer);
     MutableArray<PublishHandlingResult> result = MutableArray.ofType(PublishHandlingResult.class);
     for (Publish message : retainedMessages) {
-      result.add(startDelivering(message, subscriber));
+      result.add(startDeliveringWithoutRetain(message, subscriber));
     }
     return Array.copyOf(result);
   }
 
+  private PublishHandlingResult startDeliveringWithoutRetain(Publish publish, SingleSubscriber subscriber) {
+    try {
+      //noinspection DataFlowIssue
+      return publishOutMessageHandlers[subscriber.qos().level()].handle(publish, subscriber);
+    } catch (IndexOutOfBoundsException | NullPointerException ex) {
+      log.warning(publish, "Received not supported publish message:[%s]"::formatted);
+      return PublishHandlingResult.UNSPECIFIED_ERROR;
+    }
+  }
+
   private static String buildServiceDescription(
       @Nullable MqttPublishOutMessageHandler[] publishOutMessageHandlers) {
     var builder = new StringBuilder();