Fixing invalid message state transition

Resolves wso2#1005
tharinduwijewardane · Feb 22, 2019 · 4b8715e · 4b8715e
1 parent 3bcc2ec
commit 4b8715e
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 3 deletions.
diff --git a/modules/andes-core/broker/src/main/java/org/wso2/andes/kernel/MessageStatus.java b/modules/andes-core/broker/src/main/java/org/wso2/andes/kernel/MessageStatus.java
@@ -41,6 +41,11 @@ public enum MessageStatus {
      */
     SCHEDULED_TO_SEND,
 
+    /**
+     * Message has been recovered by the client it was sent to
+     */
+    RECOVERED,
+
     /**
      * In a topic scenario, all subscribed consumers have acknowledged receipt of message
      */
@@ -127,9 +132,12 @@ public boolean isValidPreviousState(MessageStatus previousState) {
         NO_MATCHING_CONSUMER.next = EnumSet.of(DLC_MESSAGE);
         NO_MATCHING_CONSUMER.previous = EnumSet.of(BUFFERED);
 
-        SCHEDULED_TO_SEND.next = EnumSet.of(EXPIRED, ACKED_BY_ALL, BUFFERED, DLC_MESSAGE, SLOT_RETURNED);
+        SCHEDULED_TO_SEND.next = EnumSet.of(EXPIRED, RECOVERED, ACKED_BY_ALL, BUFFERED, DLC_MESSAGE, SLOT_RETURNED);
         SCHEDULED_TO_SEND.previous = EnumSet.of(BUFFERED);
 
+        RECOVERED.next = EnumSet.of(SCHEDULED_TO_SEND, EXPIRED, DLC_MESSAGE, SLOT_RETURNED);
+        RECOVERED.previous = EnumSet.of(SCHEDULED_TO_SEND);
+
         ACKED_BY_ALL.next = EnumSet.of(PREPARED_TO_DELETE, SLOT_RETURNED);
         ACKED_BY_ALL.previous = EnumSet.of(SCHEDULED_TO_SEND);
 

diff --git a/...-core/broker/src/main/java/org/wso2/andes/kernel/NoLossBurstTopicMessageDeliveryImpl.java b/...-core/broker/src/main/java/org/wso2/andes/kernel/NoLossBurstTopicMessageDeliveryImpl.java
@@ -114,8 +114,6 @@ public int deliverMessageToSubscriptions(StorageQueue storageQueue) throws
                     break;
                 }
 
-                message.markAsScheduledToDeliver(subscriptionsToDeliver);
-
                 iterator.remove();
                 for (AndesSubscription localSubscription : subscriptionsToDeliver) {
                     MessageFlusher.getInstance().deliverMessageAsynchronously(localSubscription, message);

diff --git a/...andes-core/broker/src/main/java/org/wso2/andes/kernel/subscription/AndesSubscription.java b/...andes-core/broker/src/main/java/org/wso2/andes/kernel/subscription/AndesSubscription.java
@@ -388,6 +388,7 @@ public void rebufferUnackedMessages() throws AndesException {
      */
     private void reDeliverMessage(DeliverableAndesMetadata messageMetadata) throws AndesException {
         if (!messageMetadata.isOKToDispose()) {
+            messageMetadata.addMessageStatus(MessageStatus.RECOVERED);
             MessageFlusher.getInstance().scheduleMessageForSubscription(this, messageMetadata);
             MessageTracer.trace(messageMetadata, MessageTracer.MESSAGE_REQUEUED_SUBSCRIBER);
         } else {