Generating unique Subscription ids for each request

src/applications/bmqtool/m_bmqtool_inpututil.cpp

@@ -173,6 +173,8 @@ void InputUtil::verifyProperties(
 
             bsl::unordered_set<bsl::string> pairs;
 
+            pairs.insert("pairs_");
+
             while (it.hasNext()) {
                 bsl::string name = it.name();
 
@@ -201,7 +203,7 @@ void InputUtil::verifyProperties(
                     break;
                 }
                 case bmqt::PropertyType::e_SHORT: {
-                    BSLS_ASSERT_SAFE(it.getAsShort()() ==
+                    BSLS_ASSERT_SAFE(it.getAsShort() ==
                                      in.getPropertyAsShort(name));
                     break;
                 }

src/groups/bmq/bmqimp/bmqimp_brokersession.cpp

@@ -3531,14 +3531,20 @@ void BrokerSession::processPushEvent(const bmqp::Event& event)
              citer != sIds.end();
              ++citer) {
             bmqt::CorrelationId         correlationId;
+            unsigned int                subscriptionHandleId;
             const QueueManager::QueueSp queue =
-                d_queueManager.observePushEvent(&correlationId, *citer);
+                d_queueManager.observePushEvent(&correlationId,
+                                                &subscriptionHandleId,
+                                                *citer);
 
             BSLS_ASSERT(queue);
             queueEvent->insertQueue(citer->d_subscriptionId, queue);
 
-            queueEvent->addCorrelationId(correlationId,
-                                         citer->d_subscriptionId);
+            // Use 'subscriptionHandle' instead of the internal
+            // 'citer->d_subscriptionId' so that
+            // 'bmqimp::Event::subscriptionId()' returns 'subscriptionHandle'
+
+            queueEvent->addCorrelationId(correlationId, subscriptionHandleId);
         }
 
         // Update event bytes
@@ -5239,7 +5245,12 @@ BrokerSession::createConfigureQueueContext(const bsl::shared_ptr<Queue>& queue,
 
             bmqp_ctrlmsg::Subscription subscription(d_allocator_p);
 
-            subscription.sId() = cit->first.id();
+            const unsigned int internalSubscriptionId =
+                ++d_nextInternalSubscriptionId;
+
+            subscription.sId() = internalSubscriptionId;
+            // Using unique id instead of 'SubscriptionHandle::id()'
+
             subscription.consumers().emplace_back(ci);
 
             bmqp_ctrlmsg::ExpressionVersion::Value version;
@@ -5260,9 +5271,9 @@ BrokerSession::createConfigureQueueContext(const bsl::shared_ptr<Queue>& queue,
             subscription.expression().text()    = from.expression().text();
 
             streamParams.subscriptions().emplace_back(subscription);
-            d_queueManager.registerSubscription(queue,
-                                                cit->first.id(),
-                                                cit->first.correlationId());
+            queue->registerInternalSubscriptionId(internalSubscriptionId,
+                                                  cit->first.id(),
+                                                  cit->first.correlationId());
         }
         return context;  // RETURN
     }
@@ -5298,9 +5309,9 @@ BrokerSession::createConfigureQueueContext(const bsl::shared_ptr<Queue>& queue,
         streamParams.consumerPriority()      = options.consumerPriority();
         streamParams.consumerPriorityCount() = 1;
 
-        d_queueManager.registerSubscription(queue,
-                                            queue->subQueueId(),
-                                            bmqt::CorrelationId());
+        queue->registerInternalSubscriptionId(queue->subQueueId(),
+                                              queue->subQueueId(),
+                                              bmqt::CorrelationId());
     }
 
     return context;
@@ -5624,6 +5635,7 @@ BrokerSession::BrokerSession(
 , d_messageExpirationTimeoutHandle()
 , d_nextRequestGroupId(k_NON_BUFFERED_REQUEST_GROUP_ID)
 , d_queueRetransmissionTimeoutMap(allocator)
+, d_nextInternalSubscriptionId(bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID)
 {
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(d_scheduler_p->clockType() ==
@@ -6123,6 +6135,7 @@ void BrokerSession::onConfigureQueueResponse(
                              res == bmqt::GenericResult::e_NOT_CONNECTED ||
                              res == bmqt::GenericResult::e_NOT_SUPPORTED);
 
+            (void)res;
             BALL_LOG_INFO << "Ignore cancelled request: "
                           << context->request();
             return;  // RETURN

src/groups/bmq/bmqimp/bmqimp_brokersession.h

@@ -852,6 +852,9 @@ class BrokerSession BSLS_CPP11_FINAL {
     // retransmission timeout provided by
     // the broker
 
+    unsigned int d_nextInternalSubscriptionId;
+    // Assists generating unique ids for Configure requests.
+
   private:
     // NOT IMPLEMENTED
     BrokerSession(const BrokerSession&);

src/groups/bmq/bmqimp/bmqimp_event.h

@@ -471,7 +471,7 @@ class Event {
     /// undefined unless 0 <= 'position' < numCorrrelationIds(), and event's
     /// type() is MESSAGEEVENT, 'messageEventMode()' is READ and the
     /// underlying raw event is of type PUSH.
-    const unsigned int subscriptionId(int position) const;
+    unsigned int subscriptionId(int position) const;
 
     // MANIPULATORS
 
@@ -500,8 +500,8 @@ class Event {
     /// event's type() is MESSAGEEVENT, 'messageEventMode()' is READ and the
     /// underlying raw event is of type ACK, PUT or PUSH.
     void addCorrelationId(const bmqt::CorrelationId& correlationId,
-                          unsigned int               subscriptionId =
-                              bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID);
+                          unsigned int               subscriptionHandleId =
+                              bmqt::SubscriptionHandle::k_INVALID_HANDLE_ID);
 
     /// Insert the specified `queue` to the queues and the specified
     /// `corrId` to the list of correlationIds associated with this event.
@@ -746,7 +746,7 @@ inline const bmqt::CorrelationId& Event::correlationId(int position) const
     return d_correlationIds[position].first;
 }
 
-inline const unsigned int Event::subscriptionId(int position) const
+inline unsigned int Event::subscriptionId(int position) const
 {
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(type() == EventType::e_MESSAGE);
@@ -798,7 +798,7 @@ inline bmqp::PutEventBuilder* Event::putEventBuilder()
 }
 
 inline void Event::addCorrelationId(const bmqt::CorrelationId& correlationId,
-                                    unsigned int               subscriptionId)
+                                    unsigned int subscriptionHandleId)
 {
     // TODO: when ACK event is created locally we have to fill d_correlationIds
     //       before the raw ACK 'bmqp::Event' is created and may be used to
@@ -810,7 +810,8 @@ inline void Event::addCorrelationId(const bmqt::CorrelationId& correlationId,
     // BSLS_ASSERT_SAFE(messageEventMode() == MessageEventMode::e_READ);
     // BSLS_ASSERT_SAFE(d_rawEvent.isAckEvent());
 
-    d_correlationIds.push_back(bsl::make_pair(correlationId, subscriptionId));
+    d_correlationIds.push_back(
+        bsl::make_pair(correlationId, subscriptionHandleId));
 }
 
 }  // close package namespace

src/groups/bmq/bmqimp/bmqimp_messagedumper.cpp

@@ -341,13 +341,16 @@ void MessageDumper::dumpPushEvent(bsl::ostream& out, const bmqp::Event& event)
         unsigned int        subscriptionId;
         bmqp::RdaInfo       rdaInfo;
         bmqt::CorrelationId correlationId;
+        unsigned int        subscriptionHandleId;
 
         iter.extractQueueInfo(&qId, &subscriptionId, &rdaInfo);
 
         QueueManager::QueueSp queue =
-            d_queueManager_p->lookupQueueBySubscriptionId(&correlationId,
-                                                          qId,
-                                                          subscriptionId);
+            d_queueManager_p->lookupQueueBySubscriptionId(
+                &correlationId,
+                &subscriptionHandleId,
+                qId,
+                subscriptionId);
         BSLS_ASSERT_SAFE(queue);
 
         out << "PUSH Message #" << ++msgNum << ": "

src/groups/bmq/bmqimp/bmqimp_messagedumper.t.cpp

@@ -550,7 +550,9 @@ void Tester::registerSubscription(const bslstl::StringRef&   uri,
 
     BSLS_ASSERT_SAFE(queue);
 
-    d_queueManager.registerSubscription(queue, subscriptionId, correlationId);
+    queue->registerInternalSubscriptionId(subscriptionId,
+                                          subscriptionId,
+                                          correlationId);
 }
 
 void Tester::updateSubscriptions(const bslstl::StringRef&              uri,

src/groups/bmq/bmqimp/bmqimp_queue.cpp

@@ -325,6 +325,7 @@ Queue::Queue(bslma::Allocator* allocator)
 , d_schemaLearner(allocator)
 , d_schemaLearnerContext(d_schemaLearner.createContext())
 , d_config(allocator)
+, d_registeredInternalSubscriptionIds(allocator)
 {
     d_handleParameters.uri()   = "";
     d_handleParameters.flags() = bmqt::QueueFlagsUtil::empty();

src/groups/bmq/bmqimp/bmqimp_queue.h

@@ -172,6 +172,11 @@ struct QueueStatsUtil {
 
 /// Representation of a Queue (properties, stats, state, ...)
 class Queue {
+  public:
+    // PUBLIC TYPES
+    typedef bsl::pair<unsigned int, bmqt::CorrelationId> SubscriptionHandle;
+    // Not using private 'bmqt::SubscriptionHandle' ctor
+
   private:
     // DATA
     bslma::Allocator* d_allocator_p;
@@ -256,6 +261,13 @@ class Queue {
 
     bmqp_ctrlmsg::StreamParameters d_config;
 
+    bsl::unordered_map<unsigned int, SubscriptionHandle>
+        d_registeredInternalSubscriptionIds;
+    // This keeps SubscriptionHandle (id and CorrelationId) for Configure
+    // response processing.
+    // Supporting multiple concurrent Configure requests.
+    // TODO: This should go into ConfigureRequest context.
+
   private:
     // NOT IMPLEMENTED
 
@@ -325,6 +337,20 @@ class Queue {
     /// reinitialize the state before a new start).
     void clearStatContext();
 
+    void
+    registerInternalSubscriptionId(unsigned int internalSubscriptionId,
+                                   unsigned int subscriptionHandleId,
+                                   const bmqt::CorrelationId& correlationId);
+    // Keep the specified 'subscriptionHandleId' and 'correlationId'
+    // associated with the specified 'internalSubscriptionId' between
+    // Configure request and Configure response (until
+    // 'extractSubscriptionHandle').
+
+    SubscriptionHandle
+    extractSubscriptionHandle(unsigned int internalSubscriptionId);
+    // Lookup, copy, erase, and return the copy of what was registered
+    // by 'registerInternalSubscriptionId'.
+
     // ACCESSORS
 
     /// Return true if this Queue object has a SubQueueId having the default
@@ -528,6 +554,33 @@ inline Queue& Queue::setConfig(const bmqp_ctrlmsg::StreamParameters& value)
     return *this;
 }
 
+inline void
+Queue::registerInternalSubscriptionId(unsigned int internalSubscriptionId,
+                                      unsigned int subscriptionHandleId,
+                                      const bmqt::CorrelationId& correlationId)
+{
+    d_registeredInternalSubscriptionIds.emplace(
+        internalSubscriptionId,
+        SubscriptionHandle(subscriptionHandleId, correlationId));
+}
+
+inline Queue::SubscriptionHandle
+Queue::extractSubscriptionHandle(unsigned int internalSubscriptionId)
+{
+    bsl::unordered_map<unsigned int, SubscriptionHandle>::const_iterator cit =
+        d_registeredInternalSubscriptionIds.find(internalSubscriptionId);
+
+    if (cit == d_registeredInternalSubscriptionIds.end()) {
+        return {internalSubscriptionId, bmqt::CorrelationId()};  // RETURN
+    }
+
+    SubscriptionHandle result(cit->second);
+
+    d_registeredInternalSubscriptionIds.erase(cit);
+
+    return result;
+}
+
 // ACCESSORS
 inline QueueState::Enum Queue::state() const
 {