cleaning

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>
bloomberg · Sep 3, 2024 · cb3c67b · cb3c67b
1 parent 018c8c4
commit cb3c67b
Show file tree

Hide file tree

Showing 10 changed files with 49 additions and 41 deletions.
diff --git a/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp b/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp
@@ -5310,11 +5310,13 @@ void ClusterQueueHelper::requestToStopPushing()
     }
 }
 
-void ClusterQueueHelper::onDeconfiguredHandle(
-    const bsl::shared_ptr<StopContext>& contextSp)
+void ClusterQueueHelper::contextHolder(
+    const bsl::shared_ptr<StopContext>& contextSp,
+    const VoidFunctor&                  action)
 {
-    BALL_LOG_INFO << d_clusterData_p->identity().description()
-                  << ": deconfiguring " << " " << contextSp.numReferences();
+    if (action) {
+        action();
+    }
     (void)contextSp;
 }
 
@@ -5408,9 +5410,10 @@ void ClusterQueueHelper::processNodeStoppingNotification(
                      ++cit) {
                     cit->second.d_handle_p->deconfigureAll(
                         bdlf::BindUtil::bind(
-                            &ClusterQueueHelper::onDeconfiguredHandle,
+                            &ClusterQueueHelper::contextHolder,
                             this,
-                            contextSp));
+                            contextSp,
+                            VoidFunctor()));
                 }
                 BALL_LOG_INFO << d_clusterData_p->identity().description()
                               << ": deconfigured " << handles.size()
@@ -5476,21 +5479,27 @@ void ClusterQueueHelper::processNodeStoppingNotification(
                     continue;  // CONTINUE
                 }
 
+                VoidFunctor inner = bdlf::BindUtil::bind(
+                    &mqbi::Queue::onOpenUpstream,
+                    queue,
+                    0,
+                    bmqp::QueueId::k_DEFAULT_SUBQUEUE_ID,
+                    true);
+
+                VoidFunctor outer = bdlf::BindUtil::bind(
+                    &ClusterQueueHelper::contextHolder,
+                    this,
+                    contextSp,
+                    inner);
+
                 queue->dispatcher()->execute(
-                    bdlf::BindUtil::bind(&mqbi::Queue::onOpenUpstream,
-                                         queue,
-                                         0,
-                                         bmqp::QueueId::k_DEFAULT_SUBQUEUE_ID,
-                                         true),  // isWriterOnly
-                    queue);
-            }
+                    outer,
+                    queue,
+                    mqbi::DispatcherEventType::e_DISPATCHER);
 
-            // As a way to bind 'contextSp' to all 'onOpenUpstream'
-            d_cluster_p->dispatcher()->execute(
-                bdlf::BindUtil::bind(&ClusterQueueHelper::onDeconfiguredHandle,
-                                     this,
-                                     contextSp),
-                mqbi::DispatcherClientType::e_QUEUE);
+                // Use 'mqbi::DispatcherEventType::e_DISPATCHER' to avoid
+                // (re)enabling 'd_flushList'
+            }
         }
         else {
             // TEMPORARY, remove 'after switching to StopRequest V2

diff --git a/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.h b/src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.h
@@ -915,7 +915,8 @@ class ClusterQueueHelper : public mqbc::ClusterStateObserver,
     /// Send StopResponse to the request in the specified 'context.
     void finishStopSequenceDispatched(StopContext* context);
 
-    void onDeconfiguredHandle(const bsl::shared_ptr<StopContext>& contextSp);
+    void contextHolder(const bsl::shared_ptr<StopContext>& contextSp,
+                       const VoidFunctor&                  action);
 
     // PRIVATE ACCESSORS
 

diff --git a/src/groups/mqb/mqbblp/mqbblp_queue.cpp b/src/groups/mqb/mqbblp/mqbblp_queue.cpp
@@ -957,9 +957,7 @@ bsls::Types::Int64 Queue::countUnconfirmed(unsigned int subId)
 
 void Queue::stopPushing()
 {
-    d_state.routingContext().shutdown();
-
-    queueEngine()->resetState(true);  // keepConfirming
+    queueEngine()->resetState(true);  // isShuttingDown
 }
 
 }  // close package namespace

