Consume all messages using MassTransit

Question

I'm currently building a system using MassTransit and RabbitMQ as my messaging layer. I'm trying to find a way to have a Consumer that listens on all messages of all types on the bus. This is for our audit logging framework and we want to log all of the events going across the message bus.

Is there a way to do this in MassTransit?

Answer 1

You would need to add some type of auditing interface to your message that could be subscribed for auditing purposes. For example, if you were to create a base interface:

public interface IAuditable
{
    DateTime Timestamp {get;}
    string Username {get}
}

Or whatever properties must be commonly available for auditing. Then you can subscribe to that interface and get a copy of every message. Or you could make it an empty interface and just audit message headers. But the messages would need to implement it and publish it to get a copy.

This seems like a generally bad idea, since you're creating copies of the messages all over the place...

Another approach would be to add an observer to message consumption and use that observer to either write to the audit storage or to send a message to an audit queue and let that asynchronous consumer to write to the audit storage.

The thing is, if you're auditing every message, and every message is sending an audit message, make sure you don't observer your audit consumer or you'll die the infinite death.

The observer option is my favorite, since it not only logs the message, but allows the disposition (success/fault) to be captured, as well as the host which consumed the message, processing duration, etc.

Answer 2

MassTransit has build in support for auditing
See this link: https://masstransit-project.com/advanced/audit.html

So you're better use their built in functionallity instead of creating observers and other hacks

Two main parts need to be saved for each message to provide complete audit:

• The message itself
• Metadata

Message metadata includes:

• Message id
• Message type
• Context type (Send, Publish or Consume)
• Conversation id
• Correlation id
• Initiator id
• Request id (for request/response)
• Source address
• Destination address
• Response address (for request/response)
• Fault address