Give copies of Properties their own header table

src/rmq/rmqa/CMakeLists.txt

@@ -17,7 +17,7 @@ add_library(rmqa OBJECT
     rmqa_topologyupdate.cpp
     rmqa_tracingconsumerimpl.cpp
     rmqa_tracingmessageguard.cpp
-    rmqa_tracingproducerimpl.cpp
+    rmqa_tracingtagger.cpp
     rmqa_vhost.cpp
     rmqa_vhostimpl.cpp
 )

src/rmq/rmqa/rmqa_producerimpl.cpp

@@ -117,29 +117,55 @@ void handleConfirmOnEventLoop(
     }
 }
 
+bsl::string extractExchangeName(const rmqt::ExchangeHandle& exchangeHandle)
+{
+    bsl::shared_ptr<rmqt::Exchange> exchange(exchangeHandle.lock());
+    return exchange ? exchange->name() : "<expired exchange>";
+}
+
 } // namespace
 
+ProducerImpl::Factory::Factory()
+: d_tagger()
+{
+}
+
+ProducerImpl::Factory::Factory(
+    const bsl::shared_ptr<rmqp::ProducerTagger>& tagger)
+: d_tagger(tagger)
+{
+}
+
 ProducerImpl::Factory::~Factory() {}
 
 bsl::shared_ptr<ProducerImpl> ProducerImpl::Factory::create(
     uint16_t maxOutstandingConfirms,
-    const rmqt::ExchangeHandle&,
+    const rmqt::ExchangeHandle& exchange,
     const bsl::shared_ptr<rmqamqp::SendChannel>& channel,
     bdlmt::ThreadPool& threadPool,
     rmqio::EventLoop& eventLoop) const
 {
-    return bsl::shared_ptr<ProducerImpl>(new ProducerImpl(
-        maxOutstandingConfirms, channel, threadPool, eventLoop));
+    return bsl::shared_ptr<ProducerImpl>(
+        new ProducerImpl(maxOutstandingConfirms,
+                         channel,
+                         threadPool,
+                         eventLoop,
+                         extractExchangeName(exchange),
+                         d_tagger));
 }
 
 ProducerImpl::ProducerImpl(uint16_t maxOutstandingConfirms,
                            const bsl::shared_ptr<rmqamqp::SendChannel>& channel,
                            bdlmt::ThreadPool& threadPool,
-                           rmqio::EventLoop& eventLoop)
+                           rmqio::EventLoop& eventLoop,
+                           const bsl::string& exchangeName,
+                           const bsl::shared_ptr<rmqp::ProducerTagger>& tagger)
 : d_eventLoop(eventLoop)
 , d_channel(channel)
 , d_sharedState(bsl::shared_ptr<SharedState>(
       new SharedState(true, threadPool, maxOutstandingConfirms)))
