Net::Stomp(3pm)

📦
version 0.62 · 5 versions
, in debian trixie, ubuntu noble, ubuntu resolute, ubuntu stonking

🫶 Net::Stomp(3pm) 🫶

okay so bestie, welcome to the manual page for Net::Stomp. It's giving "I speak Stomp so you don't have to" energy, and honestly? Iconic. Let's get into it. ✨

🌸 NAME 🌸

Net::Stomp — A Streaming Text Orientated Messaging Protocol Client

📜 SYNOPSIS 📜

lowkey just look at the code, it's all there:

         # send a message to the queue 'foo'
         use Net::Stomp;
         my $stomp = Net::Stomp->new( { hostname => 'localhost', port => '61613' } );
         $stomp->connect( { login => 'hello', passcode => 'there' } );
         $stomp->send(
             { destination => '/queue/foo', body => 'test message' } );
         $stomp->disconnect;

         # subscribe to messages from the queue 'foo'
         use Net::Stomp;
         my $stomp = Net::Stomp->new( { hostname => 'localhost', port => '61613' } );
         $stomp->connect( { login => 'hello', passcode => 'there' } );
         $stomp->subscribe(
             {   destination             => '/queue/foo',
                 'ack'                   => 'client',
                 'activemq.prefetchSize' => 1
             }
         );
         while (1) {
           my $frame = $stomp->receive_frame;
           if (!defined $frame) {
             # maybe log connection problems
             next; # will reconnect automatically
           }
           warn $frame->body; # do something here
           $stomp->ack( { frame => $frame } );
         }
         $stomp->disconnect;

         # write your own frame
         my $frame = Net::Stomp::Frame->new(
              { command => $command, headers => $conf, body => $body } );
         $self->send_frame($frame);

         # connect with failover supporting similar URI to ActiveMQ
         $stomp = Net::Stomp->new({ failover => "failover://tcp://primary:61616" })
         # "?randomize=..." and other parameters are ignored currently
         $stomp = Net::Stomp->new({ failover => "failover:(tcp://primary:61616,tcp://secondary:61616)?randomize=false" })

         # Or in a more natural perl way
         $stomp = Net::Stomp->new({ hosts => [
           { hostname => 'primary', port => 61616 },
           { hostname => 'secondary', port => 61616 },
         ] });

periodt. 💅

💬 DESCRIPTION 💬

okay so this module lets you write a Stomp client. Stomp is the Streaming Text Orientated Messaging Protocol (or, and this is not a drill, the Protocol Briefly Known as TTMP and Represented by the symbol :ttmp). It's a simple and easy to implement protocol for working with Message Orientated Middleware from any language. Net::Stomp is useful for talking to Apache ActiveMQ, an open source (Apache 2.0 licensed) Java Message Service 1.1 (JMS) message broker packed with many enterprise features. 🫶

A Stomp frame consists of a command, a series of headers and a body — see Net::Stomp::Frame for more details.

For details on the protocol see <https://stomp.github.io/>.

In long-lived processes, you can use a new Net::Stomp object to send each message, but it's more polite to the broker to keep a single object around and re-use it for multiple messages; this reduce the number of TCP connections that have to be established. Net::Stomp tries very hard to re-connect whenever something goes wrong. No cap, it's a trooper. 💪

🐝 ActiveMQ-specific suggestions 🐝

To enable the ActiveMQ Broker for Stomp add the following to the activemq.xml configuration inside the <transportConnectors> section:

<transportConnector name="stomp" uri="stomp://localhost:61613"/>

To enable the ActiveMQ Broker for Stomp and SSL add the following inside the <transportConnectors> section:

<transportConnector name="stomp+ssl" uri="stomp+ssl://localhost:61612"/>

For details on Stomp in ActiveMQ See <http://activemq.apache.org/stomp.html>. ✨

🛠 CONSTRUCTOR 🛠

✨ "new" ✨

The constructor creates a new object. You must pass in a hostname and a port or set a failover configuration:

my $stomp = Net::Stomp->new( { hostname => 'localhost', port => '61613' } );

If you want to use SSL, make sure you have IO::Socket::SSL and pass in the SSL flag:

my $stomp = Net::Stomp->new( {
  hostname => 'localhost',
  port     => '61612',
  ssl      => 1,
} );

If you want to pass in IO::Socket::SSL options:

my $stomp = Net::Stomp->new( {
  hostname    => 'localhost',
  port        => '61612',
  ssl         => 1,
  ssl_options => { SSL_cipher_list => 'ALL:!EXPORT' },
} );

🔁 Failover 🔁

