FIX[BMQ]: send nego message before creating session (#886)

Signed-off-by: Emelia Lei <wlei29@bloomberg.net>

Author: emelialei88

src/groups/mqb/mqba/mqba_sessionnegotiator.cpp

@@ -458,32 +458,36 @@ SessionNegotiator::onClientIdentityMessage(bsl::ostream& errorDescription,
                            shouldExtendMessageProperties);
     }
 
-    // Create the session.  That also calculates 'maxMissedHeartbeats'
-    bsl::string description;
-    loadSessionDescription(
-        &description,
-        clientIdentity,
-        *(context->d_initialConnectionContext_p->channel().get()));
-
-    createSession(errorDescription, &session, context, description);
+    // Populate the negotiation context based on the received client identity.
+    int rc = populateNegotiationContext(errorDescription, context);
+    if (rc != 0) {
+        return session;  // RETURN
+    }
 
     // Communicate heartbeat settings.  Currently, only for SDK use
     const mqbcfg::NetworkInterfaces& niConfig = appConfig.networkInterfaces();
-
-    response.maxMissedHeartbeats() = context->d_maxMissedHeartbeat;
-
+    response.maxMissedHeartbeats()            = context->d_maxMissedHeartbeat;
     if (niConfig.tcpInterface().has_value()) {
         response.heartbeatIntervalMs() =
             niConfig.tcpInterface().value().heartbeatIntervalMs();
     }
 
-    int rc = sendNegotiationMessage(errorDescription,
-                                    negotiationResponse,
-                                    context);
+    rc = sendNegotiationMessage(errorDescription,
+                                negotiationResponse,
+                                context);
     if (rc != 0) {
-        session.reset();
+        return session;  // RETURN
     }
 
+    // Create the session.
+    bsl::string description;
+    loadSessionDescription(
+        &description,
+        clientIdentity,
+        *(context->d_initialConnectionContext_p->channel().get()));
+
+    createSession(&session, context, description);
+
     return session;
 }
 
@@ -522,7 +526,11 @@ SessionNegotiator::onBrokerResponseMessage(bsl::ostream& errorDescription,
         brokerResponse.brokerIdentity(),
         *(context->d_initialConnectionContext_p->channel().get()));
 
-    createSession(errorDescription, &session, context, description);
+    const int rc = populateNegotiationContext(errorDescription, context);
+    if (rc != 0) {
+        return session;  // RETURN
+    }
+    createSession(&session, context, description);
 
     return session;
 }
@@ -592,16 +600,26 @@ int SessionNegotiator::sendNegotiationMessage(
     return rc_SUCCESS;
 }
 
