Refactoring App registration

src/groups/mqb/mqbblp/mqbblp_cluster.cpp

@@ -2848,6 +2848,7 @@ void Cluster::onDomainReconfigured(const mqbi::Domain&     domain,
                                                   oldCfgAppIds,
                                                   newCfgAppIds);
 
+    // TODO: This should be one call - one QueueUpdateAdvisory for all Apps
     bsl::unordered_set<bsl::string>::const_iterator it = addedIds.cbegin();
     for (; it != addedIds.cend(); ++it) {
         dispatcher()->execute(

src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp

@@ -139,29 +139,27 @@ void createQueueUriKey(bmqt::Uri*        out,
 }
 
 void afterAppIdRegisteredDispatched(
-    mqbi::Queue*                                queue,
-    const mqbc::ClusterStateQueueInfo::AppInfo& appIdInfo)
+    mqbi::Queue*                                 queue,
+    const mqbc::ClusterStateQueueInfo::AppInfos& appInfos)
 {
     // executed by the *QUEUE DISPATCHER* thread
 
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(queue->dispatcher()->inDispatcherThread(queue));
 
-    queue->queueEngine()->afterAppIdRegistered(
-        mqbi::Storage::AppInfo(appIdInfo.first, appIdInfo.second));
+    queue->queueEngine()->afterAppIdRegistered(appInfos);
 }
 
 void afterAppIdUnregisteredDispatched(
-    mqbi::Queue*                                queue,
-    const mqbc::ClusterStateQueueInfo::AppInfo& appIdInfo)
+    mqbi::Queue*                                 queue,
+    const mqbc::ClusterStateQueueInfo::AppInfos& appInfos)
 {
     // executed by the *QUEUE DISPATCHER* thread
 
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(queue->dispatcher()->inDispatcherThread(queue));
 
-    queue->queueEngine()->afterAppIdUnregistered(
-        mqbi::Storage::AppInfo(appIdInfo.first, appIdInfo.second));
+    queue->queueEngine()->afterAppIdUnregistered(appInfos);
 }
 
 void handleHolderDummy(const bsl::shared_ptr<mqbi::QueueHandle>& handle)
@@ -4375,10 +4373,6 @@ void ClusterQueueHelper::onQueueUpdated(const bmqt::Uri&   uri,
         d_cluster_p->dispatcher()->inDispatcherThread(d_cluster_p));
     BSLS_ASSERT_SAFE(!d_cluster_p->isRemote());
 