There is some failover support in Net::Stomp. You can specify "failover" in a similar manner to ActiveMQ (<http://activemq.apache.org/failover-transport-reference.html>) for similarity with Java configs or using a more natural method to Perl of passing in an array-of-hashrefs in the hosts parameter. The ssl and ssl_options parameters are inherited by all hosts.

When Net::Stomp connects the first time, upon construction, it will simply try each host in the list, stopping at the first one that accepts the connection, dying if no connection attempt is successful. You can set "initial_reconnect_attempts" to 0 to mean "keep looping forever", or to an integer value to mean "only go through the list of hosts this many times" (the default value is therefore 1).

When Net::Stomp notices that the connection has been lost (inside "send_frame" or "receive_frame"), it will try to re-connect. In this case, the number of connection attempts will be limited by "reconnect_attempts", which defaults to 0, meaning "keep trying forever".

🍴 Reconnect on "fork" 🍴

By default Net::Stomp will reconnect, using a different socket, if the process "fork"s. This avoids problems when parent & child write to the socket at the same time. If, for whatever reason, you don't want this to happen, set "reconnect_on_fork" to 0 (either as a constructor parameter, or by calling the method).

🎀 ATTRIBUTES 🎀

These can be passed as constructor parameters, or used as read/write accessors. 💖

🏠 "hostname" 🏠

If you want to connect to a single broker, you can specify its hostname here. If you modify this value during the lifetime of the object, the new value will be used for the subsequent reconnect attempts.

🚪 "port" 🚪

If you want to connect to a single broker, you can specify its port here. If you modify this value during the lifetime of the object, the new value will be used for the subsequent reconnect attempts.

🔌 "socket_options" 🔌

Optional hashref, it will be passed to the IO::Socket::IP, IO::Socket::SSL, or IO::Socket::INET constructor every time we need to get a socket.

In addition to the various options supported by those classes, you can set keep_alive to a true value, which will enable TCP-level keep-alive on the socket (see the TCP Keepalive HOWTO <http://www.tldp.org/HOWTO/html_single/TCP-Keepalive-HOWTO/> for some information on that feature).

🔒 "ssl" 🔒

Boolean, defaults to false, whether we should use SSL to talk to the single broker. If you modify this value during the lifetime of the object, the new value will be used for the subsequent reconnect attempts.

🛡 "ssl_options" 🛡

Options to pass to IO::Socket::SSL when connecting via SSL to the single broker. If you modify this value during the lifetime of the object, the new value will be used for the subsequent reconnect attempts.

🌪 "failover" 🌪

Modifying this attribute after the object has been constructed has no effect. Pass this as a constructor parameter only. Its value must be a URL (as a string) in the form:

failover://(tcp://$hostname1:$port1,tcp://$hostname2:$port,...)

This is equivalent to setting "hosts" to:

[ { hostname => $hostname1, port => $port1 },
  { hostname => $hostname2, port => $port2 } ]

If the ssl and ssl_options constructor parameters are used with failover the SSL settings are applied for all hosts.

🗂 "hosts" 🗂

Arrayref of hashrefs, each having a hostname key and a port key, and optionally ssl and ssl_options. Connections will be attempted in order, looping around if necessary, depending on the values of "initial_reconnect_attempts" and "reconnect_attempts".

📍 "current_host" 📍

If using multiple hosts, this is the index (inside the "hosts" array) of the one we're currently connected to.

📝 "logger" 📝

Optional logger object, the default one is a Log::Any logger. You can pass in any object with the same API, or configure Log::Any::Adapter to route the messages to whatever logging system you need.

🍴 "reconnect_on_fork" 🍴

Boolean, defaults to true. Reconnect if a method is being invoked from a different process than the one that created the object. Don't change this unless you really know what you're doing.

🔄 "initial_reconnect_attempts" 🔄

Integer, how many times to loop through the "hosts" trying to connect, before giving up and throwing an exception, during the construction of the object. Defaults to 1. 0 means "keep trying forever". Between each connection attempt there will be a sleep of "connect_delay" seconds.

🔁 "reconnect_attempts" 🔁

Integer, how many times to loop through the "hosts" trying to connect, before giving up and throwing an exception, during "send_frame" or "receive_frame". Defaults to 0, meaning "keep trying forever". Between each connection attempt there will be a sleep of "connect_delay" seconds.

😴 "connect_delay" 😴

Integer, defaults to 5. How many seconds to sleep between connection attempts to brokers.

⏳ "timeout" ⏳

Integer, in seconds, defaults to undef. The default timeout for read operations. undef means "wait forever".

🧾 "receipt_timeout" 🧾

Integer, in seconds, defaults to undef. The default timeout while waiting for a receipt (in "send_with_receipt" and "send_transactional"). If undef, the global "timeout" is used.

💅 METHODS 💅

okay bestie, this is the main character section. all the things you actually call on your $stomp object. buckle up 🎀

💖 "connect" 💖

this is the glow-up moment where you actually start the Stomp session with the Stomp server. you can hand it a login and a passcode, plus literally whatever other headers your lil heart needs (think client-id, host, etc).

$stomp->connect( { login => 'hello', passcode => 'there' } );

it gives you back the frame the server responded with — or undef if the connection dipped on you 😭. and heads up: if that frame's command isn't CONNECTED, something went wrong. not a vibe.

💖 "send" 💖

sending a message to a queue or a topic. you must pass a destination and a body (and the body has to be a string of bytes, no funny business). you can also throw in whatever other headers you need (like transaction).

$stomp->send( { destination => '/queue/foo', body => 'test message' } );

pro tip from the manpage: it's probably a good idea to pass a content-length that matches the byte length of your body — and it's necessary if your body contains a byte 0. so like, actually do it. 🫶

always returns a true value. and the way it automatically reconnects if writing to the socket fails?? lowkey iconic.

💖 "send_with_receipt" 💖

this one sends a message and asks for a receipt, returning false if the server doesn't acknowledge it:

$stomp->send_with_receipt(
    { destination => '/queue/foo', body => 'test message' }
) or die "Couldn't send the message!";

if you're using ActiveMQ, you might also wanna make the message persistent:

$stomp->send_transactional(
    { destination => '/queue/foo', body => 'test message', persistent => 'true' }
) or die "Couldn't send the message!";

the actual frame sequence when it goes well is giving:

-> SEND
<- RECEIPT

and when it goes badly:

-> SEND
<- anything but RECEIPT

so if you're using this connection only to send (meaning you've never called "subscribe"), the only thing that could show up instead of a RECEIPT is an ERROR frame. but if you did subscribe, the broker may very well send a MESSAGE before sending the RECEIPT.

🚨 DO NOT use this method on a connection used for receiving. fr fr. that's a whole mess.

if you wanna see the RECEIPT or ERROR frame, pass a scalar as the second parameter and it'll get set to the received frame:

my $success = $stomp->send_transactional(
    { destination => '/queue/foo', body => 'test message' },
    $received_frame,
);
if (not $success) { warn $received_frame->as_string }

you can specify a timeout in the parameters, just like for "received_frame". this function will wait for that timeout, or for "receipt_timeout", or for "timeout" — whichever is defined — or forever, if none of them is defined. forever girlie, no cap.

💖 "send_transactional" 💖

this sends a message in transactional mode and returns false if the receipt of the message isn't acknowledged by the server:

$stomp->send_transactional(
    { destination => '/queue/foo', body => 'test message' }
) or die "Couldn't send the message!";

if you're using ActiveMQ, you might also wanna make the message persistent:

$stomp->send_transactional(
    { destination => '/queue/foo', body => 'test message', persistent => 'true' }
) or die "Couldn't send the message!";

send_transactional basically just wraps send_with_receipt in a STOMP transaction. cute lil wrapper ✨

the frame sequence for a successful send:

-> BEGIN
-> SEND
<- RECEIPT
-> COMMIT

and for a failed send:

-> BEGIN
-> SEND
<- anything but RECEIPT
-> ABORT

if you're using this connection only to send (i.e. you've never called "subscribe"), the only thing that could be received instead of a RECEIPT is an ERROR frame. but if you subscribed, the broker may well send a MESSAGE before sending the RECEIPT.

