Perf[MQB]: use concrete dispatcher event types

Signed-off-by: Evgeny Malygin <emalygin@bloomberg.net>

Author: 678098

src/groups/mqb/CMakeLists.txt

@@ -20,7 +20,7 @@ set_property(SOURCE "mqba/mqba_application.cpp"
              APPEND
              PROPERTY COMPILE_DEFINITIONS "BMQ_BUILD_TYPE=${CMAKE_BUILD_TYPE}")
 
-set(MQB_PRIVATE_PACKAGES mqba mqbblp mqbc mqbcmd mqbconfm mqbi mqbmock mqbnet mqbs mqbsi mqbsl mqbu)
+set(MQB_PRIVATE_PACKAGES mqba mqbblp mqbc mqbcmd mqbconfm mqbevt mqbi mqbmock mqbnet mqbs mqbsi mqbsl mqbu)
 target_bmq_style_uor( mqb PRIVATE_PACKAGES ${MQB_PRIVATE_PACKAGES})
 
 # Turn off warnings in generated code.

src/groups/mqb/group/mqb.mem

@@ -5,6 +5,7 @@ mqbc
 mqbcfg
 mqbcmd
 mqbconfm
+mqbevt
 mqbi
 mqbmock
 mqbnet

src/groups/mqb/mqba/mqba_adminsession.cpp

@@ -72,6 +72,9 @@
 #include <bsls_performancehint.h>
 #include <bsls_timeinterval.h>
 
+// MQB
+#include <mqbevt_callbackevent.h>
+
 namespace BloombergLP {
 namespace mqba {
 
@@ -443,8 +446,8 @@ void AdminSession::onDispatcherEvent(const mqbi::DispatcherEvent& event)
 
     switch (event.type()) {
     case mqbi::DispatcherEventType::e_CALLBACK: {
-        const mqbi::DispatcherCallbackEvent* realEvent =
-            event.asCallbackEvent();
+        const mqbevt::CallbackEvent* const realEvent =
+            event.the<mqbevt::CallbackEvent>();
 
         BSLS_ASSERT_SAFE(!realEvent->callback().empty());
         realEvent->callback()();

src/groups/mqb/mqba/mqba_clientsession.cpp

@@ -151,6 +151,12 @@
 #include <mqbblp_clustercatalog.h>
 #include <mqbblp_queueengineutil.h>
 #include <mqbcfg_brokerconfig.h>
+#include <mqbevt_ackevent.h>
+#include <mqbevt_callbackevent.h>
+#include <mqbevt_confirmevent.h>
+#include <mqbevt_pushevent.h>
+#include <mqbevt_putevent.h>
+#include <mqbevt_rejectevent.h>
 #include <mqbi_cluster.h>
 #include <mqbi_queue.h>
 #include <mqbnet_tcpsessionfactory.h>
@@ -1376,7 +1382,7 @@ void ClientSession::processConfigureStream(
             opLogger));
 }
 
-void ClientSession::onAckEvent(const mqbi::DispatcherAckEvent& event)
+void ClientSession::onAckEvent(const mqbevt::AckEvent& event)
 {
     // executed by the *CLIENT* dispatcher thread
 
@@ -1472,7 +1478,7 @@ void ClientSession::onAckEvent(const mqbi::DispatcherAckEvent& event)
             "onAckEvent");
 }
 
-void ClientSession::onConfirmEvent(const mqbi::DispatcherConfirmEvent& event)
+void ClientSession::onConfirmEvent(const mqbevt::ConfirmEvent& event)
 {
     // executed by the *CLIENT* dispatcher thread
 
@@ -1556,7 +1562,7 @@ void ClientSession::onConfirmEvent(const mqbi::DispatcherConfirmEvent& event)
     }
 }
 
-void ClientSession::onRejectEvent(const mqbi::DispatcherRejectEvent& event)
+void ClientSession::onRejectEvent(const mqbevt::RejectEvent& event)
 {
     // executed by the *CLIENT* dispatcher thread
 
@@ -1720,7 +1726,7 @@ bool ClientSession::validateMessage(mqbi::QueueHandle**   queueHandle,
     return true;
 }
 
-void ClientSession::onPushEvent(const mqbi::DispatcherPushEvent& event)
+void ClientSession::onPushEvent(const mqbevt::PushEvent& event)
 {
     // executed by the *CLIENT* dispatcher thread
 
@@ -1917,7 +1923,7 @@ void ClientSession::onPushEvent(const mqbi::DispatcherPushEvent& event)
     }
 }
 
