addressing review

Signed-off-by: dorjesinpo <[email protected]>
bloomberg · Dec 19, 2024 · 7aee933 · 7aee933
1 parent a6dcd0d
commit 7aee933
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 9 deletions.
diff --git a/src/groups/mqb/mqbblp/mqbblp_pushstream.h b/src/groups/mqb/mqbblp/mqbblp_pushstream.h
@@ -127,8 +127,12 @@ struct PushStream {
         void add(Element* element);
         void remove(Element* element);
 
-        /// Return 'true'
-        bool setLastPush(const bmqt::MessageGUID& lastGUID);
+        /// Return 'true' if the specified `guid` is the same as in the last
+        /// `setLastPush` call.
+        bool isLastPush(const bmqt::MessageGUID& guid);
+
+        /// Cache the specified `guid` for subsequent checks by `isLastPush`.
+        void setLastPush(const bmqt::MessageGUID& guid);
 
         const Element* last() const;
     };
@@ -620,14 +624,14 @@ inline void PushStream::App::remove(Element* element)
     d_elements.remove(element, e_APP);
 }
 
-inline bool PushStream::App::setLastPush(const bmqt::MessageGUID& lastGUID)
+inline bool PushStream::App::isLastPush(const bmqt::MessageGUID& lastGUID)
 {
-    if (d_lastGUID == lastGUID) {
-        return false;
-    }
-    d_lastGUID = lastGUID;
+    return d_lastGUID == lastGUID;
+}
 
-    return true;
+inline void PushStream::App::setLastPush(const bmqt::MessageGUID& lastGUID)
+{
+    d_lastGUID = lastGUID;
 }
 
 inline const PushStream::Element* PushStream::App::last() const

diff --git a/src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp b/src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp
@@ -603,7 +603,13 @@ void RelayQueueEngine::deliverMessages()
 
                 d_storageIter_mp->removeCurrentElement();
             }
-            else if (element->app().setLastPush(d_storageIter_mp->guid())) {
+            else if (element->app().isLastPush(d_storageIter_mp->guid())) {
+                // This `app` has already seen this message.
+                d_storageIter_mp->removeCurrentElement();
+            }
+            else {
+                element->app().setLastPush(d_storageIter_mp->guid());
+
                 if (d_appsDeliveryContext.processApp(*app, i)) {
                     // The current element has made it either to delivery or
                     // putAside and it can be removed