+, d_exchangeName(exchangeName)
+, d_tagger(tagger)
 {
     using namespace bdlf::PlaceHolders;
     channel->setCallback(bdlf::BindUtil::bind(
@@ -178,6 +204,21 @@ bool ProducerImpl::registerUniqueCallback(
     return true;
 }
 
+rmqt::Message ProducerImpl::prepareMessageForSending(
+    rmqp::Producer::ConfirmationCallback* callback,
+    const rmqt::Message& message,
+    const bsl::string& routingKey)
+{
+    rmqt::Message taggedMessage(message);
+
+    if (d_tagger) {
+        *callback = d_tagger->tagMessage(
+            &taggedMessage.properties(), routingKey, d_exchangeName, *callback);
+    }
+
+    return taggedMessage;
+}
+
 void ProducerImpl::addTransformer(
     const bsl::shared_ptr<rmqp::MessageTransformer>& transformer)
 {
@@ -261,9 +302,13 @@ rmqp::Producer::SendStatus ProducerImpl::sendImpl(
     const rmqp::Producer::ConfirmationCallback& confirmCallback,
     const bsls::TimeInterval& timeout)
 {
+    rmqp::Producer::ConfirmationCallback callback(confirmCallback);
+    const rmqt::Message taggedMessage =
+        prepareMessageForSending(&callback, message, routingKey);
+
     BALL_LOG_TRACE
         << "Waiting on send(exchange) outstanding message limit for message "
-        << message;
+        << taggedMessage;
 
     if (timeout.totalNanoseconds()) {
         if (d_sharedState->outstandingMessagesCap.timedWait(
@@ -275,19 +320,23 @@ rmqp::Producer::SendStatus ProducerImpl::sendImpl(
         d_sharedState->outstandingMessagesCap.wait();
     }
 
-    return doSend(message, routingKey, mandatoryFlag, confirmCallback);
+    return doSend(taggedMessage, routingKey, mandatoryFlag, callback);
 }
 
 rmqp::Producer::SendStatus ProducerImpl::trySend(
     const rmqt::Message& message,
     const bsl::string& routingKey,
     const rmqp::Producer::ConfirmationCallback& confirmCallback)
 {
+    rmqp::Producer::ConfirmationCallback callback(confirmCallback);
+    const rmqt::Message taggedMessage =
+        prepareMessageForSending(&callback, message, routingKey);
+
     if (!d_sharedState->outstandingMessagesCap.tryWait()) {
-        return doSend(message,
+        return doSend(taggedMessage,
                       routingKey,
                       rmqt::Mandatory::RETURN_UNROUTABLE,
-                      confirmCallback);
+                      callback);
     }
     else {
         BALL_LOG_TRACE << "Unconfirmed message limit already reached";

src/rmq/rmqa/rmqa_producerimpl.h

@@ -16,6 +16,8 @@
 #ifndef INCLUDED_RMQA_PRODUCERIMPL
 #define INCLUDED_RMQA_PRODUCERIMPL
 
+#include <rmqp_producertagger.h>
+
 #include <rmqp_messagetransformer.h>
 #include <rmqp_producer.h>
 #include <rmqt_endpoint.h>
@@ -55,20 +57,33 @@ class ProducerImpl : public rmqp::Producer {
   public:
     class Factory {
       public:
+        /// Create producers with no tagging hook.
+        Factory();
+
+        /// Create producers which invoke `tagger` on each outgoing message.
+        /// The tagger is shared by every producer this factory creates.
+        explicit Factory(const bsl::shared_ptr<rmqp::ProducerTagger>& tagger);
+
         virtual ~Factory();
         virtual bsl::shared_ptr<ProducerImpl>
         create(uint16_t maxOutstandingConfirms,
                const rmqt::ExchangeHandle& exchange,
                const bsl::shared_ptr<rmqamqp::SendChannel>& channel,
                bdlmt::ThreadPool& threadPool,
                rmqio::EventLoop& eventLoop) const;
+
+      private:
+        bsl::shared_ptr<rmqp::ProducerTagger> d_tagger;
     };
 
     // CREATORS
     ProducerImpl(uint16_t maxOutstandingConfirms,
                  const bsl::shared_ptr<rmqamqp::SendChannel>& channel,
                  bdlmt::ThreadPool& threadPool,
-                 rmqio::EventLoop& eventLoop);
+                 rmqio::EventLoop& eventLoop,
+                 const bsl::string& exchangeName = bsl::string(),
+                 const bsl::shared_ptr<rmqp::ProducerTagger>& tagger =
+                     bsl::shared_ptr<rmqp::ProducerTagger>());
 
     ~ProducerImpl() BSLS_KEYWORD_OVERRIDE;
 
@@ -135,6 +150,17 @@ class ProducerImpl : public rmqp::Producer {
         const bdlb::Guid& guid,
         const rmqp::Producer::ConfirmationCallback& confirmCallback);
 
+    /// Return a copy of `message` which owns its header table, having offered
+    /// it to the tagger. On return `*callback` is the callback to publish
+    /// with, wrapped by the tagger if it asked to be.
+    ///
+    /// Must be called on the sending thread and before any wait on the
+    /// outstanding confirm limit, as `rmqp::ProducerTagger` requires.
+    rmqt::Message
+    prepareMessageForSending(rmqp::Producer::ConfirmationCallback* callback,
+                             const rmqt::Message& message,
+                             const bsl::string& routingKey);
+
     rmqp::Producer::SendStatus
     doSend(const rmqt::Message& message,
            const bsl::string& routingKey,
@@ -159,6 +185,11 @@ class ProducerImpl : public rmqp::Producer {
 
     bsl::vector<bsl::shared_ptr<rmqp::MessageTransformer> > d_transformers;
 
+    bsl::string d_exchangeName;
+
+    /// Null when nothing is configured to tag outgoing messages.
+    bsl::shared_ptr<rmqp::ProducerTagger> d_tagger;
+
 }; // class Producer
 
 } // namespace rmqa

src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp

@@ -20,7 +20,7 @@
 #include <rmqa_noopmetricpublisher.h>
 #include <rmqa_producerimpl.h>
 #include <rmqa_tracingconsumerimpl.h>
-#include <rmqa_tracingproducerimpl.h>
+#include <rmqa_tracingtagger.h>
 #include <rmqa_vhost.h>
 #include <rmqa_vhostimpl.h>
 
@@ -404,8 +404,9 @@ rmqt::Future<rmqp::Connection> RabbitContextImpl::createNewConnection(
 
     bsl::shared_ptr<ProducerImpl::Factory> producerFactory(
         d_producerTracing
-            ? bsl::shared_ptr<ProducerImpl::Factory>(
-                  new TracingProducerImpl::Factory(endpoint, d_producerTracing))
+            ? bsl::make_shared<ProducerImpl::Factory>(
+                  bsl::shared_ptr<rmqp::ProducerTagger>(
+                      new TracingTagger(endpoint, d_producerTracing)))
             : bsl::make_shared<ProducerImpl::Factory>());
 
     rmqamqp::Connection::ConnectedCallback cb =