From 8abe60024314ee30cb130e74541c44973136bf73 Mon Sep 17 00:00:00 2001 From: Freddie Lamble Date: Tue, 8 Sep 2026 09:02:00 +0000 Subject: [PATCH 1/2] Tag outgoing messages via composition Tracing was attached to a producer by deriving from it. `TracingProducerImpl` overrode every publishing entry point, both `send` overloads and `trySend`, with three near identical bodies which copied the message, called `createAndTag` and re-wrapped the confirmation callback. That shape has to be extended by hand for every new overload, and had already failed that way once. The `send` overload taking an explicit mandatory flag silently bypassed tracing until it was noticed and a third copy of the body added in https://github.com/bloomberg/rmqcpp/pull/91. Inheritance was doing very little work here. The subclass carried three data members and one behaviour, and that behaviour is not a variation on being a producer, it is a step in publishing a message. Use composition instead. `ProducerImpl` now optionally holds an `rmqp::ProducerTagger`, invoked once per send from `prepareMessageForSending`. `rmqa::TracingTagger` adapts the configured `rmqp::ProducerTracing` onto it, and `TracingProducerImpl` is deleted along with its factory. The exchange name is passed to the tagger rather than held by it, so a tagger has no per producer state and one instance serves every producer on a connection. `sendImpl` and `trySend` share `prepareMessageForSending` because the step cannot move into `doSend`. It has to run before the wait on the outstanding confirm limit, and those two differ in precisely what sits between the two, a blocking wait against a try. The helper gives that ordering one home. This is the pattern the consumer already uses. `TracingConsumerImpl` is an empty shell around a factory which builds a plain `ConsumerImpl` with a different `MessageGuard::Factory`, so consumer tracing has always been injected as a collaborator. The producer was the odd one out. Only the empty shell class is left over on that side, and it can go whenever the consumer is next touched. The `rmqp::ProducerTracing` interface is unchanged, so tracing implementations need no change. Three behaviours they rely on which the interface cannot express are preserved. 1. The hook runs on the calling thread 2. It runs before the wait on the outstanding confirm limit 3. The context it returns lives until the broker responds --- src/rmq/rmqa/CMakeLists.txt | 2 +- src/rmq/rmqa/rmqa_producerimpl.cpp | 65 +++++++-- src/rmq/rmqa/rmqa_producerimpl.h | 33 ++++- src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp | 7 +- src/rmq/rmqa/rmqa_tracingproducerimpl.cpp | 159 ---------------------- src/rmq/rmqa/rmqa_tracingproducerimpl.h | 89 ------------ src/rmq/rmqa/rmqa_tracingtagger.cpp | 67 +++++++++ src/rmq/rmqa/rmqa_tracingtagger.h | 63 +++++++++ src/rmq/rmqp/CMakeLists.txt | 1 + src/rmq/rmqp/rmqp_producertagger.cpp | 24 ++++ src/rmq/rmqp/rmqp_producertagger.h | 65 +++++++++ src/tests/rmqa/rmqa_producerimpl.t.cpp | 32 +++-- 12 files changed, 331 insertions(+), 276 deletions(-) delete mode 100644 src/rmq/rmqa/rmqa_tracingproducerimpl.cpp delete mode 100644 src/rmq/rmqa/rmqa_tracingproducerimpl.h create mode 100644 src/rmq/rmqa/rmqa_tracingtagger.cpp create mode 100644 src/rmq/rmqa/rmqa_tracingtagger.h create mode 100644 src/rmq/rmqp/rmqp_producertagger.cpp create mode 100644 src/rmq/rmqp/rmqp_producertagger.h diff --git a/src/rmq/rmqa/CMakeLists.txt b/src/rmq/rmqa/CMakeLists.txt index 98236df5..c7e73015 100644 --- a/src/rmq/rmqa/CMakeLists.txt +++ b/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 ) diff --git a/src/rmq/rmqa/rmqa_producerimpl.cpp b/src/rmq/rmqa/rmqa_producerimpl.cpp index cdcb9b40..5569be72 100644 --- a/src/rmq/rmqa/rmqa_producerimpl.cpp +++ b/src/rmq/rmqa/rmqa_producerimpl.cpp @@ -117,29 +117,55 @@ void handleConfirmOnEventLoop( } } +bsl::string extractExchangeName(const rmqt::ExchangeHandle& exchangeHandle) +{ + bsl::shared_ptr exchange(exchangeHandle.lock()); + return exchange ? exchange->name() : ""; +} + } // namespace +ProducerImpl::Factory::Factory() +: d_tagger() +{ +} + +ProducerImpl::Factory::Factory( + const bsl::shared_ptr& tagger) +: d_tagger(tagger) +{ +} + ProducerImpl::Factory::~Factory() {} bsl::shared_ptr ProducerImpl::Factory::create( uint16_t maxOutstandingConfirms, - const rmqt::ExchangeHandle&, + const rmqt::ExchangeHandle& exchange, const bsl::shared_ptr& channel, bdlmt::ThreadPool& threadPool, rmqio::EventLoop& eventLoop) const { - return bsl::shared_ptr(new ProducerImpl( - maxOutstandingConfirms, channel, threadPool, eventLoop)); + return bsl::shared_ptr( + new ProducerImpl(maxOutstandingConfirms, + channel, + threadPool, + eventLoop, + extractExchangeName(exchange), + d_tagger)); } ProducerImpl::ProducerImpl(uint16_t maxOutstandingConfirms, const bsl::shared_ptr& channel, bdlmt::ThreadPool& threadPool, - rmqio::EventLoop& eventLoop) + rmqio::EventLoop& eventLoop, + const bsl::string& exchangeName, + const bsl::shared_ptr& tagger) : d_eventLoop(eventLoop) , d_channel(channel) , d_sharedState(bsl::shared_ptr( 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& 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,7 +320,7 @@ 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( @@ -283,11 +328,15 @@ rmqp::Producer::SendStatus ProducerImpl::trySend( 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"; diff --git a/src/rmq/rmqa/rmqa_producerimpl.h b/src/rmq/rmqa/rmqa_producerimpl.h index a789e978..373d8cc5 100644 --- a/src/rmq/rmqa/rmqa_producerimpl.h +++ b/src/rmq/rmqa/rmqa_producerimpl.h @@ -16,6 +16,8 @@ #ifndef INCLUDED_RMQA_PRODUCERIMPL #define INCLUDED_RMQA_PRODUCERIMPL +#include + #include #include #include @@ -55,6 +57,13 @@ 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& tagger); + virtual ~Factory(); virtual bsl::shared_ptr create(uint16_t maxOutstandingConfirms, @@ -62,13 +71,19 @@ class ProducerImpl : public rmqp::Producer { const bsl::shared_ptr& channel, bdlmt::ThreadPool& threadPool, rmqio::EventLoop& eventLoop) const; + + private: + bsl::shared_ptr d_tagger; }; // CREATORS ProducerImpl(uint16_t maxOutstandingConfirms, const bsl::shared_ptr& channel, bdlmt::ThreadPool& threadPool, - rmqio::EventLoop& eventLoop); + rmqio::EventLoop& eventLoop, + const bsl::string& exchangeName = bsl::string(), + const bsl::shared_ptr& tagger = + bsl::shared_ptr()); ~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`, 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 > d_transformers; + bsl::string d_exchangeName; + + /// Null when nothing is configured to tag outgoing messages. + bsl::shared_ptr d_tagger; + }; // class Producer } // namespace rmqa diff --git a/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp b/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp index 9196923d..904c7cb8 100644 --- a/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp +++ b/src/rmq/rmqa/rmqa_rabbitcontextimpl.cpp @@ -20,7 +20,7 @@ #include #include #include -#include +#include #include #include @@ -404,8 +404,9 @@ rmqt::Future RabbitContextImpl::createNewConnection( bsl::shared_ptr producerFactory( d_producerTracing - ? bsl::shared_ptr( - new TracingProducerImpl::Factory(endpoint, d_producerTracing)) + ? bsl::make_shared( + bsl::shared_ptr( + new TracingTagger(endpoint, d_producerTracing))) : bsl::make_shared()); rmqamqp::Connection::ConnectedCallback cb = diff --git a/src/rmq/rmqa/rmqa_tracingproducerimpl.cpp b/src/rmq/rmqa/rmqa_tracingproducerimpl.cpp deleted file mode 100644 index 3f2682a1..00000000 --- a/src/rmq/rmqa/rmqa_tracingproducerimpl.cpp +++ /dev/null @@ -1,159 +0,0 @@ -// Copyright 2020-2023 Bloomberg Finance L.P. -// SPDX-License-Identifier: Apache-2.0 -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include - -#include -#include - -namespace BloombergLP { -namespace rmqa { - -namespace { -void callbackAndContext(const rmqp::Producer::ConfirmationCallback& callback, - bsl::shared_ptr context, - const rmqt::Message& message, - const bsl::string& routingKey, - const rmqt::ConfirmResponse& response) -{ - callback(message, routingKey, response); - context->response(response); -} - -bsl::string extractExchangeName(const rmqt::ExchangeHandle& exchangeHandle) -{ - bsl::shared_ptr exchange(exchangeHandle.lock()); - return exchange ? exchange->name() : ""; -} - -} // namespace - -TracingProducerImpl::Factory::Factory( - const bsl::shared_ptr& endpoint, - const bsl::shared_ptr& tracing) -: d_endpoint(endpoint) -, d_tracing(tracing) -{ -} - -bsl::shared_ptr TracingProducerImpl::Factory::create( - uint16_t maxOutstandingConfirms, - const rmqt::ExchangeHandle& exchange, - const bsl::shared_ptr& channel, - bdlmt::ThreadPool& threadPool, - rmqio::EventLoop& eventLoop) const -{ - return bsl::shared_ptr( - new TracingProducerImpl(maxOutstandingConfirms, - channel, - threadPool, - eventLoop, - extractExchangeName(exchange), - d_endpoint, - d_tracing)); -} - -TracingProducerImpl::TracingProducerImpl( - uint16_t maxOutstandingConfirms, - const bsl::shared_ptr& channel, - bdlmt::ThreadPool& threadPool, - rmqio::EventLoop& eventLoop, - const bsl::string& exchangeName, - const bsl::shared_ptr& endpoint, - const bsl::shared_ptr& tracing) -: ProducerImpl(maxOutstandingConfirms, channel, threadPool, eventLoop) -, d_exchangeName(exchangeName) -, d_endpoint(endpoint) -, d_tracing(tracing) -{ -} - -rmqp::Producer::SendStatus TracingProducerImpl::send( - const rmqt::Message& message, - const bsl::string& routingKey, - 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 context = - d_tracing->createAndTag( - &(newMessage.properties()), routingKey, d_exchangeName, d_endpoint); - - return ProducerImpl::send(newMessage, - routingKey, - bdlf::BindUtil::bind(&callbackAndContext, - confirmCallback, - context, - bdlf::PlaceHolders::_1, - bdlf::PlaceHolders::_2, - bdlf::PlaceHolders::_3), - 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 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, - const rmqp::Producer::ConfirmationCallback& confirmCallback) -{ - 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 context = - d_tracing->createAndTag( - &(newMessage.properties()), routingKey, d_exchangeName, d_endpoint); - - return ProducerImpl::trySend(newMessage, - routingKey, - bdlf::BindUtil::bind(&callbackAndContext, - confirmCallback, - context, - bdlf::PlaceHolders::_1, - bdlf::PlaceHolders::_2, - bdlf::PlaceHolders::_3)); -} - -} // namespace rmqa -} // namespace BloombergLP diff --git a/src/rmq/rmqa/rmqa_tracingproducerimpl.h b/src/rmq/rmqa/rmqa_tracingproducerimpl.h deleted file mode 100644 index 1df085d5..00000000 --- a/src/rmq/rmqa/rmqa_tracingproducerimpl.h +++ /dev/null @@ -1,89 +0,0 @@ -// Copyright 2020-2023 Bloomberg Finance L.P. -// SPDX-License-Identifier: Apache-2.0 -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef INCLUDED_RMQA_TRACINGPRODUCERIMPL -#define INCLUDED_RMQA_TRACINGPRODUCERIMPL - -#include - -#include - -#include -#include - -//@PURPOSE: Implements the rmqa::Producer interface and specialises for tracing -// -//@CLASSES: -// rmqa::ProducerImpl: Manages interaction between rmqa <-> rmq internals - -namespace BloombergLP { -namespace rmqa { - -class TracingProducerImpl : public ProducerImpl { - public: - class Factory : public ProducerImpl::Factory { - public: - Factory(const bsl::shared_ptr& endpoint, - const bsl::shared_ptr& tracing); - - virtual bsl::shared_ptr - create(uint16_t maxOutstandingConfirms, - const rmqt::ExchangeHandle& exchange, - const bsl::shared_ptr& channel, - bdlmt::ThreadPool& threadPool, - rmqio::EventLoop& eventLoop) const BSLS_KEYWORD_OVERRIDE; - - private: - bsl::shared_ptr d_endpoint; - bsl::shared_ptr d_tracing; - }; - - // CREATORS - TracingProducerImpl(uint16_t maxOutstandingConfirms, - const bsl::shared_ptr& channel, - bdlmt::ThreadPool& threadPool, - rmqio::EventLoop& eventLoop, - const bsl::string& exchangeName, - const bsl::shared_ptr& endpoint, - const bsl::shared_ptr& tracing); - - SendStatus send(const rmqt::Message& message, - const bsl::string& routingKey, - 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, - const rmqp::Producer::ConfirmationCallback& confirmCallback) - BSLS_KEYWORD_OVERRIDE; - - private: - bsl::string d_exchangeName; - bsl::shared_ptr d_endpoint; - bsl::shared_ptr d_tracing; - -}; // class TracingProducerImpl - -} // namespace rmqa -} // namespace BloombergLP - -#endif diff --git a/src/rmq/rmqa/rmqa_tracingtagger.cpp b/src/rmq/rmqa/rmqa_tracingtagger.cpp new file mode 100644 index 00000000..3574a81f --- /dev/null +++ b/src/rmq/rmqa/rmqa_tracingtagger.cpp @@ -0,0 +1,67 @@ +// Copyright 2020-2023 Bloomberg Finance L.P. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include + +#include + +#include +#include + +namespace BloombergLP { +namespace rmqa { + +namespace { + +void callbackAndContext(const rmqp::Producer::ConfirmationCallback& callback, + bsl::shared_ptr context, + const rmqt::Message& message, + const bsl::string& routingKey, + const rmqt::ConfirmResponse& response) +{ + callback(message, routingKey, response); + context->response(response); +} + +} // namespace + +TracingTagger::TracingTagger( + const bsl::shared_ptr& endpoint, + const bsl::shared_ptr& tracing) +: d_endpoint(endpoint) +, d_tracing(tracing) +{ +} + +rmqp::Producer::ConfirmationCallback TracingTagger::tagMessage( + rmqt::Properties* messageProperties, + const bsl::string& routingKey, + const bsl::string& exchangeName, + const rmqp::Producer::ConfirmationCallback& confirmCallback) +{ + bsl::shared_ptr context = + d_tracing->createAndTag( + messageProperties, routingKey, exchangeName, d_endpoint); + + return bdlf::BindUtil::bind(&callbackAndContext, + confirmCallback, + context, + bdlf::PlaceHolders::_1, + bdlf::PlaceHolders::_2, + bdlf::PlaceHolders::_3); +} + +} // namespace rmqa +} // namespace BloombergLP diff --git a/src/rmq/rmqa/rmqa_tracingtagger.h b/src/rmq/rmqa/rmqa_tracingtagger.h new file mode 100644 index 00000000..cb9be845 --- /dev/null +++ b/src/rmq/rmqa/rmqa_tracingtagger.h @@ -0,0 +1,63 @@ +// Copyright 2020-2023 Bloomberg Finance L.P. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef INCLUDED_RMQA_TRACINGTAGGER +#define INCLUDED_RMQA_TRACINGTAGGER + +#include + +#include +#include + +#include +#include +#include + +//@PURPOSE: Attach a configured rmqp::ProducerTracing to a producer +// +//@CLASSES: +// rmqa::TracingTagger: adapts rmqp::ProducerTracing onto rmqp::ProducerTagger + +namespace BloombergLP { +namespace rmqa { + +/// \brief Adapts a configured `rmqp::ProducerTracing` onto the producer's +/// tagging hook, opening a tracing context per message and holding it alive +/// until the broker responds. +/// +/// Holds no per-producer state, so one instance serves every producer on a +/// connection. +class TracingTagger : public rmqp::ProducerTagger { + public: + TracingTagger(const bsl::shared_ptr& endpoint, + const bsl::shared_ptr& tracing); + + rmqp::Producer::ConfirmationCallback + tagMessage(rmqt::Properties* messageProperties, + const bsl::string& routingKey, + const bsl::string& exchangeName, + const rmqp::Producer::ConfirmationCallback& confirmCallback) + BSLS_KEYWORD_OVERRIDE; + + private: + bsl::shared_ptr d_endpoint; + bsl::shared_ptr d_tracing; + +}; // class TracingTagger + +} // namespace rmqa +} // namespace BloombergLP + +#endif diff --git a/src/rmq/rmqp/CMakeLists.txt b/src/rmq/rmqp/CMakeLists.txt index 8d858ab5..fee882f8 100644 --- a/src/rmq/rmqp/CMakeLists.txt +++ b/src/rmq/rmqp/CMakeLists.txt @@ -6,6 +6,7 @@ add_library(rmqp OBJECT rmqp_messagetransformer.cpp rmqp_metricpublisher.cpp rmqp_producer.cpp + rmqp_producertagger.cpp rmqp_producertracing.cpp rmqp_rabbitcontext.cpp rmqp_topology.cpp diff --git a/src/rmq/rmqp/rmqp_producertagger.cpp b/src/rmq/rmqp/rmqp_producertagger.cpp new file mode 100644 index 00000000..8da613e2 --- /dev/null +++ b/src/rmq/rmqp/rmqp_producertagger.cpp @@ -0,0 +1,24 @@ +// Copyright 2020-2023 Bloomberg Finance L.P. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include + +namespace BloombergLP { +namespace rmqp { + +ProducerTagger::~ProducerTagger() {} + +} // namespace rmqp +} // namespace BloombergLP diff --git a/src/rmq/rmqp/rmqp_producertagger.h b/src/rmq/rmqp/rmqp_producertagger.h new file mode 100644 index 00000000..78f8ed15 --- /dev/null +++ b/src/rmq/rmqp/rmqp_producertagger.h @@ -0,0 +1,65 @@ +// Copyright 2020-2023 Bloomberg Finance L.P. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef INCLUDED_RMQP_PRODUCERTAGGER +#define INCLUDED_RMQP_PRODUCERTAGGER + +#include +#include + +#include + +//@PURPOSE: Decorate messages on their way out of a Producer +// +//@CLASSES: +// rmqp::ProducerTagger: tags an outgoing message and wraps its confirmation +// callback + +namespace BloombergLP { +namespace rmqp { + +/// \brief A hook invoked by a producer once per send, to modify the message +/// about to be published and to wrap its confirmation callback with state +/// which must outlive the send. Distributed tracing is attached this way, by +/// implementing `rmqp::ProducerTracing`. +/// +/// One tagger is shared by every producer on a connection and is called +/// concurrently, so implementations must be thread safe. +class ProducerTagger { + public: + virtual ~ProducerTagger(); + + /// Called on the sending thread, before the message is queued and before + /// any wait on the producer's outstanding confirm limit, so an + /// implementation may read thread local state and anything it times + /// covers that wait. + /// + /// \param messageProperties owned by the producer, safe to modify in + /// place. + /// \param routingKey the routing key used for this send. + /// \param exchangeName the exchange the producer publishes to. + /// \param confirmCallback the callback supplied by the application. + /// \return the callback to publish with, or `confirmCallback` unchanged. + virtual rmqp::Producer::ConfirmationCallback + tagMessage(rmqt::Properties* messageProperties, + const bsl::string& routingKey, + const bsl::string& exchangeName, + const rmqp::Producer::ConfirmationCallback& confirmCallback) = 0; +}; + +} // namespace rmqp +} // namespace BloombergLP + +#endif diff --git a/src/tests/rmqa/rmqa_producerimpl.t.cpp b/src/tests/rmqa/rmqa_producerimpl.t.cpp index be481c45..82192269 100644 --- a/src/tests/rmqa/rmqa_producerimpl.t.cpp +++ b/src/tests/rmqa/rmqa_producerimpl.t.cpp @@ -17,7 +17,7 @@ #include -#include +#include #include #include @@ -166,11 +166,14 @@ class ProducerImplTests : public TestWithParam { bsl::shared_ptr paramPicker(ProducerType pt) { switch (pt) { - case TRACING_PRODUCER: - return bsl::make_shared( - bsl::make_shared("example-hostname", - "example-vhost"), - d_tracing); + case TRACING_PRODUCER: { + bsl::shared_ptr tagger( + new rmqa::TracingTagger( + bsl::make_shared( + "example-hostname", "example-vhost"), + d_tracing)); + return bsl::make_shared(tagger); + } default: return bsl::make_shared(); } @@ -823,7 +826,7 @@ TEST_P(ProducerImplMaxOutstandingTests, SendFromConfirmCallbackDoesNotDeadlock) d_threadPool.drain(); } -class TracingProducerImplTests : public ProducerImplMaxOutstandingTests { +class TracingTaggerTests : public ProducerImplMaxOutstandingTests { public: }; @@ -832,7 +835,7 @@ MATCHER_P(MessagePropertiesMatch, expected, "") return arg.properties() == expected; } -TEST_P(TracingProducerImplTests, SendConfirmCallsTracing) +TEST_P(TracingTaggerTests, SendConfirmCallsTracing) { // GIVEN bsl::shared_ptr tracingContext( @@ -867,13 +870,12 @@ TEST_P(TracingProducerImplTests, SendConfirmCallsTracing) d_threadPool.drain(); } -TEST_P(TracingProducerImplTests, SendWithMandatoryFlagConfirmCallsTracing) +TEST_P(TracingTaggerTests, 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. + // Regression. Tracing used to be a ProducerImpl subclass overriding each + // send() separately, and this overload was missed, so publishing with an + // explicit mandatory flag silently bypassed tracing. The tagger is reached + // from one place now, but keep the coverage. // GIVEN bsl::shared_ptr tracingContext( @@ -936,6 +938,6 @@ RMQTESTUTIL_TESTSUITE_P(AllMembers, ProducerImplTests::PrintParamName()); RMQTESTUTIL_TESTSUITE_P(AllMembers, - TracingProducerImplTests, + TracingTaggerTests, Values(TRACING_PRODUCER), ProducerImplTests::PrintParamName()); From 3b8ae11205215c5c1640f9faf4e585fe84ecca69 Mon Sep 17 00:00:00 2001 From: Freddie Lamble Date: Tue, 8 Sep 2026 09:03:10 +0000 Subject: [PATCH 2/2] Give copies of Properties their own header table `rmqt::Properties` holds its headers as a `bsl::shared_ptr`, and neither `Properties` nor `Message` declared a copy constructor, so copying either one copied the pointer and not the table. Every copy of a message therefore aliased one `FieldTable`, which is a `bsl::map` with no locking. Sending is asynchronous. `send()` posts the publish to the event loop and returns, and the headers are walked much later, on the event loop thread, to build the content header frame. A caller which reuses a `Properties` across sends is then mutating a map which is concurrently being read. ``` rmqt::Properties props = setupProps(); for (int x = 0; x < 10; ++x) { (*props.headers)["appheader"] = x; producer.send(message, routingKey, callback); } ``` Nothing about tracing is required for this. It applies equally to the transformer path, which emplaces its marker headers into the same shared table, and to a plain send with no hooks at all. A tracer setting properties on every send is simply the most likely writer to make it visible. Fix it in `Properties` rather than at the call sites which happen to matter today. The copy constructor and assignment operator now give the copy its own header table, so the copies the producer and consumer already make are correct without either of them doing anything special. A destructor is declared alongside them for consistency. One minor observable change. A caller which reads a header back out of its own table after `send()`, a trace id say, no longer sees it. That only ever worked by virtue of this aliasing, and only for a single threaded sender. `Methods_BasicProperties.Headers` asserted that `setProperties` left the stored headers pointer equal to the one passed in. That is the aliasing being removed, so it now compares the tables by value. This costs copies, which is accepted for now and left to be addressed with other publishing performance work. Measured on the send path, four `Properties` copies per send each deep copy the header table where previously they shared a pointer, and that is a lower bound because the tests exercise a mock event loop which skips the real handler copy. Declaring the copy operations also suppresses the implicit move operations on compilers which have them, so a move of a `Properties` or a `Message` is now a deep copy too. Both are worth revisiting when C++03 support can be dropped and move operations can be declared. `PropertiesTests` covers the copy and assignment behaviour directly, including that every field is copied, that a null header table stays null, and that self assignment is safe. `SendDoesNotShareHeaderTableWithCaller` runs against the plain and the tracing producer and fails against both without the fix, and `TracingDoesNotMutateCallerHeaders` drives a hook which injects in place, as the real ones do, and checks the injected header reaches the broker but not the caller. --- src/rmq/rmqa/rmqa_producerimpl.h | 6 +- src/rmq/rmqt/rmqt_properties.cpp | 63 +++++++++ src/rmq/rmqt/rmqt_properties.h | 11 ++ src/tests/rmqa/rmqa_producerimpl.t.cpp | 91 +++++++++++++ .../rmqamqpt/rmqamqpt_basicproperties.t.cpp | 3 +- src/tests/rmqt/CMakeLists.txt | 1 + src/tests/rmqt/rmqt_properties.t.cpp | 124 ++++++++++++++++++ 7 files changed, 295 insertions(+), 4 deletions(-) create mode 100644 src/tests/rmqt/rmqt_properties.t.cpp diff --git a/src/rmq/rmqa/rmqa_producerimpl.h b/src/rmq/rmqa/rmqa_producerimpl.h index 373d8cc5..049cd65a 100644 --- a/src/rmq/rmqa/rmqa_producerimpl.h +++ b/src/rmq/rmqa/rmqa_producerimpl.h @@ -150,9 +150,9 @@ class ProducerImpl : public rmqp::Producer { const bdlb::Guid& guid, const rmqp::Producer::ConfirmationCallback& confirmCallback); - /// Return a copy of `message`, having offered it to the tagger. On return - /// `*callback` is the callback to publish with, wrapped by the tagger if - /// it asked to be. + /// 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. diff --git a/src/rmq/rmqt/rmqt_properties.cpp b/src/rmq/rmqt/rmqt_properties.cpp index f40e24d8..0c90ab15 100644 --- a/src/rmq/rmqt/rmqt_properties.cpp +++ b/src/rmq/rmqt/rmqt_properties.cpp @@ -28,8 +28,71 @@ bool headersEquality(const bsl::shared_ptr& lhs, { return (lhs == rhs) || ((lhs && rhs) && (*lhs == *rhs)); } + +bsl::shared_ptr +clonedHeaders(const bsl::shared_ptr& headers) +{ + return headers ? bsl::make_shared(*headers) : headers; +} } // namespace +Properties::Properties() +: contentType() +, contentEncoding() +, headers() +, deliveryMode() +, priority() +, correlationId() +, replyTo() +, expiration() +, messageId() +, timestamp() +, type() +, userId() +, appId() +{ +} + +Properties::Properties(const Properties& original) +: contentType(original.contentType) +, contentEncoding(original.contentEncoding) +, headers(clonedHeaders(original.headers)) +, deliveryMode(original.deliveryMode) +, priority(original.priority) +, correlationId(original.correlationId) +, replyTo(original.replyTo) +, expiration(original.expiration) +, messageId(original.messageId) +, timestamp(original.timestamp) +, type(original.type) +, userId(original.userId) +, appId(original.appId) +{ +} + +Properties::~Properties() {} + +Properties& Properties::operator=(const Properties& rhs) +{ + if (this != &rhs) { + contentType = rhs.contentType; + contentEncoding = rhs.contentEncoding; + headers = clonedHeaders(rhs.headers); + deliveryMode = rhs.deliveryMode; + priority = rhs.priority; + correlationId = rhs.correlationId; + replyTo = rhs.replyTo; + expiration = rhs.expiration; + messageId = rhs.messageId; + timestamp = rhs.timestamp; + type = rhs.type; + userId = rhs.userId; + appId = rhs.appId; + } + + return *this; +} + bsl::ostream& Properties::print(bsl::ostream& stream, BSLA_MAYBE_UNUSED int level, BSLA_MAYBE_UNUSED int spacesPerLevel) const diff --git a/src/rmq/rmqt/rmqt_properties.h b/src/rmq/rmqt/rmqt_properties.h index 42b81429..c42c79f1 100644 --- a/src/rmq/rmqt/rmqt_properties.h +++ b/src/rmq/rmqt/rmqt_properties.h @@ -115,6 +115,17 @@ struct Properties { /// creating application id bdlb::NullableValue appId; + Properties(); + + /// Copy `original`, giving the copy its own header table rather than + /// sharing `original`'s, which a member-wise copy of the shared pointer + /// would do. + Properties(const Properties& original); + + ~Properties(); + + Properties& operator=(const Properties& rhs); + bsl::ostream& print(bsl::ostream& stream, int level, int spacesPerLevel) const; }; diff --git a/src/tests/rmqa/rmqa_producerimpl.t.cpp b/src/tests/rmqa/rmqa_producerimpl.t.cpp index 82192269..160bfee2 100644 --- a/src/tests/rmqa/rmqa_producerimpl.t.cpp +++ b/src/tests/rmqa/rmqa_producerimpl.t.cpp @@ -87,6 +87,25 @@ class MockConfirmCallback : public ConfirmCallback { const rmqt::ConfirmResponse& confirmResponse)); }; +const char k_INJECTED_KEY[] = "injected-header"; +const char k_INJECTED_VALUE[] = "injected-value"; + +/// Stands in for a tracing implementation which injects a header in place, as +/// the real hooks do. +bsl::shared_ptr +tagWithHeader(rmqt::Properties* properties, + const bsl::string&, + const bsl::string&, + const bsl::shared_ptr&) +{ + if (!properties->headers) { + properties->headers = bsl::make_shared(); + } + (*properties->headers)[k_INJECTED_KEY] = bsl::string(k_INJECTED_VALUE); + + return bsl::make_shared(); +} + MATCHER_P(ExchangeHandleNameEq, expected, "") { bsl::shared_ptr exch = arg.lock(); @@ -248,6 +267,43 @@ TEST_P(ProducerImplTests, PublishNotMandatory) d_threadPool.drain(); } +TEST_P(ProducerImplTests, SendDoesNotShareHeaderTableWithCaller) +{ + // send is asynchronous, so the message handed to the channel must not + // share a header table with the one the caller still owns + + bsl::shared_ptr callerHeaders( + bsl::make_shared()); + callerHeaders->insert( + bsl::make_pair(bsl::string("appheader"), bsl::string("before"))); + + rmqt::Message message(bsl::make_shared >(5)); + message.properties().headers = callerHeaders; + + EXPECT_CALL(*d_mockSendChannel, setCallback(_)); + bsl::shared_ptr producer(d_factory->create( + 1, d_exchange, d_mockSendChannel, d_threadPool, d_eventLoop)); + + rmqt::Message published; + EXPECT_CALL(*d_mockSendChannel, + publishMessage(_, bsl::string("routingKey"), _)) + .WillOnce(SaveArg<0>(&published)); + + producer->send(message, "routingKey", d_callback, d_timeout); + d_threadPool.drain(); + + // the caller carries on using the table it owns, as it is entitled to + (*callerHeaders)["appheader"] = bsl::string("after"); + callerHeaders->insert( + bsl::make_pair(bsl::string("extra"), bsl::string("value"))); + + ASSERT_TRUE(published.headers()); + EXPECT_THAT(published.headers().get(), Ne(callerHeaders.get())); + EXPECT_THAT(published.headers()->size(), Eq(1u)); + EXPECT_TRUE(published.headers()->find("appheader")->second == + rmqt::FieldValue(bsl::string("before"))); +} + TEST_P(ProducerImplTests, DuplicateMessagesReturnDuplicate) { // Ensure sending two msgs to a producer with the same GUID will return @@ -916,6 +972,41 @@ TEST_P(TracingTaggerTests, SendWithMandatoryFlagConfirmCallsTracing) d_threadPool.drain(); } +TEST_P(TracingTaggerTests, TracingDoesNotMutateCallerHeaders) +{ + // tracing hooks inject into the headers in place, so they must be handed + // a table the library owns rather than the caller's + + bsl::shared_ptr callerHeaders( + bsl::make_shared()); + callerHeaders->insert( + bsl::make_pair(bsl::string("appheader"), bsl::string("value"))); + + rmqt::Message message(bsl::make_shared >(5)); + message.properties().headers = callerHeaders; + + bsl::shared_ptr producer(d_factory->create( + 1, d_exchange, d_mockSendChannel, d_threadPool, d_eventLoop)); + + rmqt::Message published; + EXPECT_CALL(*d_mockSendChannel, + publishMessage(_, bsl::string("routingKey"), _)) + .WillOnce(SaveArg<0>(&published)); + // Invoke() is required, the gmock on some of our platforms has no + // implicit conversion from a function pointer to an Action + EXPECT_CALL(*d_tracing, createAndTag(_, _, _, _)) + .WillOnce(Invoke(&tagWithHeader)); + + producer->send(message, "routingKey", d_callback, d_timeout); + d_threadPool.drain(); + + ASSERT_TRUE(published.headers()); + EXPECT_THAT(published.headers()->count(k_INJECTED_KEY), Eq(1u)); + + EXPECT_THAT(callerHeaders->count(k_INJECTED_KEY), Eq(0u)); + EXPECT_THAT(callerHeaders->size(), Eq(1u)); +} + RMQTESTUTIL_TESTSUITE_P(AllMembers, ProducerImplTests, Values(PRODUCER, TRACING_PRODUCER), diff --git a/src/tests/rmqamqpt/rmqamqpt_basicproperties.t.cpp b/src/tests/rmqamqpt/rmqamqpt_basicproperties.t.cpp index 8cb2d2f2..b2081453 100644 --- a/src/tests/rmqamqpt/rmqamqpt_basicproperties.t.cpp +++ b/src/tests/rmqamqpt/rmqamqpt_basicproperties.t.cpp @@ -197,7 +197,8 @@ TEST(Methods_BasicProperties, Headers) basicProps.setProperties(properties); EXPECT_THAT(basicProps.propertyFlags(), Eq(0x2000)); - EXPECT_THAT(basicProps.headers(), Eq(table)); + ASSERT_TRUE(basicProps.headers()); + EXPECT_THAT(*basicProps.headers().value(), Eq(*table)); EXPECT_FALSE(basicProps.contentType()); EXPECT_FALSE(basicProps.contentEncoding()); diff --git a/src/tests/rmqt/CMakeLists.txt b/src/tests/rmqt/CMakeLists.txt index 91a92e36..4fd8f4aa 100644 --- a/src/tests/rmqt/CMakeLists.txt +++ b/src/tests/rmqt/CMakeLists.txt @@ -7,6 +7,7 @@ add_executable(rmqt_tests rmqt_future.t.cpp rmqt_message.t.cpp rmqt_plaincredentials.t.cpp + rmqt_properties.t.cpp rmqt_secureendpoint.t.cpp rmqt_simpleendpoint.t.cpp ) diff --git a/src/tests/rmqt/rmqt_properties.t.cpp b/src/tests/rmqt/rmqt_properties.t.cpp new file mode 100644 index 00000000..a8eb8260 --- /dev/null +++ b/src/tests/rmqt/rmqt_properties.t.cpp @@ -0,0 +1,124 @@ +// Copyright 2020-2023 Bloomberg Finance L.P. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include + +#include + +#include + +#include + +#include +#include +#include + +using namespace BloombergLP; +using namespace BloombergLP::rmqt; + +namespace { + +/// Every field set to a distinct non-default value, so that a field the copy +/// constructor or assignment operator forgets shows up as an inequality. +Properties fullyPopulated() +{ + Properties properties; + properties.contentType = bsl::string("application/json"); + properties.contentEncoding = bsl::string("gzip"); + properties.headers = bsl::make_shared(); + properties.headers->insert( + bsl::make_pair(bsl::string("key"), bsl::string("original"))); + properties.deliveryMode = DeliveryMode::PERSISTENT; + properties.priority = 7; + properties.correlationId = bsl::string("correlation-id"); + properties.replyTo = bsl::string("reply-to"); + properties.expiration = bsl::string("60000"); + properties.messageId = bsl::string("message-id"); + properties.timestamp = bdlt::Datetime(2026, 9, 8, 12, 30, 15); + properties.type = bsl::string("type"); + properties.userId = bsl::string("user-id"); + properties.appId = bsl::string("app-id"); + return properties; +} + +} // namespace + +TEST(PropertiesTests, CopyConstructorCopiesEveryField) +{ + const Properties original = fullyPopulated(); + const Properties copy(original); + + EXPECT_EQ(copy, original); +} + +TEST(PropertiesTests, CopyConstructorClonesHeaders) +{ + Properties original = fullyPopulated(); + Properties copy(original); + + EXPECT_NE(copy.headers.get(), original.headers.get()); + + (*copy.headers)["key"] = bsl::string("changed"); + + EXPECT_TRUE((*original.headers)["key"] == + FieldValue(bsl::string("original"))); +} + +TEST(PropertiesTests, CopyConstructorHandlesAbsentHeaders) +{ + Properties original = fullyPopulated(); + original.headers.reset(); + ASSERT_FALSE(original.headers); + + const Properties copy(original); + + EXPECT_FALSE(copy.headers); + EXPECT_EQ(copy, original); +} + +TEST(PropertiesTests, AssignmentCopiesEveryField) +{ + const Properties original = fullyPopulated(); + Properties assigned; + assigned = original; + + EXPECT_EQ(assigned, original); +} + +TEST(PropertiesTests, AssignmentClonesHeaders) +{ + Properties original = fullyPopulated(); + Properties assigned; + assigned = original; + + EXPECT_NE(assigned.headers.get(), original.headers.get()); + + (*assigned.headers)["key"] = bsl::string("changed"); + + EXPECT_TRUE((*original.headers)["key"] == + FieldValue(bsl::string("original"))); +} + +TEST(PropertiesTests, SelfAssignmentKeepsHeaders) +{ + Properties properties = fullyPopulated(); + Properties& alias = properties; + + properties = alias; + + ASSERT_TRUE(properties.headers); + EXPECT_TRUE((*properties.headers)["key"] == + FieldValue(bsl::string("original"))); +}