🚨 DO NOT use this method on a connection used for receiving. same warning, same energy, still very real.

if you wanna see the RECEIPT or ERROR frame, pass a scalar as a second parameter and it'll be set to the received frame:

my $success = $stomp->send_transactional(
    { destination => '/queue/foo', body => 'test message' },
    $received_frame,
);
if (not $success) { warn $received_frame->as_string }

you can specify a timeout in the parameters, just like for "received_frame". this function will wait for that timeout, or for "receipt_timeout", or for "timeout", whichever is defined — or forever, if none is defined. yeah. forever. 💀

💖 "disconnect" 💖

bye bestie 👋

$stomp->disconnect;

if you call any other method after this, a new connection gets established automatically (to the next failover host, if there's more than one). it literally cannot stay away from you.

always returns a true value.

💖 "subscribe" 💖

this subscribes you to a queue or topic. you must pass a destination.

always returns a true value.

the acknowledge mode (header ack) defaults to auto, which means frames get considered delivered right after they've been sent to a client. the other option is client, which means messages only count as delivered once the client specifically acknowledges them with an ACK frame (peep "ack").

and here's the slay: when Net::Stomp reconnects after a failure, all subscriptions get re-instated, each with its own options. it remembers. 💅

other options:

🌸 "selector"

specifies a JMS Selector using SQL 92 syntax, as specified in the JMS 1.1 specification. this lets a filter be applied to each message as part of the subscription.

🌸 "id"

a unique identifier for this subscription. super useful if you subscribe to the same destination more than once (like with different selectors), so that messages arriving will have a subscription header with this value if they arrived because of this subscription. iykyk.

🌸 "activemq.dispatchAsync"

should messages be dispatched synchronously or asynchronously from the producer thread for non-durable topics in the broker. for fast consumers, set this to false. for slow consumers, set it to true so that dispatching won't block fast consumers.

🌸 "activemq.exclusive"

would I like to be an Exclusive Consumer on a queue. (yes that's the original phrasing and yes it's a mood.)

🌸 "activemq.maximumPendingMessageLimit"

for Slow Consumer Handling on non-durable topics by dropping old messages — you can set a maximum pending limit, and once a slow consumer backs up to this high water mark, we begin to discard old messages.

🌸 "activemq.noLocal"

specifies whether or not locally sent messages should be ignored for subscriptions. set to true to filter out locally sent messages.

🌸 "activemq.prefetchSize"

specifies the maximum number of pending messages that will be dispatched to the client. once this maximum is reached, no more messages are dispatched until the client acknowledges a message. set to 1 for very fair distribution of messages across consumers where processing messages can be slow.

🌸 "activemq.priority"

sets the priority of the consumer so that dispatching can be weighted in priority order.

🌸 "activemq.retroactive"

for non-durable topics, do you wish this subscription to the retroactive.

🌸 "activemq.subscriptionName"

for durable topic subscriptions you must specify the same "client-id" on the connection and "subscriptionName" on the subscribe.

and here's the whole thing in action:

$stomp->subscribe(
    {   destination             => '/queue/foo',
        'ack'                   => 'client',
        'activemq.prefetchSize' => 1
    }
);

💖 "unsubscribe" 💖

this unsubscribes you to a queue or topic. you must pass in a destination or an id:

$stomp->unsubcribe({ destination => '/queue/foo' });

always returns a true value.

💖 "receive_frame" 💖

this blocks and hands you the next Stomp frame — or undef if there was a connection problem. 😭

my $frame = $stomp->receive_frame;
warn $frame->body; # do something here

by default this method blocks until a frame can be returned, or for however long the timeout attribute says. if you wanna wait for a specific time, pass a timeout argument:

# Wait half a second for a frame, else return undef
$stomp->receive_frame({ timeout => 0.5 })

💖 "can_read" 💖

returns whether there's new data waiting to be read from the STOMP server. optionally takes a timeout in seconds:

my $can_read = $stomp->can_read;
my $can_read = $stomp->can_read({ timeout => '0.1' });

undef says block until something can be read. 0 says poll and return immediately. this method ignores the value of the timeout attribute — it's built different.

💖 "ack" 💖

acknowledges that you've received and processed a frame, and all frames before it (if you're using client acknowledgements):

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

