GSP
Quick Navigator

Search Site

Unix VPS
A - Starter
B - Basic
C - Preferred
D - Commercial
MPS - Dedicated
Previous VPSs
* Sign Up! *

Support
Contact Us
Online Help
Handbooks
Domain Status
Man Pages

FAQ
Virtual Servers
Pricing
Billing
Technical

Network
Facilities
Connectivity
Topology Map

Miscellaneous
Server Agreement
Year 2038
Credits
 

USA Flag

 

 

Man Pages


Manual Reference Pages  -  NET::STOMP (3)

.ds Aq ’

NAME

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

CONTENTS

SYNOPSIS



  # 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 },
  ] });



DESCRIPTION

This module allows you to write a Stomp client. Stomp is the Streaming Text Orientated Messaging Protocol (or 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.

    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

CWnew

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 },
  } );



You can pass a logger object, for example a Log::Log4perl logger:



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



Warnings and errors will be logged instead of written to STDERR.

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.

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 forks. 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.

CWhostname

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.

CWport

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.

CWsocket_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).

CWssl

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.

CWssl_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.

CWfailover

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 } ]



CWhosts

Arrayref of hashrefs, each having a hostname key and a port key, and optionall 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".

CWcurrent_host

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

CWlogger

Optional logger object, the default one just logs to STDERR (see Net::Stomp::StupidLogger). You can pass in any object that implements (at least) the warn and fatal methods. They will be passed a string to log.

CWreconnect_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.

CWinitial_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.

CWreconnect_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.

CWconnect_delay

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

CWtimeout

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

CWreceipt_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

CWconnect

This starts the Stomp session with the Stomp server. You may pass in a login and passcode options, plus whatever other headers you may need (e.g. client-id, host).



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



Returns the frame that the server responded with (or undef if the connection was lost). If that frame’s command is not CONNECTED, something went wrong.

CWsend

This sends a message to a queue or topic. You must pass in a destination and a body (which must be a string of bytes). You can also pass whatever other headers you may need (e.g. transaction).



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



It’s probably a good idea to pass a content-length corresponding to the byte length of the body; this is necessary if the body contains a byte 0.

Always returns a true value. It automatically reconnects if writing to the socket fails.

CWsend_with_receipt

This sends a message asking for a receipt, and returns false if the receipt of the message is not acknowledged by the server:



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



If using ActiveMQ, you might also want to make the message persistent:



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



The actual frame sequence for a successful sending is:



  -> SEND
  <- RECEIPT



The actual frame sequence for a failed sending is:



  -> SEND
  <- anything but RECEIPT



If you are 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. <B>DO NOTB> use this method on a connection used for receiving.

If you want to see the RECEIPT or ERROR frame, pass a scalar as a second parameter to the method, and it will 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 parametrs, 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.

CWsend_transactional

This sends a message in transactional mode and returns false if the receipt of the message is not acknowledged by the server:



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



If using ActiveMQ, you might also want to make the message persistent:



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



send_transactional just wraps send_with_receipt in a STOMP transaction.

The actual frame sequence for a successful sending is:



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



The actual frame sequence for a failed sending is:



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



If you are 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. <B>DO NOTB> use this method on a connection used for receiving.

If you want to see the RECEIPT or ERROR frame, pass a scalar as a second parameter to the method, and it will 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 parametrs, 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.

CWdisconnect

This disconnects from the Stomp server:



  $stomp->disconnect;



If you call any other method after this, a new connection will be established automatically (to the next failover host, if there’s more than one).

Always returns a true value.

CWsubscribe

This subscribes you to a queue or topic. You must pass in a destination.

Always returns a true value.

The acknowledge mode (header ack) defaults to auto, which means that frames will be considered delivered after they have been sent to a client. The other option is client, which means that messages will only be considered delivered after the client specifically acknowledges them with an ACK frame (see "ack").

When Net::Stomp reconnects after a failure, all subscriptions will be re-instated, each with its own options.

Other options:
selector Specifies a JMS Selector using SQL 92 syntax as specified in the JMS 1.1 specification. This allows a filter to be applied to each message as part of the subscription.
id A unique identifier for this subscription. Very useful if you subscribe to the same destination more than once (e.g. with different selectors), so that messages arriving will have a subscription header with this value if they arrived because of this subscription.
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 will not block fast consumers.
activemq.exclusive Would I like to be an Exclusive Consumer on a queue.
activemq.maximumPendingMessageLimit For Slow Consumer Handling on non-durable topics by dropping old messages - we can set a maximum pending limit which 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.



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



CWunsubscribe

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.

CWreceive_frame

This blocks and returns 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 will block until a frame can be returned, or for however long the timeout attribue says. If you wish to wait for a specified time pass a timeout argument:



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



CWcan_read

This returns whether there is 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 to poll and return immediately. This method ignores the value of the timeout attribute.

CWack

This acknowledges that you have received and processed a frame and all frames before it (if you are using client acknowledgements):



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



Always returns a true value.

CWsend_frame

If this module does not provide enough help for sending frames, you may construct your own frame and send it:



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



This is the method used by all the other methods that send frames. It will keep trying to send the frame as hard as it can, reconnecting if the connection breaks (limited by "reconnect_attempts"). If no connection can be established, and "reconnect_attempts" is not 0, this method will die.

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

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.

Search for    or go to Top of page |  Section 3 |  Main Index


perl v5.20.3 NET::STOMP (3) 2015-06-27

Powered by GSP Visit the GSP FreeBSD Man Page Interface.
Output converted with manServer 1.07.