Propagate NServiceBus headers to SQS message attribute

[jpalac]

Background

Amazon SQS allows for the inclusion of structured metadata with messages using message attributes. Currently the NServiceBus.AmazonSQS transport stores all message headers as part of the message body - message attributes are not used.
It may be advantageous to store all/some of the NSB headers as message attributes as a consumer can use these to handle a message in a particular way without having to process the message body first.

Limitations

• A maximum of 10 message attributes is allowed per message.
• There is no limit on the size of a single attribute.
• All components of a message attribute are included in the 256 KB message size restriction. This means that any of the below modification options would need to be done before the uplift to S3 because the size of the message will change.

Possible implementaions

Copy all NServiceBus headers into an attribute

For all outgoing messages, we could serialize the NServiceBus headers dictionary and store it inside a single SQS message attribute (NServiceBus.Headers). When we receive a message with that attribute, we know it was sent from an NServiceBus endpoint and can restore the headers.

sqsTransport.SendHeadersViaMessageAttribute();

To ensure we don’t hit the size limitations of the transport, we should remove the headers from the outgoing message body. In fact, we might be able to drop the TransportMessage wrapper altogether and simply send the serialized message body as the SQS Message body.

Scenarios:

• Old sender, new receiver: Not a problem. The new message attribute will not be sent.
• New sender, new receiver: Not a problem. We know how to handle the new message attribute.
• New sender, old receiver: This is a problem. The old receiver will not know how to retrieve the NServiceBus headers. We could include the MessageType header that we use for native messages, which would allow the old receiver to process the message, but it will not be able to operate on the headers and could lead to subtle bugs.
  • We might be able to give customers a custom pipeline behavior that extracts those message headers and sets them properly.

With these scenarios in mind, this API is only safe if all of the endpoints are able to handle the new attribute in some way.

Propagate individual headers into message attributes

We could provide a configuration option to uplift specific NServiceBus headers into SQS message attributes. This would allow customers to create filters that look at the attributes they care about.

sqsTransport.SendHeaderViaMessageAttribute(Headers.ReplyTo);
sqsTransport.SendHeaderViaMessageAttribute(Headers.CorrelationId);

OR:

sqsTransport.SendHeadersViaMessageAttribute(Headers.ReplyTo, Headers.CorrelationId);

To meet the size limitations of the transport, we may want to consider removing the header from the NServiceBus headers sent on the transport wrapper. If we do that we run into all the same issues listed above.

Custom extension point to modify SQS message

Endpoint level

Add an extension point to allow modification of the SQS message after NServiceBus has created it.
This would allow someone to manually write code that propagates NServiceBus headers to whatever they want.

// MessageId is already in the headers if you really want it
sqsTransport.ModifyOutgoingMessages((headers, body, sqsMessage) => 
{
  sqsMessage.MessageAttributes[Headers.ReplyTo] = headers[Headers.ReplyTo];
});

This option provides maximum flexibility but is easy to use incorrectly. It might also be difficult to plug this in before the message size calculation and hence S3 uplift.

Message level

We could provide a similar outgoing message access to what we have done on the ASB transport.
This would require it being set on every relevant message send, but it may be easier to plug in than an endpoint level solution. It would however mean that this would only work when the outbox is not used.

[jpalac]

This was discussed as part of the Platform Enhancement TF.
It was decided that since there's no specific use case for this, there's no point in trying to overengineer a generic solution - best to wait for a specific requirement and try to solve that instead.

[mauroservienti]

Re-opened in light of the spike in #1931

[mauroservienti]

Spiked in #1962. We decided not to proceed for now with this change and instead wait for a more concrete scenario.