pub/sub with transactions and error handling with MassTransit

Question

Please help getting me started

I have been surfing my ass of the hole weekend, sorry my French, without finding what I am looking for.

This is the process I am trying to move to a message queue

• Every night pdf files are landing on our ftp server, around 45,000.
• For each file a record is created in ms sql server in a queue table
• A service is reading the queue and processing the files on disk.

This is what I want pub/sub with transaction’s and error handling

• The file-watcher writes to a queue in a transaction instead of the database. This I guess is faster.
• I then want to subscribe to the pdf- queue so I don’t have to poll the database when a new file arrives.

some question’s

• How do I publish to the queue with transactions?
• How do I retrieve a message from the queue and then abort if there is an error when processing the message.
• What is the best way to open / close queue connections? Is there a light weight sessions object?

And yes I know there is a simple pub/sub example :-) but I am unable to glue it all together

Any help is much appreciated.

Answer 1

How do I publish to the queue with transactions?

If the queue is transactional, when configuring your bus, use msmq://machine/queue_name?tx=true to ensure MassTransit knows it's a transactional queue. Now this means all machines involved need to be able to enroll in DTC. This may or may not be trivial to setup. This assumes the use of MSMQ. I'd suggest using RabbitMQ unless you need to enroll in the distributed transaction.

How do I retrieve a message from the queue and then abort if there is an error when processing the message.

MassTransit supports auto-retries (5x?) on MSMQ transactional queues if an error is caught. You can also catch the error and a RetryLater() to toss it back on the queue.

What is the best way to open / close queue connections? Is there a light weight sessions object?

If you are using MassTransit, MT handles all the connection objects for you. No need to open/close connections to the queues.

Really, you can do a Bus.Instance.Publish(new FileArrivedMessage(filename)) on your file listener service; and on the Consumer side just register a Consumes<FileArrivedMessage>.All implementation. They each need their own queue but should be able to communicate pretty easily.

You can always get more help from the MT mailing list as well. https://groups.google.com/forum/#!forum/masstransit-discuss