-void ClientSession::onPutEvent(const mqbi::DispatcherPutEvent& event)
+void ClientSession::onPutEvent(const mqbevt::PutEvent& event)
 {
     // executed by the *CLIENT* dispatcher thread
 
@@ -2641,33 +2647,45 @@ void ClientSession::processEvent(const bmqp::Event& event,
         }
 
         // Not a control or leader message, it's either a put or a confirm ..
-        mqbi::DispatcherEventType::Enum eventType;
 
+        bsl::shared_ptr<bdlbb::Blob> blobSp =
+            d_state.d_blobSpPool_p->getObject();
+        *blobSp = *(event.blob());
+
+        // Dispatch the event
+        // TODO(678098): revisit, use per-IO thread event source
         if (event.isPutEvent()) {
-            eventType = mqbi::DispatcherEventType::e_PUT;
+            bsl::shared_ptr<mqbevt::PutEvent> event_sp =
+                dispatcher()
+                    ->getDefaultEventSource()
+                    ->getEvent<mqbevt::PutEvent>();
+            event_sp->setBlob(blobSp).setSource(this);
+            dispatcher()->dispatchEvent(bslmf::MovableRefUtil::move(event_sp),
+                                        this);
         }
         else if (event.isConfirmEvent()) {
-            eventType = mqbi::DispatcherEventType::e_CONFIRM;
+            bsl::shared_ptr<mqbevt::ConfirmEvent> event_sp =
+                dispatcher()
+                    ->getDefaultEventSource()
+                    ->getEvent<mqbevt::ConfirmEvent>();
+            event_sp->setBlob(blobSp).setSource(this);
+            dispatcher()->dispatchEvent(bslmf::MovableRefUtil::move(event_sp),
+                                        this);
         }
         else if (event.isRejectEvent()) {
-            eventType = mqbi::DispatcherEventType::e_REJECT;
+            bsl::shared_ptr<mqbevt::RejectEvent> event_sp =
+                dispatcher()
+                    ->getDefaultEventSource()
+                    ->getEvent<mqbevt::RejectEvent>();
+            event_sp->setBlob(blobSp).setSource(this);
+            dispatcher()->dispatchEvent(bslmf::MovableRefUtil::move(event_sp),
+                                        this);
         }
         else {
             BALL_LOG_ERROR << "#CLIENT_UNEXPECTED_EVENT " << description()
                            << ": Unexpected event type: " << event;
             return;  // RETURN
         }
-
-        // Dispatch the event
-        // TODO(678098): revisit, use per-IO thread event source
-        mqbi::Dispatcher::DispatcherEventSp dispEvent =
-            dispatcher()->getDefaultEventSource()->getEvent();
-        bsl::shared_ptr<bdlbb::Blob> blobSp =
-            d_state.d_blobSpPool_p->getObject();
-        *blobSp = *(event.blob());
-        (*dispEvent).setType(eventType).setSource(this).setBlob(blobSp);
-        dispatcher()->dispatchEvent(bslmf::MovableRefUtil::move(dispEvent),
-                                    this);
     }
 }
 
