Net::ST...utorial(3)

📦 version 2.3, in epel 9

🎀 NAME 🎀

okay so welcome to Net::STOMP::Client::Tutorial(3) — literally the getting-started guide for STOMP and Net::STOMP::Client, and yes bestie it's giving main character energy 🦋

✨ INTRODUCTION ✨

Here's a five-parts tutorial to get you started with Net::STOMP::Client. no cap, five parts, we're doing the whole thing 💅

One thing though: you do need a basic knowledge of STOMP already. If you're starting from zero, go read:

which is the STOMP 1.2 protocol specification ✨

which is the Fuse Message Broker STOMP Tutorial 🌸

🔥 PART 1: CONNECTING TO A BROKER 🔥

Net::STOMP::Client, like the other modules chilling under the Net::* hierarchy, gives you an object oriented interface to a network protocol. iconic behaviour honestly 👑

🧚 CREATING AN OBJECT 🧚

To connect to a broker you first gotta create an object. That object is what you'll use later to actually interact with the broker. When the module makes the object, it tries to connect to the broker using either plain TCP or SSL. At the STOMP level? literally nothing happens yet. 🫶

To create the object you obviously need to say where you're connecting. Two ways, pick your fighter:

Option one, a single uri parameter:

$stomp = Net::STOMP::Client->new(uri => "stomp://mybroker:6163");

Option two, the host and port duo:

$stomp = Net::STOMP::Client->new(
    host => "mybroker",
    port => 6163,
);

🔐 USING SSL 🔐

SSL is a whole thing — more complex, because more parameters gotta be given. heads up: IO::Socket::SSL is used behind the scene, so you can go read its documentation for more info. Here's a complete example with certificate based authentication:

$stomp = Net::STOMP::Client->new(
    uri => "stomp+ssl://mybroker:6162",
    sockopts => {
        # path of the directory containing trusted certificates
        SSL_ca_path     => "/etc/ssl/ca",
        # client certificate to present
        SSL_cert_file   => "/etc/ssl/client-cert.pem",
        # client private key
        SSL_key_file    => "/etc/ssl/client-key.pem",
        # passphrase of the client private key
        SSL_passwd_cb   => sub { return("secret") },
        # verify mode (SSL_VERIFY_PEER)
        SSL_verify_mode => 1,
    },
);
💀 note: recent versions of IO::Socket::SSL will throw a warning at you if the verify mode isn't explicitly set. so like, set it. 🙈

👀 GETTING PEER INFORMATION 👀

Once you're connected — at TCP or SSL level — you can get info about the broker with the peer method. for instance:

$peer = $stomp->peer();
printf("connected to broker %s (IP %s), port %d\n",
    $peer->host(), $peer->addr(), $peer->port());

🫱 CONNECTING 🫱

After creating the broker object, you must start a STOMP session by sending a CONNECT frame. It's literally this simple:

$stomp->connect();

If authentication is required, you have to pass extra info at this stage. for instance:

$stomp->connect(
    login    => "guest",
    passcode => "welcome",
);

And in case the broker uses virtual hosts, you can add a host parameter to the connect method:

$stomp->connect(
    login    => "guest",
    passcode => "welcome",
    host     => "/",
);

If you don't give it explicitly, the host parameter defaults to whatever you gave in the new method, via the host or uri parameter. so it's got your back 💖

At this point the session is established and you can finally send and/or receive messages. slay.

📊 GETTING SERVER INFORMATION 📊

Once you're connected at STOMP level, you can get info about the broker using the server and version methods. for instance:

printf("speaking STOMP %s with server %s\n",
    $stomp->version(), $stomp->server() || "UNKNOWN");

👋 DISCONNECTING 👋

When you're done messaging, you should gracefully end the session by sending a DISCONNECT frame with:

$stomp->disconnect();
⚠️ note: STOMP does not support reconnection. once the session has ended, the broker object cannot be used anymore. it's giving one-time-use, no second chances 😭

🌈 WRAP UP 🌈

Putting all this together, here's a complete program that simply connects, starts and ends a session, printing information along the way:

use Net::STOMP::Client;
$stomp = Net::STOMP::Client->new(uri => "stomp://mybroker:6163");
$peer = $stomp->peer();
printf("connected to broker %s (IP %s), port %d\n",
    $peer->host(), $peer->addr(), $peer->port());
$stomp->connect();
printf("speaking STOMP %s with server %s\n",
    $stomp->version(), $stomp->server() || "UNKNOWN");
printf("session %s started\n", $stomp->session());
$stomp->disconnect();
printf("session ended\n");

💌 PART 2: SENDING MESSAGES 💌

📤 SENDING MESSAGES 📤

A message is made of a header (a list of key/value pairs) and a body. both are optional ✨

To send a message you have to issue a SEND frame. for instance:

$stomp->send(
    destination => "/queue/test",
    subject     => "this is a test",
    time        => time(),
    body        => "Hello world!\n",
);

🧾 USING RECEIPTS 🧾

By default you get zero confirmation that the message actually reached the broker. which is lowkey stressful ngl 😭 If you want that confirmation you gotta use receipts. The following code snippet sends two messages with a receipt header containing a pseudo-unique id, and waits for matching RECEIPT frames coming from the broker. It's easy because Net::STOMP::Client keeps track of which receipts are expected and haven't been received yet. obsessed.

$stomp->send(
    destination => "/queue/test",
    body        => "Test of receipts 1...\n",
    receipt     => $stomp->uuid(),
);
$stomp->send(
    destination => "/queue/test",
    body        => "Test of receipts 2...\n",
    receipt     => $stomp->uuid(),
);
# wait at most 10 seconds for all pending receipts
$stomp->wait_for_receipts(timeout => 10);
# complain if some receipts are still pending
die("Receipt not received!\n") if $stomp->receipts();
💡 note: all STOMP frames can carry a receipt header, so this isn't restricted to message sending. receipts for everyone 🎀

🔄 USING TRANSACTIONS 🔄

On top of that, you can use transactions to group the sending of several messages so that either none or all of them get handled by the broker. it's giving all-or-nothing, very drama 💅

# create a unique transaction id
$tid = $stomp->uuid();
# begin the transaction
$stomp->begin(transaction => $tid);
# send two messages as part of this transaction
$stomp->send(
    destination => "/queue/test1",
    body        => "message 1",
    transaction => $tid,
);
$stomp->send(
    destination => "/queue/test2",
    body        => "message 2",
    transaction => $tid,
);
# either abort or commit
if (... something bad happened...) {
    # abort/rollback the transaction
    $stomp->abort(transaction => $tid);
    # no messages have been delivered to the broker
} else {
    # commit the transaction
    $stomp->commit(transaction => $tid);
    # both messages have been delivered to the broker
}

📬 PART 3: RECEIVING MESSAGES 📬

🏷️ USING SUBSCRIPTIONS 🏷️

In order to receive frames you first have to subscribe to one or more destinations. literally as easy as:

$stomp->subscribe(destination => "/queue/test");

When you're done you simply unsubscribe with:

$stomp->unsubscribe(destination => "/queue/test");

It's good practice to add an id header to uniquely identify the subscription. All messages that are part of this subscription will have a matching subscription header. And this id can also be used to unsubscribe. multitasking icon 💅

Here's the thing though: the code above only works with STOMP 1.0. In STOMP 1.1 and above, the id header was made mandatory, so you must use something like:

$stomp->subscribe(
    destination => "/queue/test",
    id          => "testsub",
);
# received messages will contain: subscription:testsub
$stomp->unsubscribe(id => "testsub");

📥 RECEIVING FRAMES 📥

While you're subscribed to some destinations, the broker may decide at any time to send you MESSAGE frames. You can process those frames with a simple loop:

while ($frame = $stomp->wait_for_frames()) {
    # ... do something with the received frame ...
}

The code above is blocking and will loop forever. You can add a timeout option to have a non-blocking loop:

while (1) {
    # wait at most one second for a new frame
    $frame = $stomp->wait_for_frames(timeout => 1);
    # do what is appropriate
    if ($frame) {
        # ... do something with the received frame ...
    } else {
        # nothing received
    }
}

Because of the asynchronous nature of STOMP, receiving messages is a bit tricky 😭 you cannot know a priori which types of frames will be sent when. For instance, you may want to send messages (with receipts) while you're subscribed to some destinations, and you may receive a MESSAGE frame while you'd like to wait for a RECEIPT frame, or vice versa. chaotic, honestly.

The wait_for_frames method described above will wait for any frame, not only message frames. It's up to you to check that what you receive is a MESSAGE frame or not. That can be done with something like:

if ($frame->command() eq "MESSAGE") {
    # ... do something with the received message ...
} else {
    # something else than a message frame
}

🌟 WRAP UP 🌟

Putting all this together, here's a complete program that receives ten messages from to /queue/test:

use Net::STOMP::Client;
$stomp = Net::STOMP::Client->new(uri => "stomp://mybroker:6163");
# the next line will be explained in the next part of the tutorial ;-)
$stomp->message_callback(sub { return(1) });
$stomp->connect();
$sid = $stomp->uuid();
$stomp->subscribe(
    destination => "/queue/test",
    # we use the generated subscription id
    id          => $sid,
    # we want a receipt on our SUBSCRIBE frame
    receipt     => $stomp->uuid(),
);
$count = 0;
while ($count < 10) {
    $frame = $stomp->wait_for_frames(timeout => 1);
    if ($frame) {
        if ($frame->command() eq "MESSAGE") {
            $count++;
            printf("received message %d with id %s\n",
                   $count, $frame->header("message-id"));
        } else {
            # this will catch the RECEIPT frame
            printf("%s frame received\n", $frame->command());
        }
    } else {
        print("waiting for messages...\n");
    }
}
$stomp->unsubscribe(id => $sid);
$stomp->disconnect();

💅 PART 4: USING CALLBACKS 💅

Okay bestie, so remember in part 3 how STOMP being asynchronous makes it kinda tricky to handle all the different frame types properly? Yeah. That's exactly why callbacks exist. Net::STOMP::Client supports callbacks — little pieces of code that get called in well-defined situations. And there are two levels: global and local. 🦋

✨ GLOBAL CALLBACKS ✨

Global (per command) callbacks get called every time a frame is received. Net::STOMP::Client ships with default callbacks that should be enough for all frame types except MESSAGE frames. For those, it's genuinely up to the coder to decide what they want to do with the received messages. 💖

Here's an example with a message callback counting the messages received:

$stomp->message_callback(sub {
    my($self, $frame) = @_;
    $count++;
    return($self);
});

These callbacks are kinda global, and it's good practice not to change them during a session. If you don't need a global message callback, you can supply the dummy:

$stomp->message_callback(sub { return(1) });

Here's how to re-write a simplified version of the inner part of the receiving program from part 3 with a global callback:

$count = 0;
sub msg_cb ($$) {
    my($self, $frame) = @_;
    $count++;
    printf("received message %d with id %s\n",
           $count, $frame->header("message-id"));
    return($self);
}
$stomp->message_callback(\&msg_cb);
$stomp->wait_for_frames() while $count < 10;

🌈 LOCAL CALLBACKS 🌈

Local (per invocation) callbacks are called by the wait_for_frame method. Their return value controls what wait_for_frame does:

  • false: wait_for_frame should wait for more frames
  • true: wait_for_frame can stop and return this value

Here's how to use wait_for_frames with a local callback to wait until we receive a MESSAGE frame that contains quit in the body:

sub msg_cb ($$) {
    my($self, $frame) = @_;
    return(0) unless $frame->command() eq "MESSAGE";
    return(0) unless $frame->body() =~ /quit/;
    return($frame);
}
$frame = $stomp->wait_for_frames(callback => \&msg_cb);

So yeah — you can put the logic either in the global callbacks or in the local callbacks. The best practice is to have a single global message callback that doesn't change throughout the execution of the program, and then optionally put in local callbacks whatever may change from one place of the program to another. 💫