-    if (!d_cluster_p->isCSLModeEnabled()) {
-        return;  // RETURN
-    }
-
     if (!uri.isValid()) {
         // This is an appID update for the entire domain, instead of any
         // individual queue. Nothing to do for the queue helper.
@@ -4394,53 +4388,51 @@ void ClusterQueueHelper::onQueueUpdated(const bmqt::Uri&   uri,
     const int      partitionId = qiter->second->partitionId();
     BSLS_ASSERT_SAFE(partitionId != mqbs::DataStore::k_INVALID_PARTITION_ID);
 
-    for (AppInfosCIter cit = addedAppIds.cbegin(); cit != addedAppIds.cend();
-         ++cit) {
+    if (d_cluster_p->isCSLModeEnabled()) {
         if (!d_clusterState_p->isSelfPrimary(partitionId) || queue == 0) {
             // Note: In non-CSL mode, the queue creation callback is
             // invoked at replica nodes when they receive a queue creation
             // record from the primary in the partition stream.
 
-            mqbi::Storage::AppInfos one(1, d_allocator_p);
-            one.emplace(*cit);
-
             d_storageManager_p->updateQueueReplica(
                 partitionId,
                 uri,
                 qiter->second->key(),
-                one,
+                addedAppIds,
                 d_clusterState_p->domainStates()
                     .at(uri.qualifiedDomain())
                     ->domain());
         }
-        if (queue) {
-            d_cluster_p->dispatcher()->execute(
-                bdlf::BindUtil::bind(afterAppIdRegisteredDispatched,
-                                     queue,
-                                     *cit),
-                queue);
+
+        for (AppInfosCIter cit = removedAppIds.cbegin();
+             cit != removedAppIds.cend();
+             ++cit) {
+            if (!d_clusterState_p->isSelfPrimary(partitionId) || queue == 0) {
+                // Note: In non-CSL mode, the queue deletion callback is
+                // invoked at replica nodes when they receive a queue deletion
+                // record from the primary in the partition stream.
+                d_storageManager_p->unregisterQueueReplica(
+                    partitionId,
+                    uri,
+                    qiter->second->key(),
+                    cit->second);
+            }
         }
     }
 
-    for (AppInfosCIter cit = removedAppIds.cbegin();
-         cit != removedAppIds.cend();
-         ++cit) {
-        if (!d_clusterState_p->isSelfPrimary(partitionId) || queue == 0) {
-            // Note: In non-CSL mode, the queue deletion callback is
-            // invoked at replica nodes when they receive a queue deletion
-            // record from the primary in the partition stream.
-            d_storageManager_p->unregisterQueueReplica(partitionId,
-                                                       uri,
-                                                       qiter->second->key(),
-                                                       cit->second);
-        }
-        if (queue) {
-            d_cluster_p->dispatcher()->execute(
-                bdlf::BindUtil::bind(afterAppIdUnregisteredDispatched,
-                                     queue,
-                                     *cit),
-                queue);
-        }
+    if (queue) {
+        // TODO: replace with one call
+        d_cluster_p->dispatcher()->execute(
+            bdlf::BindUtil::bind(afterAppIdRegisteredDispatched,
+                                 queue,
+                                 addedAppIds),
+            queue);
+
+        d_cluster_p->dispatcher()->execute(
+            bdlf::BindUtil::bind(afterAppIdUnregisteredDispatched,
+                                 queue,
+                                 removedAppIds),
+            queue);
     }
 
     bmqu::Printer<AppInfos> printer1(&addedAppIds);

src/groups/mqb/mqbblp/mqbblp_domain.cpp

@@ -68,32 +68,6 @@ void queueHolderDummy(const bsl::shared_ptr<mqbi::Queue>& queue)
     BALL_LOG_INFO << "Deleted queue '" << queue->uri().canonical() << "'";
 }
 
-void afterAppIdRegisteredDispatched(mqbi::Queue*       queue,
-                                    const bsl::string& appId)
-{
-    // executed by the *QUEUE DISPATCHER* thread
-
-    // PRECONDITIONS
-    BSLS_ASSERT_SAFE(queue->dispatcher()->inDispatcherThread(queue));
-
-    queue->queueEngine()->afterAppIdRegistered(
-        mqbi::Storage::AppInfo(appId, mqbu::StorageKey()));
-}
-
-void afterAppIdUnregisteredDispatched(mqbi::Queue*       queue,
-                                      const bsl::string& appId)
-{
-    // executed by the *QUEUE DISPATCHER* thread
-
-    // PRECONDITIONS
-    BSLS_ASSERT_SAFE(queue->dispatcher()->inDispatcherThread(queue));
-
-    // Note: Inputing nullKey here is okay since this routine will be removed
-    //       when we switch to CSL workflow.
-    queue->queueEngine()->afterAppIdUnregistered(
-        mqbi::Storage::AppInfo(appId, mqbu::StorageKey()));
-}
-
 /// Validates an application subscription.
 bool validdateSubscriptionExpression(bsl::ostream& errorDescription,
                                      const mqbconfm::Expression& expression,
@@ -463,51 +437,6 @@ int Domain::configure(bsl::ostream&           errorDescription,
         BSLS_ASSERT_OPT(oldConfig.has_value());
         BSLS_ASSERT_OPT(d_config.has_value());
 
-        // In non-CSL mode, manually dispatch AppId registration callbacks.
-        if (!d_cluster_sp->isCSLModeEnabled() &&
-            d_config.value().mode().isFanoutValue()) {
-            // Compute list of added and removed App IDs.
-            bsl::unordered_set<bsl::string> oldCfgAppIds(
-                oldConfig.value().mode().fanout().appIDs().cbegin(),
-                oldConfig.value().mode().fanout().appIDs().cend(),
-                d_allocator_p);
-            bsl::unordered_set<bsl::string> newCfgAppIds(
-                d_config.value().mode().fanout().appIDs().cbegin(),
-                d_config.value().mode().fanout().appIDs().cend(),
-                d_allocator_p);
-
-            bsl::unordered_set<bsl::string> addedIds, removedIds;
-            mqbc::StorageUtil::loadAddedAndRemovedEntries(&addedIds,
-                                                          &removedIds,
-                                                          oldCfgAppIds,
-                                                          newCfgAppIds);
-
-            bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);
-
-            // Invoke callbacks for each added and removed ID on each queue.
-            bsl::unordered_set<bsl::string>::const_iterator it =
-                addedIds.cbegin();
-            QueueMap::const_iterator qIt;
-            for (; it != addedIds.cend(); it++) {
-                for (qIt = d_queues.cbegin(); qIt != d_queues.cend(); ++qIt) {
-                    d_dispatcher_p->execute(
-                        bdlf::BindUtil::bind(afterAppIdRegisteredDispatched,
-                                             qIt->second.get(),
-                                             *it),
-                        qIt->second.get());
-                }
-            }
-            for (it = removedIds.cbegin(); it != removedIds.cend(); ++it) {
-                for (qIt = d_queues.cbegin(); qIt != d_queues.cend(); ++qIt) {
-                    d_dispatcher_p->execute(
-                        bdlf::BindUtil::bind(afterAppIdUnregisteredDispatched,
-                                             qIt->second.get(),
-                                             *it),
-                        qIt->second.get());
-                }
-            }
-        }
-
         // Notify the 'cluster' of the updated configuration, so it can write
         // any needed update-advisories to the CSL.
         d_cluster_sp->onDomainReconfigured(*this,

src/groups/mqb/mqbblp/mqbblp_localqueue.cpp

@@ -161,7 +161,7 @@ int LocalQueue::configure(bsl::ostream& errorDescription, bool isReconfigure)
                                 d_allocator_p);
     }
 
-    rc = d_queueEngine_mp->configure(errorDescription);
+    rc = d_queueEngine_mp->configure(errorDescription, isReconfigure);
     if (rc != 0) {
         return 10 * rc + rc_QUEUE_ENGINE_CFG_FAILURE;  // RETURN
     }

src/groups/mqb/mqbblp/mqbblp_queue.cpp

@@ -343,7 +343,7 @@ void Queue::convertToLocalDispatched()
 
     d_state.setId(bmqp::QueueId::k_PRIMARY_QUEUE_ID);
     createLocal();
-    rc = d_localQueue_mp->configure(errorDescription, true);
+    rc = d_localQueue_mp->configure(errorDescription, false);
     if (rc != 0) {
         BALL_LOG_ERROR
             << "#QUEUE_CONVERTION_FAILURE " << d_state.uri()
@@ -483,7 +483,6 @@ Queue::Queue(const bmqt::Uri&                          uri,
     //      storage.
 
     d_state.setStorageManager(storageManager)
-        .setAppKeyGenerator(storageManager)
         .setMiscWorkThreadPool(threadPool)
         .setRoutingConfig(routingCfg)
         .setMessageThrottleConfig(messageThrottleConfig);

src/groups/mqb/mqbblp/mqbblp_queueconsumptionmonitor.cpp

@@ -170,30 +170,26 @@ QueueConsumptionMonitor::setMaxIdleTime(bsls::Types::Int64 value)
     return *this;
 }
 
-void QueueConsumptionMonitor::registerSubStream(const mqbu::StorageKey& key)
+void QueueConsumptionMonitor::registerSubStream(const bsl::string& id)
 {
     // Should always be called from the queue thread, but will be invoked from
     // the cluster thread once upon queue creation.
 
     // PRECONDITIONS
-    BSLS_ASSERT_SAFE(key != mqbu::StorageKey::k_NULL_KEY ||
-                     d_subStreamInfos.empty());
-    BSLS_ASSERT_SAFE(d_subStreamInfos.find(mqbu::StorageKey::k_NULL_KEY) ==
-                     d_subStreamInfos.end());
-    BSLS_ASSERT_SAFE(d_subStreamInfos.find(key) == d_subStreamInfos.end());
+    BSLS_ASSERT_SAFE(d_subStreamInfos.find(id) == d_subStreamInfos.end());
 
-    d_subStreamInfos.insert(bsl::make_pair(key, SubStreamInfo()));
+    d_subStreamInfos.insert(bsl::make_pair(id, SubStreamInfo()));
 }
 
-void QueueConsumptionMonitor::unregisterSubStream(const mqbu::StorageKey& key)
+void QueueConsumptionMonitor::unregisterSubStream(const bsl::string& id)
 {
     // executed by the *QUEUE DISPATCHER* thread
 
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(d_queueState_p->queue()->dispatcher()->inDispatcherThread(
         d_queueState_p->queue()));
 
-    SubStreamInfoMapConstIter iter = d_subStreamInfos.find(key);
+    SubStreamInfoMapConstIter iter = d_subStreamInfos.find(id);
     BSLS_ASSERT_SAFE(iter != d_subStreamInfos.end());
     d_subStreamInfos.erase(iter);
 }
@@ -231,7 +227,7 @@ void QueueConsumptionMonitor::onTimer(bsls::Types::Int64 currentTimer)
          iter != last;
          ++iter) {
         SubStreamInfo&          info   = iter->second;
-        const mqbu::StorageKey& appKey = iter->first;
+        const bsl::string&      id     = iter->first;
         if (info.d_messageSent) {
             // Queue is 'alive' because at least one message was sent
             // since the last 'timer'.
@@ -241,7 +237,7 @@ void QueueConsumptionMonitor::onTimer(bsls::Types::Int64 currentTimer)
 
             if (info.d_state == State::e_IDLE) {
                 // object was in idle state
-                onTransitionToAlive(&info, appKey);
+                onTransitionToAlive(&info, id);
                 continue;  // CONTINUE
             }
 
@@ -253,7 +249,7 @@ void QueueConsumptionMonitor::onTimer(bsls::Types::Int64 currentTimer)
             // No delivered messages in the last 'maxIdleTime'.
 
             // Call callback to log alarm if there are undelivered messages.
-            const bool haveUndelivered = d_loggingCb(appKey,
+            const bool haveUndelivered = d_loggingCb(id,
                                                      info.d_state ==
                                                          State::e_ALIVE);
 
@@ -269,16 +265,15 @@ void QueueConsumptionMonitor::onTimer(bsls::Types::Int64 currentTimer)
                 // so transition to alive.
                 if (info.d_state == State::e_IDLE) {
                     info.d_lastKnownGoodTimer = d_currentTimer;
-                    onTransitionToAlive(&info, appKey);
+                    onTransitionToAlive(&info, id);
                 }
             }
         }
     }
 }
 
-void QueueConsumptionMonitor::onTransitionToAlive(
-    SubStreamInfo*          subStreamInfo,
-    const mqbu::StorageKey& appKey)
+void QueueConsumptionMonitor::onTransitionToAlive(SubStreamInfo* subStreamInfo,
+                                                  const bsl::string& id)
 {
     // executed by the *QUEUE DISPATCHER* thread
 
@@ -291,12 +286,7 @@ void QueueConsumptionMonitor::onTransitionToAlive(
     bdlma::LocalSequentialAllocator<2048> localAllocator(0);
 
     bmqt::UriBuilder uriBuilder(d_queueState_p->uri(), &localAllocator);
-    bsl::string      appId;
-
-    if (!appKey.isNull() &&
-        d_queueState_p->storage()->hasVirtualStorage(appKey, &appId)) {
-        uriBuilder.setId(appId);
-    }
+    uriBuilder.setId(id);
 
     bmqt::Uri uri(&localAllocator);
     uriBuilder.uri(&uri);