@@ -2824,23 +2842,23 @@ void ClientSession::onDispatcherEvent(const mqbi::DispatcherEvent& event)
 
     switch (event.type()) {
     case mqbi::DispatcherEventType::e_CONFIRM: {
-        onConfirmEvent(*(event.asConfirmEvent()));
+        onConfirmEvent(*(event.the<mqbevt::ConfirmEvent>()));
     } break;
     case mqbi::DispatcherEventType::e_REJECT: {
-        onRejectEvent(*(event.asRejectEvent()));
+        onRejectEvent(*(event.the<mqbevt::RejectEvent>()));
     } break;
     case mqbi::DispatcherEventType::e_PUSH: {
-        onPushEvent(*(event.asPushEvent()));
+        onPushEvent(*(event.the<mqbevt::PushEvent>()));
     } break;
     case mqbi::DispatcherEventType::e_PUT: {
-        onPutEvent(*(event.asPutEvent()));
+        onPutEvent(*(event.the<mqbevt::PutEvent>()));
     } break;
     case mqbi::DispatcherEventType::e_ACK: {
-        onAckEvent(*(event.asAckEvent()));
+        onAckEvent(*(event.the<mqbevt::AckEvent>()));
     } break;
     case mqbi::DispatcherEventType::e_CALLBACK: {
-        const mqbi::DispatcherCallbackEvent* realEvent =
-            event.asCallbackEvent();
+        const mqbevt::CallbackEvent* const realEvent =
+            event.the<mqbevt::CallbackEvent>();
 
         BSLS_ASSERT_SAFE(!realEvent->callback().empty());
         flush();  // Flush any pending messages to guarantee ordering of events

src/groups/mqb/mqba/mqba_clientsession.h

@@ -92,6 +92,13 @@ class ClusterCatalog;
 namespace bmqst {
 class StatContext;
 }
+namespace mqbevt {
+class AckEvent;
+class ConfirmEvent;
+class PushEvent;
+class PutEvent;
+class RejectEvent;
+}
 
 namespace mqba {
 
@@ -486,19 +493,19 @@ class ClientSession : public mqbnet::Session,
         const bsl::shared_ptr<bmqsys::OperationLogger>& opLogger);
 
     /// Process the specified ack `event`.
-    void onAckEvent(const mqbi::DispatcherAckEvent& event);
+    void onAckEvent(const mqbevt::AckEvent& event);
 
     /// Process the specified confirm `event`.
-    void onConfirmEvent(const mqbi::DispatcherConfirmEvent& event);
+    void onConfirmEvent(const mqbevt::ConfirmEvent& event);
 
     /// Process the specified reject `event`.
-    void onRejectEvent(const mqbi::DispatcherRejectEvent& event);
+    void onRejectEvent(const mqbevt::RejectEvent& event);
 
     /// Process the specified push `event` received from the dispatcher.
-    void onPushEvent(const mqbi::DispatcherPushEvent& event);
+    void onPushEvent(const mqbevt::PushEvent& event);
 
     /// Process the specified put `event`.
-    void onPutEvent(const mqbi::DispatcherPutEvent& event);
+    void onPutEvent(const mqbevt::PutEvent& event);
 
     /// Validate a message of the specified `eventType` using the specified
     /// `queueId`. Return true if the message is valid and false otherwise.

src/groups/mqb/mqba/mqba_clientsession.t.cpp

@@ -19,6 +19,9 @@
 // MQB
 #include <mqbcfg_brokerconfig.h>
 #include <mqbcfg_messages.h>
+#include <mqbevt_ackevent.h>
+#include <mqbevt_pushevent.h>
+#include <mqbevt_putevent.h>
 #include <mqbi_queue.h>
 #include <mqbmock_cluster.h>
 #include <mqbmock_dispatcher.h>
@@ -641,10 +644,6 @@ T assertFail()
 
 /// The `TestBench` holds system components together.
 class TestBench {
-  private:
-    // PRIVATE TYPES
-    typedef mqbmock::Dispatcher::EventGuard EventGuard;
-
   public:
     // DATA
     bdlbb::PooledBlobBufferFactory            d_bufferFactory;
@@ -824,13 +823,11 @@ class TestBench {
             guid,
             queueId);
 
-        mqbi::Dispatcher::DispatcherEventSp event =
-            bsl::allocate_shared<mqbi::DispatcherEvent>(d_allocator_p);
-        (*event)
-            .setType(mqbi::DispatcherEventType::e_ACK)
-            .setAckMessage(ackMessage);
+        bsl::shared_ptr<mqbevt::AckEvent> event_sp =
+            bsl::allocate_shared<mqbevt::AckEvent>(d_allocator_p);
+        (*event_sp).setAckMessage(ackMessage);
 
-        dispatch(event);
+        dispatch(event_sp);
     }
 
     /// Sends a `Put` event for the specified `queueId`, `msgGUID` and
@@ -870,16 +867,6 @@ class TestBench {
                                       &d_bufferFactory,
                                       cat);
 
-        mqbi::Dispatcher::DispatcherEventSp event =
-            bsl::allocate_shared<mqbi::DispatcherEvent>(d_allocator_p);
-        (*event)
-            .setType(mqbi::DispatcherEventType::e_PUT)
-            .setIsRelay(true)  // Relay message
-            .setSource(&d_cs)  // DispatcherClient *value
-            .setPutHeader(putHeader)
-            .setBlob(eventBlob)  // const bsl::shared_ptr<bdlbb::Blob>& value
-            .setCompressionAlgorithmType(cat);
-
         // Internal-ticket D167598037.
         // Verify that PutMessageIterator does not change the input.
         bmqp::Event rawEvent(eventBlob.get(),
@@ -898,7 +885,12 @@ class TestBench {
         BSLS_ASSERT(pIt2.next());
         BSLS_ASSERT(pIt3.next());
 
-        dispatch(event);
+        bsl::shared_ptr<mqbevt::PutEvent> event_sp =
+            bsl::allocate_shared<mqbevt::PutEvent>(d_allocator_p);
+        (*event_sp).setPutHeader(putHeader).setBlob(eventBlob).setSource(
+            &d_cs);
+
+        dispatch(event_sp);
     }
 
     /// Sends a `Push` event for the specified `queueId`, `msgGUID` and
@@ -912,20 +904,20 @@ class TestBench {
     {
         PVV("Sending PUSH with queueId=" << queueId << ", guid=" << msgGUID);
 
-        mqbi::Dispatcher::DispatcherEventSp event =
-            bsl::allocate_shared<mqbi::DispatcherEvent>(d_allocator_p);
+        bsl::shared_ptr<mqbevt::PushEvent> event_sp =
+            bsl::allocate_shared<mqbevt::PushEvent>(d_allocator_p);
 
-        (*event)
-            .setType(mqbi::DispatcherEventType::e_PUSH)
-            .setSource(&d_cs)  // DispatcherClient *value
+        (*event_sp)
+            .setSource(&d_cs)
             .setQueueId(queueId)
             .setBlob(blob)
             .setGuid(msgGUID)
             .setMessagePropertiesInfo(logic)
             .setCompressionAlgorithmType(cat);
 
-        dispatch(event);
+        dispatch(event_sp);
     }
+
     bool validateData(const bdlbb::Blob& blob,
                       int                offset,
                       int                length = k_PAYLOAD_LENGTH)
@@ -1013,7 +1005,8 @@ class TestBench {
     /// using the specified `event`.
     void dispatch(mqbi::Dispatcher::DispatcherEventSp event)
     {
-        EventGuard guard(d_mockDispatcher._withEvent(&d_cs, event));
+        mqbmock::Dispatcher::EventGuard guard(
+            d_mockDispatcher._withEvent(&d_cs, event));
 
         d_cs.onDispatcherEvent(*event);
     }
@@ -1943,9 +1936,6 @@ static void test9_newStylePush()
 
     BMQTST_ASSERT_EQ(bmqt::EventBuilderResult::e_SUCCESS, rc);
 
-    mqbi::Dispatcher::DispatcherEventSp putEvent =
-        bsl::allocate_shared<mqbi::DispatcherEvent>(
-            bmqtst::TestHelperUtil::allocator());
     bmqp::Event           rawEvent(peb.blob().get(),
                          bmqtst::TestHelperUtil::allocator());
 
@@ -1957,14 +1947,15 @@ static void test9_newStylePush()
     rawEvent.loadPutMessageIterator(&putIt, false);
     BSLS_ASSERT(putIt.next());
 
-    (*putEvent)
-        .setType(mqbi::DispatcherEventType::e_PUT)
-        .setIsRelay(true)     // Relay message
-        .setSource(&tb.d_cs)  // DispatcherClient *value
+    bsl::shared_ptr<mqbevt::PutEvent> event_sp =
+        bsl::allocate_shared<mqbevt::PutEvent>(
+            bmqtst::TestHelperUtil::allocator());
+    (*event_sp)
+        .setSource(&tb.d_cs)
         .setPutHeader(putIt.header())
-        .setBlob(peb.blob());  // const bsl::shared_ptr<bdlbb::Blob>& value
+        .setBlob(peb.blob());
 
-    tb.dispatch(putEvent);
+    tb.dispatch(event_sp);
 
     // Check if a message was sent
     const bsl::vector<MyMockQueueHandle::Post>& postMessages =
@@ -2060,9 +2051,6 @@ static void test10_newStyleCompressedPush()
 
     BMQTST_ASSERT_EQ(bmqt::EventBuilderResult::e_SUCCESS, rc);
 
-    mqbi::Dispatcher::DispatcherEventSp putEvent =
-        bsl::allocate_shared<mqbi::DispatcherEvent>(
-            bmqtst::TestHelperUtil::allocator());
     bmqp::Event           rawEvent(peb.blob().get(),
                          bmqtst::TestHelperUtil::allocator());
 
@@ -2074,15 +2062,15 @@ static void test10_newStyleCompressedPush()
     rawEvent.loadPutMessageIterator(&putIt, false);
     BSLS_ASSERT(putIt.next());
 
-    (*putEvent)
-        .setType(mqbi::DispatcherEventType::e_PUT)
-        .setIsRelay(true)     // Relay message
-        .setSource(&tb.d_cs)  // DispatcherClient *value
+    bsl::shared_ptr<mqbevt::PutEvent> event_sp =
+        bsl::allocate_shared<mqbevt::PutEvent>(
+            bmqtst::TestHelperUtil::allocator());
+    (*event_sp)
+        .setSource(&tb.d_cs)
         .setPutHeader(putIt.header())
-        .setBlob(peb.blob())  // const bsl::shared_ptr<bdlbb::Blob>& value
-        .setCompressionAlgorithmType(bmqt::CompressionAlgorithmType::e_ZLIB);
+        .setBlob(peb.blob());
 
-    tb.dispatch(putEvent);
+    tb.dispatch(event_sp);
 
     // Check if a message was sent
     const bsl::vector<MyMockQueueHandle::Post>& postMessages =