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..049cd65a 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` 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 > 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/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 be481c45..160bfee2 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 @@ -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(); @@ -166,11 +185,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(); } @@ -245,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 @@ -823,7 +882,7 @@ TEST_P(ProducerImplMaxOutstandingTests, SendFromConfirmCallbackDoesNotDeadlock) d_threadPool.drain(); } -class TracingProducerImplTests : public ProducerImplMaxOutstandingTests { +class TracingTaggerTests : public ProducerImplMaxOutstandingTests { public: }; @@ -832,7 +891,7 @@ MATCHER_P(MessagePropertiesMatch, expected, "") return arg.properties() == expected; } -TEST_P(TracingProducerImplTests, SendConfirmCallsTracing) +TEST_P(TracingTaggerTests, SendConfirmCallsTracing) { // GIVEN bsl::shared_ptr tracingContext( @@ -867,13 +926,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( @@ -914,6 +972,41 @@ TEST_P(TracingProducerImplTests, 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), @@ -936,6 +1029,6 @@ RMQTESTUTIL_TESTSUITE_P(AllMembers, ProducerImplTests::PrintParamName()); RMQTESTUTIL_TESTSUITE_P(AllMembers, - TracingProducerImplTests, + TracingTaggerTests, Values(TRACING_PRODUCER), ProducerImplTests::PrintParamName()); 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"))); +}