(* This example creates a queue, and continously pulls messages from
the queue. It does not put any messages onto the queue, though.
Use t_sender.ml to do this.
*)
#use "topfind";;
#require "netamqp";;
open Netamqp_types
open Printf
let () =
Netamqp_endpoint.Debug.enable := true;
Netamqp_transport.Debug.enable := true
let esys = Unixqueue.create_unix_event_system()
(* We assume there is a RabbitMQ on localhost, listening on the default
port:
*)
let p = `TCP(`Inet("localhost", Netamqp_endpoint.default_port))
let ep = Netamqp_endpoint.create p (`AMQP_0_9 `One) esys
let c = Netamqp_connection.create ep
(* In RabbitMQ there is a built-in default user, "guest". The password
is also "guest". We authenticate as this user.
*)
let auth = Netamqp_connection.plain_auth "guest" "guest"
(* For this application we use channel 1 on the created connection: *)
let channel = 1
(* The name of the queue: *)
let qname = "test_xy"
(* The routing key says how the queue can be reached (the address): *)
let routing_key = qname ^ "_routing_key"
(* Call the following function to start the receiver. The function does
not finish, type CTRL-C to force it
*)
let receiver() =
(* At this point we create the TCP connection and establish the
AMQP-managed connection logic. "en_US" is the locale of server-generated
error messages. "/" is the virtual host.
*)
Netamqp_connection.open_s c [ auth ] (`Pref "en_US") "/";
eprintf "*** Connection could be opened, and the proto handshake is done!\n%!";
(* Now open the data channel. Channels are multiplexed over connections *)
let co = Netamqp_channel.open_s c channel in
eprintf "*** Channel could be opened!\n%!";
(* We declare the queue. This happens by sending a Queue-declare message
to the server and expecting a Queue-declare-ok message as response.
These control messages are also called methods. "Queue" is the class.
For documentation see the file amqp0-9-1.xml, and for the exact
Ocaml typing Netamqp_method_0_9.mli.
The functions sync_c2s_e/s are designed for request/response pairs
where the first method is emitted on the client side (c2s = client
to server). The "_e" variant (not used here) makes use of an
Ocamlnet engine. The "_s" variant waits until the response arrives.
What we effectively do: We create a queue if it not already exists with
name qname. We enable the auto-delete feature - the queue is deleted
when the last accessor is closed.
*)
let (r, _) =
Netamqp_endpoint.sync_c2s_s
ep
(`AMQP_0_9 (`Queue_declare(0, qname, false, false, false,
(* auto-delete: *) true, false,
[])))
None (* This value would allow to send content data along with the
method. Only certain methods permit this, though.
*)
channel
(-1.0) (* timeout *) in
( match r with
| `AMQP_0_9 (`Queue_declare_ok(_,_,_)) -> ()
| _ -> assert false
);
eprintf "*** Queue declared!\n%!";
(* Another call: We bind the queue to an exchange. The exchange determines
which messages are routed to which queue. There are pre-declared
exchanges, and we use here "amq.direct". This is a direct exchange
meaning that all content messages with the given routing_key are
added to the queue.
*)
let (r, _) =
Netamqp_endpoint.sync_c2s_s
ep
(`AMQP_0_9 (`Queue_bind(0, qname, (* exchange: *) "amq.direct",
routing_key,
false, [])))
None
channel
(-1.0) in
( match r with
| `AMQP_0_9 (`Queue_bind_ok) ->
()
| _ ->
assert false
);
eprintf "*** Queue binding established!\n%!";
(* We want now to achieve that we get all messages arriving at the queue.
In order to do so, we have to tell the server that we consume
from the queue. This is actually done in the next code block below.
First, we configure what happens when messages arrive. (If we did
not do this, the methods from the server carrying the queue messages
would be dropped because of the missing registration.)
The server will send us a Basic-deliver method for each queue message,
and this method carries the data of the message as additional
content payload. We register here a handler so all Basic-deliver
methods arriving on the channel will be forwarded to our callback
function.
The payload data is made available in d_opt. Normally, this is
always None, but if the method carries content, it is
[Some(header,body)]. The header is here [`P_basic(...)] with a
lot of arguments (P=properties, and "basic" because we are
using the Basic class for message handling). See t_sender.ml
how the header looks exactly.
The body is not a string but a list of mstring. The mstring object
is an abstraction defined in the Ocamlnet library "rpc"
(Xdr_mstring). It is generally used for large binary data strings.
It has two interesting features: First, it can not only be backed
by normal strings to store the data blob but also by bigarrays of
char. (There is special support in Ocamlnet for these bigarrays,
also called "memory" there, e.g. there are special versions of
Unix.read and Unix.write without any size limits and without any
data copying in the ocaml wrapper.) The second feature is that
an mstring can also pick any substring of the base representation
as content. In general, the mstring abstraction avoids string
copying. There are a number of helper functions in Xdr_mstring
and also in Netamqp_rtypes.
Each AMQP queue message needs to be acknowledged (unless this is
turned off). An ACK is done by sending the Basic-ack method with
the same delivery_tag we got in Basic-deliver. If we did not
ACK we would not get the next queue message. (N.B. One can configure
this with the Basic-qop method.)
We see here two new ways of handling methods:
- register_async_s2c: install a callback so that a function is
invoked when a certain type of method arrives. This is only
for one-way methods ("async methods" in AMQP speak).
- async_c2s: send a one-way method to the server
*)
Netamqp_endpoint.register_async_s2c
ep
(`AMQP_0_9 `Basic_deliver)
channel
(fun m d_opt ->
match m with
| `AMQP_0_9 (`Basic_deliver(_, delivery_tag, _, _, _) )->
eprintf "*** Got message!%!";
( match d_opt with
| None ->
eprintf "*** No data, though\m%!"
| Some(header,body) ->
let n = Xdr_mstring.length_mstrings body in
eprintf "*** DATA: %s\n"
(if n > 100 then
sprintf "[size: %d]" n
else
Xdr_mstring.concat_mstrings body
)
);
(* ACK this message *)
Netamqp_endpoint.async_c2s
ep
(`AMQP_0_9(`Basic_ack(delivery_tag, false)))
None
channel
| _ ->
assert false
);
(* After we registered the handler, we can enable queue consumption.
This is done by calling Basic-consume and expecting Basic-consume-ok
as response.
The "tag" is the consumption tag. Useful for cancelling consumption.
*)
let (r, _) =
Netamqp_endpoint.sync_c2s_s
ep
(`AMQP_0_9 (`Basic_consume(0, qname, "", false, false, false,
false, [] )))
None
channel
(-1.0) in
let consumer_tag =
match r with
| `AMQP_0_9 (`Basic_consume_ok tag) -> tag
| _ -> assert false in
eprintf "*** Created consumer\n%!";
(* As data reception is asynchronous business we need to run the
event system to activate it. Note that this event loop runs
forever.
*)
Unixqueue.run esys;
co
let close co =
if Netamqp_channel.is_open co then (
Netamqp_channel.close_s co;
eprintf "*** Channel could be closed!\n%!";
);
Netamqp_connection.close_s c;
eprintf "*** Connection could be closed!\n%!"
