<commit_before>#+TITLE:     Immutant Messaging

* Introduction
** HornetQ

   Immutant encapsulates the JBoss [[http://www.jboss.org/hornetq/][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.

   The term "messaging" encompasses a large area of functionality.
   Messaging solutions are used to achieve loosely-coupled,
   asynchronous systems. The primary actors in a messaging-based
   system are messages, destinations, consumers, and producers. The
   broker mediates the relationships between the other actors.

* Destinations

  A destination represents a rendezvous point where messages are
  exchanged. A message may be sent to a destination by one actor,
  and received from the destination by another.

  There are two main types of destinations: *queues* (point-to-point)
  and *topics* (publish-subscribe). All destinations allow multiple
  actors to send messages. The type of destination determines how the
  message is routed. A queue delivers the message to a single
  recipient (possibly one of many candidates). And a topic delivers
  the message to all interested recipients. In both cases, the message
  producers have no direct knowledge of the message consumers.

** Deployment

   Use the [[./apidoc/immutant.messaging-api.html#immutant.messaging/start][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 matches =#"^.?queue"=,
   it's a queue; if it matches =#"^.?topic"=, it's a topic. It's common
   to separate sections of the queue name with =.= or =/=, but is not 
   required.

   #+begin_src clojure
     (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 "queuedance")
     
     ;; to start topics - these are all valid names
     (msg/start "/topic/news")
     (msg/start "topic/infotainment")
     (msg/start ".topic")
     (msg/start "topicality")
   #+end_src

   You can invoke =start= from anywhere in your application, but
   typically it's done in the [[./initialization.html#initialization-immutant-clj][immutant.clj]] initialization file.

   While =start= has a complement, [[./apidoc/immutant.messaging-api.html#immutant.messaging/stop][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.

* Messages

  The unit of communication within a messaging system is a message.  A
  message may either be simply a blob of octets, or it might have some
  higher-order, application-defined semantics. All messages include a
  set of headers, similar to email.

** Publishing

   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 [[./apidoc/immutant.messaging-api.html#immutant.messaging/publish][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 [[http://clojure.org/metadata][Clojure metadata]] attached, it will be passed
   as [[http://docs.oracle.com/javaee/6/api/javax/jms/Message.html][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:

   | Option            | Default                        | Description                         |
   |-------------------+--------------------------------+-------------------------------------|
   |                   | <c>                            | <35>                                |
   | =:encoding=       | =:clojure=                     | One of =:clojure=, =:json= or =:text= |
   | =:priority=       | =4=                            | An integer (0-9) or one of =:low=, =:normal=, =:high= and =:critical= which correspond to 0, 4, 7 and 9, respectively |
   | =:ttl=            | =0=                            | An integer greater than 0, indicating the number of milliseconds after which the message is discarded if not consumed. A 0 indicates that the message should be held indefinitely. |
   | =:persistent=     | =true=                         | If true, undelivered messages survive restarts. |
   | =:properties=     | ={}=                           | A hash of arbitrary metadata upon which JMS selector expressions may be constructed to filter received messages |
   | =:correlation-id= | =nil=                          | Used to set the JMSCorrelationID (see [[http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#setJMSCorrelationID(java.lang.String)][setJMSCorrelationID]]) |
   | =:host=           | =nil=                          | A remote HornetQ host to connect to. |
   | =:port=           | =nil=, or =5445= if =:host= is set | A remote HornetQ port to connect to. Requires =:host= to be set. |
   | =:username=       | =nil=                          | The username to authenticate the connection with (if the broker has authentication enabled). Requires =:password= to be set. |
   | =:password=       | =nil=                          | The password to authenticate the connection with (if the broker has authentication enabled). Requires =:username= to be set. |
   #+TBLFM: 

   The =:json= encoding is useful when the message consumers aren't
   written in Clojure. For example, [[http://torquebox.org][TorqueBox]] Ruby processors will
   automatically convert json-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.

*** Some Examples

    #+begin_src clojure
      ;; A simple string
      (msg/publish "/queue/work" "simple string")
      ;; Notify everyone something interesting just happened
      (msg/publish "topic/news" {:event "VISIT" :url "/sales-inquiry"})
      ;; Move this message to the front of the line
      (msg/publish "/queue/work" some-message :priority :high :ttl 1000)
      ;; Make messages as complex as necessary
      (msg/publish "/queue/work" {:a "b" :c [1 2 3 {:foo 42}]})
      ;; Make messages consumable by a Ruby app
      (msg/publish "/queue/work" {:a "b" :c [1 2 3 {:foo 42}]} :encoding :json)
      ;; Publish to a remote broker
      (msg/publish "queue.remote-work" "a message" :host "foo.example.com" :port 5445)
      ;; The received message's metadata => {:foo 42, :bar 1}
      (msg/publish q (with-meta msg {:foo 42 :bar 0}) :properties {:bar 1})
    #+end_src

** Receiving

   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 [[./apidoc/immutant.messaging-api.html#immutant.messaging.core/get-properties][immutant.messaging.core/get-properties]].

   Immutant features three functions for consuming messages.
   - [[./apidoc/immutant.messaging-api.html#immutant.messaging/receive][immutant.messaging/receive]] Blocks the caller until a message arrives and returns
     the decoded message
   - [[./apidoc/immutant.messaging-api.html#immutant.messaging/message-seq][immutant.messaging/message-seq]] Creates a lazy sequence of messages
   - [[./apidoc/immutant.messaging-api.html#immutant.messaging/listen][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:

   | Option      | Default                            | Description                         |
   |-------------+------------------------------------+-------------------------------------|
   |             | <c>                                | <35>                                |
   | =:timeout=  | =10000=                            | An expiration in milliseconds, after which nil is returned; a value of 0 means wait forever |
   | =:selector= | =nil=                              | A JMS expression used to filter messages according to the values of arbitrary =:properties= |
   | =:decode?=  | =true=                             | If true, the decoded message body is returned. Otherwise, the javax.jms.Message object is returned |
   | =:host=     | =nil=                              | A remote HornetQ host to connect to. |
   | =:port=     | =nil=, or =5445= if =:host= is set | A remote HornetQ port to connect to. Requires =:host= to be set. |
   | =:username= | =nil=                              | The username to authenticate the connection with (if the broker has authentication enabled). Requires =:password= to be set. |
   | =:password= | =nil=                              | The password to authenticate the connection with (if the broker has authentication enabled). Requires =:username= to be set. |

   For more details on message selectors, see [[http://docs.oracle.com/javaee/6/api/javax/jms/Message.html][javax.jms.Message]]. 

   The =listen= method takes two parameters: the destination name and
   a function accepting one parameter which will be applied to any
   received message. The following key/value options are also
   available:

   | Option         | Default                            | Description                         |
   |----------------+------------------------------------+-------------------------------------|
   |                | <c>                                | <35>                                |
   | =:concurrency= | =1=                                | The maximum number of listening threads that can simultaneouly call the function |
   | =:selector=    | =nil=                              | Same as =receive=                   |
   | =:decode?=     | =true=                             | If true, the decoded message body is passed to f. Otherwise, the javax.jms.Message object is passed |
   | =:host=        | =nil=                              | A remote HornetQ host to connect to. |
   | =:port=        | =nil=, or =5445= if =:host= is set | A remote HornetQ port to connect to. Requires =:host= to be set. |
   | =:username=    | =nil=                              | The username to authenticate the connection with (if the broker has authentication enabled). Requires =:password= to be set. |
   | =:password=    | =nil=                              | The password to authenticate the connection with (if the broker has authentication enabled). Requires =:username= to be set. |
   #+TBLFM: $2=nil=, or =5445= if =:host= is set

   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 [[./apidoc/immutant.messaging-api.html#immutant.xa.transaction/not-supported][immutant.xa.transaction/not-supported]] macro.

   See [[./transactions.html][Distributed Transactions]] for more details.

*** Some Examples

    #+begin_src clojure
      ;; Wait on a task
      (let [task (msg/receive "/queue/work")]
        (perform task))
      
      ;; Case-sensitive work queues?
      (msg/listen ".queue.lower" #(msg/publish "/queue/upper" (.toUpperCase %)))
      
      ;; Listen to a remote queue
      (msg/listen "queue/remote" #(do-someting %) :host "foo.example.com" :port 5445)
      
      ;; Contrived laziness
      (let [messages (message-seq queue)]
        (doseq [i (range 4)] (publish queue i))
        (= (range 4) (take 4 messages)))
    #+end_src

* Request/Response
   :PROPERTIES:
   :CUSTOM_ID: messaging-request-response
   :END:

  Immutant also provides an implementation of the [[http://en.wikipedia.org/wiki/Request-response][request/response]] pattern 
  for synchronous work distribution. This feature is provided by two 
  cleverly named functions: =request= and =respond=. Since they leverage
  the existing messaging subsystem, the work is automatically distributed
  across multiple workers within the same JVM or across multiple nodes if
  in a cluster.

** Request
   
   The [[./apidoc/immutant.messaging-api.html#immutant.messaging/request][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=, with one additional option:

   | Option     | Default | Description                         |
   |------------+---------+-------------------------------------|
   |            | <c>     | <35>                                |
   | =:timeout= | =10000= | Time in ms for the delayed receive to wait once it it is dereferenced, after which nil is returned |

** Respond

   The [[./apidoc/immutant.messaging-api.html#immutant.messaging/respond][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.

** Some Examples

   A basic example:

   #+begin_src clojure
     (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 @result)) ;; => 6
   #+end_src

   An example of using properties and selectors to segment work on the same 
   queue:
   
   #+begin_src clojure
     (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'")
     
     @(msg/request "/queue/work" [1 2 3 4] :properties {"operation" "add"}) ;; => 9
     
     @(msg/request "/queue/work" [1 2 3 4] :properties {"operation" "multiply"}) ;; => 24
   #+end_src






<commit_msg>Docs for [IMMUTANT-29]<commit_after>#+TITLE:     Immutant Messaging

* Introduction
** HornetQ

   Immutant encapsulates the JBoss [[http://www.jboss.org/hornetq/][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.

   The term "messaging" encompasses a large area of functionality.
   Messaging solutions are used to achieve loosely-coupled,
   asynchronous systems. The primary actors in a messaging-based
   system are messages, destinations, consumers, and producers. The
   broker mediates the relationships between the other actors.

* Destinations

  A destination represents a rendezvous point where messages are
  exchanged. A message may be sent to a destination by one actor,
  and received from the destination by another.

  There are two main types of destinations: *queues* (point-to-point)
  and *topics* (publish-subscribe). All destinations allow multiple
  actors to send messages. The type of destination determines how the
  message is routed. A queue delivers the message to a single
  recipient (possibly one of many candidates). And a topic delivers
  the message to all interested recipients. In both cases, the message
  producers have no direct knowledge of the message consumers.

** Durable Topic Subscribers

   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.

** Deployment

   Use the [[./apidoc/immutant.messaging-api.html#immutant.messaging/start][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 matches =#"^.?queue"=,
   it's a queue; if it matches =#"^.?topic"=, it's a topic. It's common
   to separate sections of the queue name with =.= or =/=, but is not 
   required.

   #+begin_src clojure
     (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 "queuedance")
     
     ;; to start topics - these are all valid names
     (msg/start "/topic/news")
     (msg/start "topic/infotainment")
     (msg/start ".topic")
     (msg/start "topicality")
   #+end_src

   You can invoke =start= from anywhere in your application, but
   typically it's done in the [[./initialization.html#initialization-immutant-clj][immutant.clj]] initialization file.

   While =start= has a complement, [[./apidoc/immutant.messaging-api.html#immutant.messaging/stop][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.

* Messages

  The unit of communication within a messaging system is a message.  A
  message may either be simply a blob of octets, or it might have some
  higher-order, application-defined semantics. All messages include a
  set of headers, similar to email.

** Publishing

   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 [[./apidoc/immutant.messaging-api.html#immutant.messaging/publish][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 [[http://clojure.org/metadata][Clojure metadata]] attached, it will be passed
   as [[http://docs.oracle.com/javaee/6/api/javax/jms/Message.html][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:

   | Option            | Default                        | Description                         |
   |-------------------+--------------------------------+-------------------------------------|
   |                   | <c>                            | <35>                                |
   | =:encoding=       | =:clojure=                     | One of =:clojure=, =:json= or =:text= |
   | =:priority=       | =4=                            | An integer (0-9) or one of =:low=, =:normal=, =:high= and =:critical= which correspond to 0, 4, 7 and 9, respectively |
   | =:ttl=            | =0=                            | An integer greater than 0, indicating the number of milliseconds after which the message is discarded if not consumed. A 0 indicates that the message should be held indefinitely. |
   | =:persistent=     | =true=                         | If true, undelivered messages survive restarts. |
   | =:properties=     | ={}=                           | A hash of arbitrary metadata upon which JMS selector expressions may be constructed to filter received messages |
   | =:correlation-id= | =nil=                          | Used to set the JMSCorrelationID (see [[http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#setJMSCorrelationID(java.lang.String)][setJMSCorrelationID]]) |
   | =:host=           | =nil=                          | A remote HornetQ host to connect to. |
   | =:port=           | =nil=, or =5445= if =:host= is set | A remote HornetQ port to connect to. Requires =:host= to be set. |
   | =:username=       | =nil=                          | The username to authenticate the connection with (if the broker has authentication enabled). Requires =:password= to be set. |
   | =:password=       | =nil=                          | The password to authenticate the connection with (if the broker has authentication enabled). Requires =:username= to be set. |
   #+TBLFM: 

   The =:json= encoding is useful when the message consumers aren't
   written in Clojure. For example, [[http://torquebox.org][TorqueBox]] Ruby processors will
   automatically convert json-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.

*** Some Examples

    #+begin_src clojure
      ;; A simple string
      (msg/publish "/queue/work" "simple string")
      ;; Notify everyone something interesting just happened
      (msg/publish "topic/news" {:event "VISIT" :url "/sales-inquiry"})
      ;; Move this message to the front of the line
      (msg/publish "/queue/work" some-message :priority :high :ttl 1000)
      ;; Make messages as complex as necessary
      (msg/publish "/queue/work" {:a "b" :c [1 2 3 {:foo 42}]})
      ;; Make messages consumable by a Ruby app
      (msg/publish "/queue/work" {:a "b" :c [1 2 3 {:foo 42}]} :encoding :json)
      ;; Publish to a remote broker
      (msg/publish "queue.remote-work" "a message" :host "foo.example.com" :port 5445)
      ;; The received message's metadata => {:foo 42, :bar 1}
      (msg/publish q (with-meta msg {:foo 42 :bar 0}) :properties {:bar 1})
    #+end_src

** Receiving

   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 [[./apidoc/immutant.messaging-api.html#immutant.messaging.core/get-properties][immutant.messaging.core/get-properties]].

   Immutant features three functions for consuming messages.
   - [[./apidoc/immutant.messaging-api.html#immutant.messaging/receive][immutant.messaging/receive]] Blocks the caller until a message arrives and returns
     the decoded message
   - [[./apidoc/immutant.messaging-api.html#immutant.messaging/message-seq][immutant.messaging/message-seq]] Creates a lazy sequence of messages
   - [[./apidoc/immutant.messaging-api.html#immutant.messaging/listen][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:

   | Option       | Default                            | Description                         |
   |--------------+------------------------------------+-------------------------------------|
   |              | <c>                                | <35>                                |
   | =:timeout=   | =10000=                            | An expiration in milliseconds, after which nil is returned; a value of 0 means wait forever |
   | =:selector=  | =nil=                              | A JMS expression used to filter messages according to the values of arbitrary =:properties= |
   | =:decode?=   | =true=                             | If true, the decoded message body is returned. Otherwise, the javax.jms.Message object is returned |
   | =:client-id= | =nil=                              | Identifies a durable topic subscriber; ignored for queues |
   | =:host=      | =nil=                              | A remote HornetQ host to connect to. |
   | =:port=      | =nil=, or =5445= if =:host= is set | A remote HornetQ port to connect to. Requires =:host= to be set. |
   | =:username=  | =nil=                              | The username to authenticate the connection with (if the broker has authentication enabled). Requires =:password= to be set. |
   | =:password=  | =nil=                              | The password to authenticate the connection with (if the broker has authentication enabled). Requires =:username= to be set. |

   For more details on message selectors, see [[http://docs.oracle.com/javaee/6/api/javax/jms/Message.html][javax.jms.Message]]. 

   The =listen= method takes two parameters: the destination name and
   a function accepting one parameter which will be applied to any
   received message. The following key/value options are also
   available:

   | Option         | Default                            | Description                         |
   |----------------+------------------------------------+-------------------------------------|
   |                | <c>                                | <35>                                |
   | =:concurrency= | =1=                                | The maximum number of listening threads that can simultaneouly call the function |
   | =:selector=    | =nil=                              | Same as =receive=                   |
   | =:decode?=     | =true=                             | If true, the decoded message body is passed to f. Otherwise, the javax.jms.Message object is passed |
   | =:client-id=   | =nil=                              | Identifies a durable topic subscriber; ignored for queues |
   | =:host=        | =nil=                              | A remote HornetQ host to connect to. |
   | =:port=        | =nil=, or =5445= if =:host= is set | A remote HornetQ port to connect to. Requires =:host= to be set. |
   | =:username=    | =nil=                              | The username to authenticate the connection with (if the broker has authentication enabled). Requires =:password= to be set. |
   | =:password=    | =nil=                              | The password to authenticate the connection with (if the broker has authentication enabled). Requires =:username= to be set. |

   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 [[./apidoc/immutant.messaging-api.html#immutant.xa.transaction/not-supported][immutant.xa.transaction/not-supported]] macro.

   See [[./transactions.html][Distributed Transactions]] for more details.

*** Some Examples

    #+begin_src clojure
      ;; Wait on a task
      (let [task (msg/receive "/queue/work")]
        (perform task))
      
      ;; Case-sensitive work queues?
      (msg/listen ".queue.lower" #(msg/publish "/queue/upper" (.toUpperCase %)))
      
      ;; Listen to a remote queue
      (msg/listen "queue/remote" #(do-someting %) :host "foo.example.com" :port 5445)
      
      ;; Contrived laziness
      (let [messages (message-seq queue)]
        (doseq [i (range 4)] (publish queue i))
        (= (range 4) (take 4 messages)))
    #+end_src

* Request/Response
   :PROPERTIES:
   :CUSTOM_ID: messaging-request-response
   :END:

  Immutant also provides an implementation of the [[http://en.wikipedia.org/wiki/Request-response][request/response]] pattern 
  for synchronous work distribution. This feature is provided by two 
  cleverly named functions: =request= and =respond=. Since they leverage
  the existing messaging subsystem, the work is automatically distributed
  across multiple workers within the same JVM or across multiple nodes if
  in a cluster.

** Request
   
   The [[./apidoc/immutant.messaging-api.html#immutant.messaging/request][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=, with one additional option:

   | Option     | Default | Description                         |
   |------------+---------+-------------------------------------|
   |            | <c>     | <35>                                |
   | =:timeout= | =10000= | Time in ms for the delayed receive to wait once it it is dereferenced, after which nil is returned |

** Respond

   The [[./apidoc/immutant.messaging-api.html#immutant.messaging/respond][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.

** Some Examples

   A basic example:

   #+begin_src clojure
     (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 @result)) ;; => 6
   #+end_src

   An example of using properties and selectors to segment work on the same 
   queue:
   
   #+begin_src clojure
     (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'")
     
     @(msg/request "/queue/work" [1 2 3 4] :properties {"operation" "add"}) ;; => 9
     
     @(msg/request "/queue/work" [1 2 3 4] :properties {"operation" "multiply"}) ;; => 24
   #+end_src







{'fraction_non_alphanumeric': 0.10891990291262135, 'fraction_numerical': 0.005794902912621359, 'mean_word_length': 2.4262993762993763, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 40, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6856451', 'n_tokens_mistral': 9156, 'n_tokens_neox': 8542, 'n_words': 3573}