diff --git a/src/groups/mqb/mqbblp/mqbblp_queuehandle.cpp b/src/groups/mqb/mqbblp/mqbblp_queuehandle.cpp
@@ -1000,7 +1000,11 @@ void QueueHandle::deconfigureAll(
         bdlf::BindUtil::bind(&QueueHandle::deconfigureDispatched,
                              this,
                              deconfiguredCb),
-        d_queue_sp.get());
+        d_queue_sp.get(),
+        mqbi::DispatcherEventType::e_DISPATCHER);
+
+    // Use 'mqbi::DispatcherEventType::e_DISPATCHER' to avoid (re)enabling
+    // 'd_flushList'
 }
 
 void QueueHandle::deconfigureDispatched(

diff --git a/src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp b/src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp
@@ -909,15 +909,18 @@ int RelayQueueEngine::configure(
     return 0;
 }
 
-void RelayQueueEngine::resetState(bool keepConfirming)
+void RelayQueueEngine::resetState(bool isShuttingDown)
 {
     // d_self.reset(this);
 
     for (AppsMap::iterator it = d_apps.begin(); it != d_apps.end(); ++it) {
         it->second->reset();
-        // Keep the routing which new engine can reuse
+        if (isShuttingDown) {
+            it->second->d_routing_sp->reset();
+        }
+        // else, keep the routing which new engine can reuse
     }
-    if (!keepConfirming) {
+    if (!isShuttingDown) {
         d_apps.clear();
     }
 }

diff --git a/src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h b/src/groups/mqb/mqbblp/mqbblp_relayqueueengine.h
@@ -379,7 +379,7 @@ class RelayQueueEngine : public mqbi::QueueEngine {
     /// Reset the internal state of this engine.  If the optionally specified
     /// 'keepConfirming' is 'true', keep the data structures for CONFIRMs
     /// processing.
-    virtual void resetState(bool keepConfirming = false) BSLS_KEYWORD_OVERRIDE;
+    virtual void resetState(bool isShuttingDown = false) BSLS_KEYWORD_OVERRIDE;
 
     /// Rebuild the internal state of this engine.  This method is invoked
     /// when the queue this engine is associated with is created from an

diff --git a/src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp b/src/groups/mqb/mqbblp/mqbblp_rootqueueengine.cpp
@@ -425,7 +425,7 @@ int RootQueueEngine::initializeAppId(const bsl::string& appId,
     return 0;
 }
 
-void RootQueueEngine::resetState(bool keepConfirming)
+void RootQueueEngine::resetState(bool isShuttingDown)
 {
     for (Apps::iterator it = d_apps.begin(); it != d_apps.end(); ++it) {
         it->value()->reset();
@@ -434,7 +434,7 @@ void RootQueueEngine::resetState(bool keepConfirming)
 
     d_consumptionMonitor.reset();
 
-    if (!keepConfirming) {
+    if (!isShuttingDown) {
         d_apps.clear();
     }
 }

diff --git a/src/groups/mqb/mqbblp/mqbblp_rootqueueengine.h b/src/groups/mqb/mqbblp/mqbblp_rootqueueengine.h
@@ -261,7 +261,7 @@ class RootQueueEngine BSLS_KEYWORD_FINAL : public mqbi::QueueEngine {
     /// Reset the internal state of this engine.  If the optionally specified
     /// 'keepConfirming' is 'true', keep the data structures for CONFIRMs
     /// processing.
-    virtual void resetState(bool keepConfirming = false) BSLS_KEYWORD_OVERRIDE;
+    virtual void resetState(bool isShuttingDown = false) BSLS_KEYWORD_OVERRIDE;
 
     /// Rebuild the internal state of this engine.  This method is invoked
     /// when the queue this engine is associated with is created from an

diff --git a/src/groups/mqb/mqbblp/mqbblp_routers.h b/src/groups/mqb/mqbblp/mqbblp_routers.h
@@ -609,8 +609,6 @@ class Routers {
         bool onUsable(unsigned int* upstreamSubQueueId,
                       unsigned int  upstreamSubscriptionId);
 
-        void shutdown();
-
         void loadInternals(mqbcmd::Routing* out) const;
     };
 
@@ -885,11 +883,6 @@ Routers::QueueRoutingContext::onUsable(unsigned int* upstreamSubQueueId,
     return false;
 }
 
-inline void Routers::QueueRoutingContext::shutdown()
-{
-    d_isShuttingDown = true;
-}
-
 // -----------------------------
 // struct Routers::Expression
 // -----------------------------

diff --git a/src/groups/mqb/mqbi/mqbi_queueengine.h b/src/groups/mqb/mqbi/mqbi_queueengine.h
@@ -75,9 +75,9 @@ class QueueEngine {
     virtual int configure(bsl::ostream& errorDescription) = 0;
 
     /// Reset the internal state of this engine.  If the optionally specified
-    /// 'keepConfirming' is 'true', keep the data structures for CONFIRMs
-    /// processing.
-    virtual void resetState(bool keepConfirming = false) = 0;
+    /// 'isShuttingDown' is 'true', clear the routing state but keep the Apps
+    /// state for CONFIRMs processing.
+    virtual void resetState(bool isShuttingDown = false) = 0;
 
     /// Rebuild the internal state of this engine.  This method is invoked
     /// when the queue this engine is associated with is created from an