learn schema while iterating event

Signed-off-by: dorjesinpo <[email protected]>

Author: dorjesinpo

Diff for: src/groups/bmq/bmqimp/bmqimp_queuemanager.cpp

@@ -314,12 +314,7 @@ void QueueManager::observePushEvent(Event*                          queueEvent,
 
     queueEvent->insertQueue(info.d_subscriptionId, queue);
 
-    bmqp::MessageProperties::SchemaPtr schema = schemaLearner().learn(
-        schemaLearner().createContext(queueId),
-        bmqp::MessagePropertiesInfo(info.d_header),
-        bdlbb::Blob());
-
-    queueEvent->addContext(correlationId, subscriptionHandleId, schema);
+    queueEvent->addContext(correlationId, subscriptionHandleId, info.d_schema);
 }
 
 int QueueManager::onPushEvent(QueueManager::EventInfos* eventInfos,
@@ -367,6 +362,33 @@ int QueueManager::onPushEvent(QueueManager::EventInfos* eventInfos,
             BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
             return rc * 10 + rc_OPTIONS_LOAD_ERROR;  // RETURN
         }
+
+        const bmqp::PushHeader&           header  = msgIterator.header();
+        int                               queueId = header.queueId();
+        const bmqp::MessagePropertiesInfo input(header);
+
+        bmqp::MessageProperties::SchemaPtr* schemaHolder =
+            schemaLearner().observe(d_schemaLearner.createContext(queueId),
+                                    input);
+
+        bmqp::MessageProperties::SchemaPtr schema;
+
+        if (schemaHolder) {
+            schema = *schemaHolder;
+            if (!schema) {
+                // Learn new Schema by reading all MessageProperties.
+                bdlbb::Blob appData(d_allocator_p);
+                if (msgIterator.loadApplicationData(&appData)) {
+                    bmqp::MessageProperties mps(d_allocator_p);
+
+                    if (mps.streamIn(appData, input.isExtended()) == 0) {
+                        // Learn new schema.
+                        *schemaHolder = schema = mps.makeSchema(d_allocator_p);
+                    }
+                }
+            }
+        }
+
         eventInfos->resize(eventInfos->size() + 1);
         if ((optionsView.find(bmqp::OptionType::e_SUB_QUEUE_INFOS) !=
              optionsView.end()) ||
@@ -388,10 +410,11 @@ int QueueManager::onPushEvent(QueueManager::EventInfos* eventInfos,
                 *hasMessageWithMultipleSubQueueIds = true;
             }
             else {
-                eventInfos->back().d_ids.push_back(bmqp::EventUtilQueueInfo(
-                    subQueueInfos[0].id(),
-                    msgIterator.header(),
-                    msgIterator.applicationDataSize()));
+                eventInfos->back().d_ids.push_back(
+                    bmqp::EventUtilQueueInfo(subQueueInfos[0].id(),
+                                             msgIterator.header(),
+                                             msgIterator.applicationDataSize(),
+                                             schema));
             }
 
             // Update message count
@@ -401,7 +424,8 @@ int QueueManager::onPushEvent(QueueManager::EventInfos* eventInfos,
             eventInfos->back().d_ids.push_back(bmqp::EventUtilQueueInfo(
                 bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID,
                 msgIterator.header(),
-                msgIterator.applicationDataSize()));
+                msgIterator.applicationDataSize(),
+                schema));
 
             // Update message count
             ++(*messageCount);

Diff for: src/groups/bmq/bmqp/bmqp_eventutil.cpp

@@ -115,6 +115,8 @@ class Flattener {
     OptionsView d_optionsView;
     // Helps go through Options
 
+    bmqp::SchemaLearner* d_schemaLearner;
+
     // PRIVATE TYPES
     typedef bdlma::LocalSequentialAllocator<
         32 * sizeof(Protocol::SubQueueIdsArrayOld::value_type)>
@@ -187,7 +189,8 @@ class Flattener {
               const Event&                     event,
               bdlbb::BlobBufferFactory*        bufferFactory,
               BlobSpPool*                      blobSpPool_p,
-              bslma::Allocator*                allocator);
+              bslma::Allocator*                allocator,
+              bmqp::SchemaLearner*             schemaLearner = 0);
 
     // MANIPULATORS
 
@@ -372,13 +375,38 @@ Flattener::packMesage(const Protocol::SubQueueInfosArray& subQInfo)
         return result;  // RETURN
     }
 
+    int                               queueId = header.queueId();
+    const bmqp::MessagePropertiesInfo input(header);
+
+    bmqp::MessageProperties::SchemaPtr schema;
+
+    if (d_schemaLearner) {
+        bmqp::MessageProperties::SchemaPtr* schemaHolder =
+            d_schemaLearner->observe(d_schemaLearner->createContext(queueId),
+                                     input);
+
+        if (schemaHolder) {
+            schema = *schemaHolder;
+            if (!schema) {
+                // Learn new Schema by reading all MessageProperties.
+                bmqp::MessageProperties mps(d_allocator_p);
+
+                if (mps.streamIn(d_appData, input.isExtended()) == 0) {
+                    // Learn new schema.
+                    *schemaHolder = schema = mps.makeSchema(d_allocator_p);
+                }
+            }
+        }
+    }
+
     // Successfully packed the current message and associated it with the
     // current SubQueueId being processed.  Add the corresponding queueId for
     // this message associated with the current flattened event.
     d_currEventInfo.d_ids.push_back(
         bmqp::EventUtilQueueInfo(subQInfo[0].id(),
                                  header,
-                                 d_msgIterator.applicationDataSize()));
+                                 d_msgIterator.applicationDataSize(),
+                                 schema));
 
     return result;
 }
