Chapter 7. Immutant Messaging

Immutant encapsulates the JBoss HornetQ message broker. It is automatically available to you, with no additional configuration required to start the messaging service. HornetQ supports clustered messaging, which provides auto-discovery, load-balancing, and failover, among other things.

Typically, messages published to a topic are only delivered to consumers connected to the topic at that time. But if a consumer identifies itself with a unique name, then the broker will accumulate messages for that client when it's disconnected and deliver them in order whenever the client reconnects.

You implicitly create a durable topic subscriber by passing the :client-id option to the receive and listen functions, described below. As long as you pass the same :client-id in subsequent receive/listen calls, you'll receive every message published to that topic, whether you were connected at the time or not.

Use the immutant.messaging/start function to define a messaging destination. A simple naming convention designates the destination as either a queue or a topic: if its name contains "queue", it's a queue; if it contains "topic", it's a topic. If you need to use queue or topic names that don't match the above convention, you can wrap the name in a call to immutant.messaging/as-queue or immutant.messaging/as-topic respectively. If you do wrap a destination name, you'll need to pass the wrapped version to any immutant.messaging function that takes a destination name. It's common to separate sections of the destination name with . or /, but is not required.

The start function accepts the following options:

If a :selector is provided, then only messages with metadata/properties matching that expression will be accepted for delivery.

Some examples:

(require '[immutant.messaging :as msg])

;; to start queues - these are all valid names
(msg/start "/queue/work")
(msg/start ".queue.play")
(msg/start "queue.sing")
(msg/start "dancequeued")

;; wrap an ambiguous queue name in as-queue
(msg/start (msg/as-queue "no-q-word"))

;; to start topics - these are all valid names
(msg/start "/topic/news")
(msg/start "topic/infotainment")
(msg/start ".topic")
(msg/start "topicality")
(msg/start "topicality")
(msg/start "some.kinda.topic")

;; wrap an ambiguous topic name in as-topic
(msg/start (msg/as-topic "no-t-word"))

;; only blue messages will be queued for delivery
(msg/start "/queue/blue" :selector "color = 'blue'")
;; like this one...
(msg/publish "/queue/blue" "success" :properties {:color "blue"})

Starting a destination is dynamic, and can occur anywhere in your application code. You can invoke start anytime during the lifecycle of your application.

While start has a complement, immutant.messaging/stop, you needn't call it directly. It will be invoked when your application is undeployed. And it's important to note that start is idempotent: if an endpoint has already been started, likely by a cooperating application, the call is effectively a no-op. Similarly, a call to stop will silently fail if the endpoint is in use by any other application.

Each messaging destination has associated controllers that can be used to see message counts, list & remove messages, and perform other operations. Two different controllers are provided by HornetQ for queues, and one for topics, each with slightly different controls. All are available via JMX or from immutant.messaging.hornetq/destination-controller.

For a given destination name (queue.example and topic.example in this case), you can access the MBeans via JMX with the following addresses:

# to access the JMS Queue MBean
org.hornetq:module=JMS,type=Queue,name="queue.example"

# to access the JMS Topic MBean
org.hornetq:module=JMS,type=Topic,name="topic.example"

# to access the HornetQ 'core' Queue MBean
org.hornetq:module=Core,type=Queue,address="jms.queue.queue.example",name="jms.queue.queue.example"

or via code with:

(require '[immutant.messaging.hornetq :as hq])

;; for the JMS queue controller
(hq/destination-controller "queue.example")

;; for the JMS topic controller
(hq/destination-controller "queue.topic")

;; for the HornetQ 'core' queue controller
(hq/destination-controller "queue.example" :core)

The returned controller depends on the type of the given destination and, for queues, the requested control-type (which defaults to :jms):

Any component or client code that creates messages and gives them to the message broker for delivery is considered a producer. Generally speaking, the producer does not know the details of the destination or any of its consumers.

In Immutant, there is only one way to send a message, whether to a queue or a topic: via the immutant.messaging/publish function. It accepts two required parameters: the name of the destination and the message content, which can be just about anything.

If the message has any Clojure metadata attached, it will be passed as the JMS Message's properties, the names of which are subject to certain naming restrictions (they must be valid Java identifiers) since they may be used as expressions in selectors (see below). It's also possible to pass properties via the :properties option, which will override any matching keys in the payload metadata, if present.

The publish function accepts the following options:

The :json and :edn encodings are useful when the message consumers aren't written in Clojure. For example, TorqueBox Ruby processors will automatically convert edn-encoded messages generated by a Clojure function into their analogous Ruby data structures, so as long as you limit the content of your messages to standard collections and types, they are transparently interoperable between Clojure and Ruby in either direction.

Any component that waits for messages to be delivered to it by the message broker is consider a consumer. Typically, a consumer is unaware of the producer or any other consumers.

If the published message payload contains metadata, the received message should have it, too, transferred in the form of JMS properties, subject to any overridden values passed in the :properties option (see above). If the payload cannot accept metadata, the message properties can be converted to a convenient Clojure hash using immutant.messaging.core/get-properties.

Immutant features three functions for consuming messages.

• immutant.messaging/receive Blocks the caller until a message arrives and returns the decoded message
• immutant.messaging/message-seq Creates a lazy sequence of messages
• immutant.messaging/listen Register a handler function that will receive the decoded message when it arrives

Both receive and message-seq expect the destination name as the first parameter, and optionally, the following key/value pairs:

By default, the dynamic variable, clojure.core/*read-eval* is set to false when decoding messages. To override this, you should set :decode? to false and bind *read-eval* to true before passing the encoded message to immutant.messaging.codecs/decode-with-metadata yourself.

For more details on message selectors, see javax.jms.Message.

The listen function takes two parameters: the destination name and a function accepting one parameter which will be applied to any received message. All of the above options except :timeout are supported, plus listen also accepts the following:

listen is asynchronous - if you need to synchronize on its completion, you should deref the result.

By default, message handlers are transactional, so the function invoked in response to a message effectively demarcates a transaction that will be automatically committed if no exceptions are raised in the handler, and otherwise rolled back.

Any messages published within the handler automatically become part of its transaction, by default. So they won't be delivered until that transaction commits. To override this behavior, wrap your handler inside the immutant.xa.transaction/not-supported macro.

See Distributed Transactions for more details.

The immutant.messaging/request function takes a queue, a message, and an optional list of options. It publishes the message to the queue, marking it as a synchronous message and returns a delay that will receive the response from the worker initiated by the respond function. It accepts the same options as publish.

The immutant.messaging/respond method takes a queue, a function, and an optional list of options. It sets up a listener (via the listen function) that applies the given function to any received message and publishes the result back to the queue for the delay returned by request to receive. It accepts the same options as listen.

A basic example:

(require '[immutant.messaging :as msg])

;; setup a responder
(msg/respond "/queue/work" (partial apply +))

;; send a request
(let [result (msg/request "/queue/work" [1 2 3])]
  (println (deref result 1000 nil)) ;; => 6

An example of using properties and selectors to segment work on the same queue:

(require '[immutant.messaging :as msg])

;; respond to 'add' messages
(msg/respond "/queue/work" (partial apply +) :selector "operation='add'")

;; respond to 'multiply' messages
(msg/respond "/queue/work" (partial apply *) :selector "operation='multiply'")

(deref
 (msg/request "/queue/work" [1 2 3 4] :properties {"operation" "add"})
 1000 nil) ;; => 9

(deref
 (msg/request "/queue/work" [1 2 3 4] :properties {"operation" "multiply"})
 1000 nil) ;; => 24

You create a pipeline with the immutant.pipeline/pipeline function. The pipeline function takes a unique (within the scope of the application) name, one or more single-arity functions, and optional keyword argument options, returning a function that acts as an entry point into the pipeline:

(require '[immutant.pipeline :as pl])

(defonce foo-pipeline
  (pl/pipeline "foo"
    function-that-does-something
    another-function))

The function returned by pipeline acts as an entry function, placing its argument onto the pipeline when called, returning a delay around the end of the pipeline (by default):

(require '[immutant.pipeline :as pl])

(defonce foo-pipeline
  (pl/pipeline "foo"
    function-that-does-something
    another-function))

(deref (foo-pipeline {:ham :biscuit}) 10000 :timeout!)

Pipelines store the result of an execution by default, allowing it to be retrieved by dereferencing the delay returned by the pipeline-fn call. To prevent results that may not be retrieved from being stored indefinitely, they have a default time-to-live of 1 hour. You can control the retention time by passing a :result-ttl option to pipeline. It is specified in milliseconds, with a value of 0 indicating that the result should be saved indefinitely, and -1 indicating that the results should be discarded immediately. If you set the :result-ttl to -1, any attempt to dereference the returned delay will raise an error.

If the result from a step is reference, it will be dereferenced before being passed to the next step. This allows you to use a pipeline within a pipeline. The amount of time to wait for the deref is controlled by the :step-deref-timeout option, and defaults to 10 seconds. Setting it to 0 will cause it to wait forever, which will tie up a thread

(require '[immutant.pipeline :as pl])

(defonce pipeline-x
  (pl/pipeline :x
    function-that-does-something
    another-function))

(defonce pipeline-y
  (pl/pipeline :y
    yet-another-function
    pipeline-x
    and-another
    :step-deref-timeout 60000))

By default, the pipeline entry function places its argument onto the front of the pipeline. You can insert the data into the pipeline at any step by passing a :step keyword argument. The step name would be the name you provided as an option for that step using the step function, or the index of the step in the list of steps if you haven't provided a name:

(require '[immutant.pipeline :as pl])

(defonce foo-pipeline
  (pl/pipeline "foo"
    function-that-does-something
    (pl/step another-function :name :another)
    a-third-function))

;; insert at head
(foo-pipeline {:ham :biscuit})

;; skip the first step
(foo-pipeline {:ham :biscuit} :step :another)

;; insert at the last step 
(foo-pipeline {:ham :biscuit} :step 2)

The following vars have bound values inside a step or error-handler invocation:

When an exception occurs in a step function, an error-handler function will be invoked if provided for the pipeline or for the particular step. This function will be passed the exception and the original data passed to the step function, and have all of the above bindings available:

;; a naive error handler that sleeps then retries on a network error,
;; logging and discarding otherwise
(defn error-handler [ex data]
  (if (instance? NoRouteToHostException ex)
    (do
      (Thread/sleep 1000)
      (pl/*pipeline* data :step pl/*current-step*))
    (println "ERROR:" ex)))

(require '[immutant.pipeline :as pl])

(pl/pipeline "foo"
  connects-to-foo
  connects-to-bar
  :error-handler error-handler)

If no error-handler function is provided, the error handling semantics provided my HornetQ are used, which causes the offending step to be retried up to ten times before giving up.

If, in a step function, you determine that the data requires no further processing, you can halt that particular pipeline execution by returning a special flag symbol - immutant.pipeline/halt:

(require '[immutant.pipeline :as pl])

;; halt the pipeline at the second step, causing another-function to
;; not be called
(pl/pipeline "foo"
  function-that-does-something
  #(if (:some-done-condition %)
     pl/halt
     %)
  another-function)

When your application is undeployed, Immutant will automatically shut down the pipeline. If you need to stop the pipeline at runtime, use the immutant.pipeline/stop function:

(require '[immutant.pipeline :as pl])

(let [pipeline (pl/pipeline "foo" ...)]
  ...
  (pl/stop pipeline))