Overload the send function in TracingProducerImpl

src/rmq/rmqa/rmqa_tracingproducerimpl.cpp

@@ -105,6 +105,33 @@ rmqp::Producer::SendStatus TracingProducerImpl::send(
                               timeout);
 }
 
+rmqp::Producer::SendStatus TracingProducerImpl::send(
+    const rmqt::Message& message,
+    const bsl::string& routingKey,
+    rmqt::Mandatory::Value mandatoryFlag,
+    const rmqp::Producer::ConfirmationCallback& confirmCallback,
+    const bsls::TimeInterval& timeout)
+{
+    rmqt::Message newMessage(message);
+    // ideally we'd have move semantics on the message to avoid
+    // copying the metadata note that this is not a deep copy of
+    // the message payload
+    bsl::shared_ptr<rmqp::ProducerTracing::Context> context =
+        d_tracing->createAndTag(
+            &(newMessage.properties()), routingKey, d_exchangeName, d_endpoint);
+
+    return ProducerImpl::send(newMessage,
+                              routingKey,
+                              mandatoryFlag,
+                              bdlf::BindUtil::bind(&callbackAndContext,
+                                                   confirmCallback,
+                                                   context,
+                                                   bdlf::PlaceHolders::_1,
+                                                   bdlf::PlaceHolders::_2,
+                                                   bdlf::PlaceHolders::_3),
+                              timeout);
+}
+
 rmqp::Producer::SendStatus TracingProducerImpl::trySend(
     const rmqt::Message& message,
     const bsl::string& routingKey,

src/rmq/rmqa/rmqa_tracingproducerimpl.h

@@ -64,6 +64,12 @@ class TracingProducerImpl : public ProducerImpl {
                     const rmqp::Producer::ConfirmationCallback& confirmCallback,
                     const bsls::TimeInterval& timeout) BSLS_KEYWORD_OVERRIDE;
 
+    SendStatus send(const rmqt::Message& message,
+                    const bsl::string& routingKey,
+                    rmqt::Mandatory::Value mandatoryFlag,
+                    const rmqp::Producer::ConfirmationCallback& confirmCallback,
+                    const bsls::TimeInterval& timeout) BSLS_KEYWORD_OVERRIDE;
+
     SendStatus
     trySend(const rmqt::Message& message,
             const bsl::string& routingKey,

src/tests/rmqa/rmqa_producerimpl.t.cpp

@@ -867,6 +867,53 @@ TEST_P(TracingProducerImplTests, SendConfirmCallsTracing)
     d_threadPool.drain();
 }
 
+TEST_P(TracingProducerImplTests, SendWithMandatoryFlagConfirmCallsTracing)
+{
+    // Regression: the send() overload accepting an explicit mandatory flag must
+    // also create a tracing context and tag the message, just like the
+    // four-argument send(). Previously TracingProducerImpl only overrode the
+    // four-argument send(), so publishing with an explicit mandatory flag
+    // silently bypassed tracing.
+
+    // GIVEN
+    bsl::shared_ptr<MockProducerTracing::MockContext> tracingContext(
+        bsl::make_shared<MockProducerTracing::MockContext>());
+    rmqt::Properties specialProperties = d_message.properties();
+    specialProperties.headers          = bsl::make_shared<rmqt::FieldTable>();
+    specialProperties.headers->insert(
+        bsl::make_pair(bsl::string("special"), bsl::string("property")));
+
+    bsl::shared_ptr<rmqa::ProducerImpl> producer(d_factory->create(
+        1, d_exchange, d_mockSendChannel, d_threadPool, d_eventLoop));
+
+    // The tagged message must be published, and the caller-supplied mandatory
+    // flag must be forwarded unchanged to the channel.
+    EXPECT_CALL(*d_mockSendChannel,
+                publishMessage(MessagePropertiesMatch(specialProperties),
+                               bsl::string("routingKey"),
+                               rmqt::Mandatory::DISCARD_UNROUTABLE));
+    EXPECT_CALL(
+        *d_tracing,
+        createAndTag(
+            _, bsl::string("routingKey"), bsl::string("test-exchange"), _))
+        .WillOnce(
+            DoAll(SetArgPointee<0>(specialProperties), Return(tracingContext)));
+    // WHEN
+    producer->send(d_message,
+                   "routingKey",
+                   rmqt::Mandatory::DISCARD_UNROUTABLE,
+                   d_callback,
+                   d_timeout);
+
+    const rmqt::ConfirmResponse confirmResponse(rmqt::ConfirmResponse::ACK);
+
+    EXPECT_CALL(*tracingContext, response(confirmResponse)).WillOnce(Return());
+
+    d_injectConfirm(d_message, d_exchange->name(), confirmResponse);
+
+    d_threadPool.drain();
+}
+
 RMQTESTUTIL_TESTSUITE_P(AllMembers,
                         ProducerImplTests,
                         Values(PRODUCER, TRACING_PRODUCER),