@@ -404,14 +432,16 @@ Flattener::Flattener(bsl::vector<EventUtilEventInfo>* eventInfos,
                      const Event&                     event,
                      bdlbb::BlobBufferFactory*        bufferFactory,
                      BlobSpPool*                      blobSpPool_p,
-                     bslma::Allocator*                allocator)
+                     bslma::Allocator*                allocator,
+                     bmqp::SchemaLearner*             schemaLearner)
 : d_eventInfos_p(eventInfos)
 , d_allocator_p(allocator)
 , d_builder(blobSpPool_p, allocator)
 , d_msgIterator(bufferFactory, allocator)
 , d_currEventInfo(allocator)
 , d_appData(bufferFactory, allocator)
 , d_optionsView(allocator)
+, d_schemaLearner(schemaLearner)
 {
     event.loadPushMessageIterator(&d_msgIterator);
     BSLS_ASSERT_SAFE(d_msgIterator.isValid());

Diff for: src/groups/bmq/bmqp/bmqp_eventutil.h

@@ -33,6 +33,7 @@
 #include <bmqp_blobpoolutil.h>
 #include <bmqp_protocol.h>
 #include <bmqp_queueid.h>
+#include <bmqp_schemalearner.h>
 
 // BDE
 #include <bdlbb_blob.h>
@@ -65,9 +66,12 @@ struct EventUtilQueueInfo {
     const bmqp::PushHeader d_header;
     const int              d_applicationDataSize;
 
+    const bmqp::MessageProperties::SchemaPtr d_schema;
+
     EventUtilQueueInfo(unsigned int            subscriptionId,
                        const bmqp::PushHeader& header,
-                       int                     applicationDataSize);
+                       int                     applicationDataSize,
+                       const bmqp::MessageProperties::SchemaPtr schema);
 };
 // =========================
 // struct EventUtilEventInfo
@@ -146,12 +150,15 @@ struct EventUtil {
 // struct EventUtilQueueInfo
 // --------------------------
 
-inline EventUtilQueueInfo::EventUtilQueueInfo(unsigned int subscriptionId,
-                                              const bmqp::PushHeader& header,
-                                              int appDataSize)
+inline EventUtilQueueInfo::EventUtilQueueInfo(
+    unsigned int                             subscriptionId,
+    const bmqp::PushHeader&                  header,
+    int                                      appDataSize,
+    const bmqp::MessageProperties::SchemaPtr schema)
 : d_subscriptionId(subscriptionId)
 , d_header(header)
 , d_applicationDataSize(appDataSize)
+, d_schema(schema)
 {
     // NOTHING
 }

Diff for: src/groups/bmq/bmqp/bmqp_schemalearner.cpp

@@ -138,16 +138,14 @@ SchemaLearner::Context SchemaLearner::createContext(int foreignId)
     return context->second;
 }
 
-SchemaLearner::SchemaPtr
-SchemaLearner::learn(Context&                     context,
-                     const MessagePropertiesInfo& input,
-                     const bdlbb::Blob&           blob)
+SchemaLearner::SchemaPtr*
+SchemaLearner::observe(Context& context, const MessagePropertiesInfo& input)
 {
     const SchemaIdType inputId = input.schemaId();
 
     if (!isPresentAndValid(inputId)) {
         // Nothing to do
-        return SchemaPtr();  // RETURN
+        return 0;  // RETURN
     }
 
     BSLS_ASSERT_SAFE(context);
@@ -171,19 +169,7 @@ SchemaLearner::learn(Context&                     context,
         // Must destroy previously learned Schema
     }
 
-    const SchemaPtr& result = contextHandle->d_schema_sp;
-
-    if (!result) {
-        // Learn new Schema by reading all MessageProperties.
-        MessageProperties mps(d_allocator_p);
-        int               rc = mps.streamIn(blob, input, result);
-
-        if (rc == 0) {
-            // Learn new schema.
-            contextHandle->d_schema_sp = mps.makeSchema(d_allocator_p);
-        }
-    }
-    return result;
+    return &contextHandle->d_schema_sp;
 }
 
 MessagePropertiesInfo

Diff for: src/groups/bmq/bmqp/bmqp_schemalearner.h

@@ -215,11 +215,8 @@ class SchemaLearner {
 
     /// If the specified `input` indicates recycling, reset previously learned
     /// schema accumulated in the specified `context` and associated with the
-    /// id in `input`.  Return the previously learned schema if it is valid,
-    /// learn and return new schema otherwise.
-    SchemaPtr learn(Context&                     context,
-                    const MessagePropertiesInfo& input,
-                    const bdlbb::Blob&           blob);
+    /// id in `input`.  Return the address of the corresponding schema holder.
+    SchemaPtr* observe(Context& context, const MessagePropertiesInfo& input);
 
     /// Return `MessagePropertiesLogic` with a unique id associated with the
     /// specified `context` and the id in the specified `input`.  If there

Diff for: src/groups/bmq/bmqp/bmqp_schemalearner.t.cpp

@@ -209,7 +209,7 @@ static void test3_observingTest()
 
     bmqp::MessagePropertiesInfo recycledInput(true, 1, true);
 
-    theLearner.learn(server, recycledInput, bdlbb::Blob());
+    theLearner.observe(server, recycledInput);
 
     BMQTST_ASSERT_EQ(0, theLearner.read(server, &out2, input, blob));