So far, the customer support team at CCM has been relying only on e-mails to interact with individual users. They've recently added the RabbitMQ-powered public address system discussed in Chapter 3, Switching to Server-push. Now that direct user messages can get routed to users by e-mails, they're interested in the possibility of sending such messages to individual users from the back office application. Furthermore, if possible they would like users who don't have an inbox queue on RabbitMQ to get the message e-mailed to them immediately instead of having to wait for the seven days' TTL.
In terms of messaging architecture, you're in a known territory; this is the exact same model as the one you put in place in Chapter 2, Creating an Application Inbox, for user-to-user messages, as illustrated in the following screenshot. The only difference is that, unlike the main application, the back office will not create and bind a user queue prior to sending a message. Instead, the back office will have to somewhat detect that no such queue pre-exists and revert to an e-mail delivery for the message.
The back office will use the user-inboxes direct exchange for direct messages
What's unclear is how to achieve the second part of the requirements: how can the back office check the pre-existence of a queue? The AMQP specification doesn't define a direct way to do this. The RabbitMQ management plugin exposes a REST API that could be used to check the existence of a queue; it's a tempting approach, but you'd rather stay within the confines of what AMQP offers by default. Moreover, this could expose you to a check then act type of race condition. Indeed, the queue could be created by another process after you verify that it doesn't exist. So, after digging deeper into the AMQP specification, you're excited to discover a feature that allows you to achieve your goal in a safe and elegant manner. This feature is called mandatory delivery.
When a message is published on an exchange with the mandatory flag set to true, it will be returned by RabbitMQ if the message cannot be delivered to a queue. A message cannot be delivered to a queue either because no queue is bound to the exchange, or because none of the bound queues have a routing key that would match the routing rules of the exchange. In the current case, it would mean that no user inbox queue is bound to a routing key that matches the addressee's user ID.
Note
AMQP defines another delivery-related flag named immediate, which is a step beyond mandatory in the sense it ensures that the message has been actually delivered to a consumer. RabbitMQ has chosen not to support this feature for it is nontrivial to implement efficiently and elegantly, especially in clustered environments.
The trick about returned messages is that RabbitMQ doesn't return them synchronously as a response to the publish operation; it returns them in an asynchronous fashion. This means that for the developer, a specific message handler will have to be registered with RabbitMQ in order to receive the returned messages. This leads to the overall architecture illustrated in the following diagram:
A dedicated handler takes care of returned messages
Note
If you come from a JMS background, you're probably wondering about transactional delivery. AMQP supports the notion of transactions on a per-channel basis, both for consumers and producers. That said, it comes with several nontrivial "gotchas", due to a certain lack of clarity in the specification. Generally speaking, the acknowledgment/rejection mechanism is preferred for transactions.
With this decided, it's now time to add the necessary code in the back office.
In CCM's back office, you're going to add a method very similar to the one used for the public address messaging system (seen in Chapter 3, Switching to Server-push); however, this time we will add support for returned messages. The Ruby client library supports this feature very elegantly as you'll soon find out. The following code is the code you need to add to support the mandatory delivery of messages to user inboxes and to handle potentially returned messages. The method that takes care of e-mailing the messages to the users has not been included for brevity, as follows:
channel = AMQP::Channel.new(connection)
exchange = channel.direct(
'user-inboxes',
:durable => true,
:auto_delete => false) do |exchange, declare_ok|
exchange.on_return do |basic_return, metadata, payload|
email_message_to_user(payload)
end
message_id = SecureRandom.uuid
message_json = JSON.generate({
:time_sent => (Time.now.to_f*1000).to_i,
:sender_id => -1, # special value for support
:addressee_id => user_id,
:subject => 'Direct message from CS',
:content => 'A private message from customer support...'})
routing_key = "user-inbox.#{user_id}"
exchange.publish(
message_json,
:routing_key => routing_key,
:content_type => 'application/vnd.ccm.pmsg.v1+json',
:content_encoding => 'UTF-8',
:message_id => message_id,
:persistent => true,
:nowait => false,
:mandatory => true) do
puts "Published message ID: #{message_id} to: #{routing_key}"
end
endIn case you don't recall the code used for sending to the public address fanout exchange, we highlighted the lines that differ in this code. In short, the main notable points are as follows:
- The
user-inboxesdirect exchange is used for message publication - Returned messages are handled through a closure registered via
exchange.on_return - The message JSON payload now defines an
addressee_idfield, which is required by the schema for direct user messages - Messages are published with the required
routing_keyto target a particular user inbox and with themandatoryflag set totrue
That is all! The feature is ready to go live. As you can see, the major "gotcha" was in understanding that messages are returned asynchronously and need to be handled that way.