Overload the send function in TracingProducerImpl - #91
Merged
Conversation
Rahmeen14
marked this pull request as ready for review
August 18, 2026 13:47
flamble
approved these changes
Aug 18, 2026
Rahmeen14
enabled auto-merge (squash)
August 18, 2026 14:10
flamble
added a commit
that referenced
this pull request
Sep 9, 2026
### Problem statement This fixes an issue with `rmqt::Properties`, which holds its headers as a `bsl::shared_ptr<rmqt::FieldTable>`, not having it's own copy before enqueuing a send or modifying them for tracing. ### Proposed changes #### 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 #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 #### Give copies of Properties their own header table `rmqt::Properties` holds its headers as a `bsl::shared_ptr<rmqt::FieldTable>`, 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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The send() overload accepting an explicit mandatory flag must also exist. Previously TracingProducerImpl only overrode the four-argument send(), so publishing with an explicit mandatory flag silently bypassed tracing.