always returns a true value.

💖 "nack" 💖

tells the remote end that you've been unable to process a received frame (if you're using client acknowledgements). see the individual stomp server documentation for info about additional fields that can be passed to alter NACK behavior:

$stomp->nack( { frame => $frame } );

always returns a true value.

💖 "send_frame" 💖

if this module doesn't give you enough help for sending frames, you can construct your own and send it yourself. DIY queen 👑

# write your own frame
my $frame = Net::Stomp::Frame->new(
     { command => $command, headers => $conf, body => $body } );
$self->send_frame($frame);

this is the method every other frame-sending method uses. it will keep trying to send the frame as hard as it can, reconnecting if the connection breaks (limited by "reconnect_attempts"). and if no connection can be established, and "reconnect_attempts" isn't 0, this method will die.

💀 so like, wrap it. just saying.

always returns an empty list.

🌈 SEE ALSO 🌈

Net::Stomp::Frame.

🎀 SOURCE REPOSITORY 🎀

https://github.com/dakkar/Net-Stomp

👑 AUTHORS 👑

Leon Brocard <acme@astray.com>, Thom May <thom.may@betfair.com>, Michael S. Fischer <michael@dynamine.net>, Ash Berlin <ash_github@firemirror.com>

🦋 CONTRIBUTORS 🦋

Paul Driver <frodwith@cpan.org>, Andreas Faafeng <aff@cpan.org>, Vigith Maurice <vigith@yahoo-inc.com>, Stephen Fralich <sjf4@uw.edu>, Squeeks <squeek@cpan.org>, Chisel Wright <chisel@chizography.net>, Gianni Ceccarelli <dakkar@thenautilus.net>

Copyright (C) 2006-9, Leon Brocard Copyright (C) 2009, Thom May, Betfair.com Copyright (C) 2010, Ash Berlin, Net-a-Porter.com Copyright (C) 2010, Michael S. Fischer

This module is free software; you can redistribute it or modify it under the same terms as Perl itself.

and that's the whole page bestie 💖 this lil bundle of joy is perl v5.38.2 from 2024-02-24, filed under Net::Stomp(3pm). go forth and stomp responsibly, mwah 🫶✨

📦 where this page comes from