🎀 WRAP UP 🎀

Here's how to re-write the receiving program from part 3 with a global callback only counting the number of messages, plus a local callback doing the printing:

use Net::STOMP::Client;
$stomp = Net::STOMP::Client->new(uri => "stomp://mybroker:6163");
$stomp->connect();
sub msg_cb ($$) {
    my($self, $frame) = @_;
    my $cmd = $frame->command();
    if ($cmd eq "MESSAGE") {
        printf("received message %d with id %s\n",
               $count, $frame->header("message-id"));
    } else {
        printf("%s frame received\n", $cmd);
    }
    return($frame);
}
$stomp->message_callback(sub { $count++ });
$sid = $stomp->uuid();
$stomp->subscribe(
    destination => "/queue/test",
    id          => $sid,
    receipt     => $stomp->uuid(),
);
$count = 0;
while ($count < 10) {
    $stomp->wait_for_frames(
        callback => \&msg_cb,
        timeout => 1,
    ) or print("waiting for messages...\n");
}
$stomp->unsubscribe(id => $sid);
$stomp->disconnect();

🔥 PART 5: ADVANCED FEATURES 🔥

👑 ACKNOWLEDGMENT MODES 👑

Unless specified otherwise, subscriptions are made in "auto" mode. That means a message is considered delivered by the broker as soon as it sends the corresponding MESSAGE frame. The client may not receive the frame, or it could exit before processing it. This could result in message loss. 😭

⚠️ Bestie, message loss is real — if you care about not dropping messages, keep reading.

To avoid message loss, you can change the subscription acknowledgment mode to "client" instead of "auto". This is an option of the SUBSCRIBE frame:

$stomp->subscribe(
    destination => "/queue/test",
    id          => $sid,
    ack         => "client",
);

In "client" mode, the client must explicitly acknowledge the messages it has successfully processed. That's done by sending an ACK frame with a "message-id" header matching the one of the received message:

$stomp->ack("message-id" => $frame->header("message-id"));

In practice this is more complex, because the exact way to acknowledge messages changed between STOMP 1.0 (only message-id required), STOMP 1.1 (both message-id and subscription required), and STOMP 1.2 (only id required). 🙈

Net::STOMP::Client has a unique feature to ease acknowledgments: you can directly pass the frame that holds the received message, and the module will properly acknowledge it regardless of the STOMP protocol version used:

$stomp->ack(frame => $frame);

Messages that have not been acknowledged by the end of the session will be resent by the broker. 👀

Also note: starting with STOMP 1.1 there's also a "client-individual" mode. Please consult the protocol specification for more details.

⚡ EFFICIENT I/O ⚡

The high-level methods handle one frame at a time. That can be inefficient for small frames. For instance, the send method will build a frame object, encode it, and send it on the wire with at least one call to syswrite — maybe for very few bytes. 💀

The low-level methods let you control this better and queue messages in memory before sending them. That way you group data and use I/O more efficiently. Slay. ✨

Here's how to queue ten messages and send them in one go:

foreach my $n (1 .. 10) {
    $frame = Net::STOMP::Client::Frame->new(
        command => "SEND",
        headers => { destination => "/topic/test" },
        body    => "message $n",
    );
    # simply add the frame to the outgoing queue
    $stomp->queue_frame($frame);
}
# no timeout given: block until all data has been sent
$stomp->send_data();

🌸 SEE ALSO 🌸

  • IO::Socket::SSL
  • Net::STOMP::Client
  • Net::STOMP::Client::Connection
  • Net::STOMP::Client::Frame
  • Net::STOMP::Client::HeartBeat
  • Net::STOMP::Client::Peer
  • Net::STOMP::Client::Receipt
  • Net::STOMP::Client::Version

🫶 AUTHOR 🫶

Lionel Cons <http://cern.ch/lionel.cons>

Copyright (C) CERN 2010-2017

This page's footer says source/version perl v5.32.1, dated 2019-07-26 — mwah 💋✨

📦 where this page comes from