-void SessionNegotiator::createSession(bsl::ostream& errorDescription,
-                                      bsl::shared_ptr<mqbnet::Session>* out,
-                                      const NegotiationContextSp& context,
-                                      const bsl::string&          description)
+int SessionNegotiator::populateNegotiationContext(
+    bsl::ostream&               errorDescription,
+    const NegotiationContextSp& context)
 {
+    enum RcEnum {
+        // Value for the various RC error categories
+        rc_SUCCESS                 = 0,
+        rc_GET_CLUSTER_NODE_FAILED = -1
+    };
+
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(context->d_connectionType !=
                      mqbnet::ConnectionType::e_UNKNOWN);
     BSLS_ASSERT_SAFE(!context->d_negotiationMessage.isUndefinedValue());
 
+    if (context->d_connectionType == mqbnet::ConnectionType::e_ADMIN) {
+        // Nothing to do for admin connection
+        return rc_SUCCESS;  // RETURN
+    }
+
     const bmqp_ctrlmsg::NegotiationMessage& negoMsg =
         context->d_negotiationMessage;
     const bmqp_ctrlmsg::ClientIdentity& peerIdentity =
@@ -612,41 +630,7 @@ void SessionNegotiator::createSession(bsl::ostream& errorDescription,
     const mqbcfg::NetworkInterfaces& niConfig = brkrCfg.networkInterfaces();
     int                              maxMissedHeartbeats = 0;
 
-    if (context->d_connectionType == mqbnet::ConnectionType::e_ADMIN) {
-        mqba::AdminSession* session = new (*d_allocator_p)
-            AdminSession(context->d_initialConnectionContext_p->channel(),
-                         negoMsg,
-                         description,
-                         d_dispatcher_p,
-                         d_blobSpPool_p,
-                         d_scheduler_p,
-                         d_adminCb,
-                         d_allocator_p);
-
-        out->reset(session, d_allocator_p);
-    }
-    else if (context->d_connectionType == mqbnet::ConnectionType::e_CLIENT) {
-        // Create a dedicated stats subcontext for this client
-        bmqst::StatContextConfiguration statContextCfg(description);
-        statContextCfg.storeExpiredSubcontextValues(true);
-        bslma::ManagedPtr<bmqst::StatContext> statContext =
-            d_statContext_p->addSubcontext(statContextCfg);
-
-        mqba::ClientSession* session = new (*d_allocator_p)
-            ClientSession(context->d_initialConnectionContext_p->channel(),
-                          negoMsg,
-                          description,
-                          d_dispatcher_p,
-                          d_clusterCatalog_p,
-                          d_domainFactory_p,
-                          statContext,
-                          d_blobSpPool_p,
-                          d_bufferFactory_p,
-                          d_scheduler_p,
-                          d_allocator_p);
-
-        out->reset(session, d_allocator_p);
-
+    if (context->d_connectionType == mqbnet::ConnectionType::e_CLIENT) {
         // Configure heartbeat
         if (negoMsg.clientIdentity().clientType() ==
             bmqp_ctrlmsg::ClientType::E_TCPCLIENT) {
@@ -683,17 +667,9 @@ void SessionNegotiator::createSession(bsl::ostream& errorDescription,
             peerIdentity.clusterNodeId());
 
         if (!clusterNode) {
-            return;  // RETURN
+            return rc_GET_CLUSTER_NODE_FAILED;  // RETURN
         }
 
-        out->reset(new (*d_allocator_p) mqbnet::DummySession(
-                       context->d_initialConnectionContext_p->channel(),
-                       negoMsg,
-                       clusterNode,
-                       description,
-                       d_allocator_p),
-                   d_allocator_p);
-
         // Configure heartbeat
         if (clusterNode->cluster()->selfNodeId() ==
             mqbnet::Cluster::k_INVALID_NODE_ID) {
@@ -705,6 +681,75 @@ void SessionNegotiator::createSession(bsl::ostream& errorDescription,
     }
 
     context->d_maxMissedHeartbeat = maxMissedHeartbeats;
+
+    return rc_SUCCESS;
+}
+
+void SessionNegotiator::createSession(bsl::shared_ptr<mqbnet::Session>* out,
+                                      const NegotiationContextSp& context,
+                                      const bsl::string&          description)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(context->d_connectionType !=
+                     mqbnet::ConnectionType::e_UNKNOWN);
+    BSLS_ASSERT_SAFE(!context->d_negotiationMessage.isUndefinedValue());
+
+    const bmqp_ctrlmsg::NegotiationMessage& negoMsg =
+        context->d_negotiationMessage;
+    const bsl::shared_ptr<bmqio::Channel>& channel =
+        context->d_initialConnectionContext_p->channel();
+
+    if (context->d_connectionType == mqbnet::ConnectionType::e_ADMIN) {
+        mqba::AdminSession* session = new (*d_allocator_p)
+            AdminSession(channel,
+                         negoMsg,
+                         description,
+                         d_dispatcher_p,
+                         d_blobSpPool_p,
+                         d_scheduler_p,
+                         d_adminCb,
+                         d_allocator_p);
+
+        out->reset(session, d_allocator_p);
+    }
+    else if (context->d_connectionType == mqbnet::ConnectionType::e_CLIENT) {
+        // Create a dedicated stats subcontext for this client
+        bmqst::StatContextConfiguration statContextCfg(description);
+        statContextCfg.storeExpiredSubcontextValues(true);
+        bslma::ManagedPtr<bmqst::StatContext> statContext =
+            d_statContext_p->addSubcontext(statContextCfg);
+
+        mqba::ClientSession* session = new (*d_allocator_p)
+            ClientSession(channel,
+                          negoMsg,
+                          description,
+                          d_dispatcher_p,
+                          d_clusterCatalog_p,
+                          d_domainFactory_p,
+                          statContext,
+                          d_blobSpPool_p,
+                          d_bufferFactory_p,
+                          d_scheduler_p,
+                          d_allocator_p);
+
+        out->reset(session, d_allocator_p);
+    }
+    else {
+        const bmqp_ctrlmsg::ClientIdentity& peerIdentity =
+            negoMsg.isClientIdentityValue()
+                ? negoMsg.clientIdentity()
+                : negoMsg.brokerResponse().brokerIdentity();
+
+        mqbnet::ClusterNode* clusterNode = context->d_cluster_p->lookupNode(
+            peerIdentity.clusterNodeId());
+
+        out->reset(new (*d_allocator_p) mqbnet::DummySession(channel,
+                                                             negoMsg,
+                                                             clusterNode,
+                                                             description,
+                                                             d_allocator_p),
+                   d_allocator_p);
+    }
 }
 
 bool SessionNegotiator::checkIsDeprecatedSdkVersion(

src/groups/mqb/mqba/mqba_sessionnegotiator.h

@@ -160,12 +160,18 @@ class SessionNegotiator : public mqbnet::Negotiator {
                                const bmqp_ctrlmsg::NegotiationMessage& message,
                                const NegotiationContextSp& context);
 
+    /// Populate some other fields in `context` given the specified
+    /// ConnectionType and NegotiationMessage.  Return 0 on success, or
+    /// return a non-zero code on error and populate the specified
+    /// `errorDescription` with a description of the error.
+    int populateNegotiationContext(bsl::ostream& errorDescription,
+                                   const NegotiationContextSp& context);
+
     /// Load into the specified `out` a new session created using the
     /// specified `context` and `description`; or leave `out` untouched and
     /// populate the specified `errorDescription` with a description of the
     /// error in case of failure.
-    void createSession(bsl::ostream&                     errorDescription,
-                       bsl::shared_ptr<mqbnet::Session>* out,
+    void createSession(bsl::shared_ptr<mqbnet::Session>* out,
                        const NegotiationContextSp&       context,
                        const bsl::string&                description);