The Perl Toolchain Summit needs more sponsors. If your company depends on Perl, please support this very important event.
 / 
Net-Async-AMQP-2.000
River stage zero No dependents
++
/ Net::Async::AMQP

NAME

Net::Async::AMQP - provides client interface to AMQP using IO::Async

VERSION

version 2.000

SYNOPSIS

 use IO::Async::Loop;
 use Net::Async::AMQP;
 my $loop = IO::Async::Loop->new;
 $loop->add(my $amqp = Net::Async::AMQP->new);
 $amqp->connect(
   host => 'localhost',
   user => 'guest',
   pass => 'guest',
 )->get;

DESCRIPTION

Does AMQP things. Note that the API may change before the stable 1.000 release - "ALTERNATIVE AMQP IMPLEMENTATIONS" are listed below if you want to evaluate other options.

If you want a higher-level API which manages channels and connections, try Net::Async::AMQP::ConnectionManager.

Examples are in the examples/ directory.

AMQP support

The following AMQP features are supported:

  • Queue declare, bind, delete

  • Exchange declare, delete

  • Consumer setup and cancellation

  • Message publishing

  • Explicit ACK

  • QoS

  • SSL

RabbitMQ-specific features

RabbitMQ provides some additional features:

  • Exchange-to-exchange binding

  • Server flow control notification

  • Consumer cancellation notification

  • Reject

  • TTL for message expiry

  • 255-level priorities

Missing features

The following features aren't currently implemented - raise a request via RT or by email ("AUTHOR") if you want any of these:

  • Transactions

  • Flow control

  • SASL auth

This implementation is designed to handle many simultaneous channels and connections. If you just want a single consumer/publisher, one of the librabbitmq-c implementations may be sufficient.

CONSTANTS

AUTH_MECH

Defines the mechanism used for authentication. Currently only AMQPLAIN is supported.

PAYLOAD_HEADER_LENGTH

Length of header used in payload messages. Defined by the AMQP standard as 8 bytes.

MAX_FRAME_SIZE

Largest amount of data we'll attempt to send in a single frame. Actual frame limit will be negotiated with the remote server. Defaults to 262144.

MAX_CHANNELS

Maximum number of channels to request. Defaults to the AMQP limit (65535). Attempting to set this any higher will not end well, it's an unsigned 16-bit value.

HEARTBEAT_INTERVAL

Interval in seconds between heartbeat frames, zero to disable. Can be overridden by PERL_AMQP_HEARTBEAT_INTERVAL in the environment, default is 0 (disabled).

PACKAGE VARIABLES

$XML_SPEC

This defines the path to the AMQP XML spec, which Net::AMQP uses to create methods and handlers for the appropriate version of the MQ protocol.

Defaults to an extended version of the 0.9.1 protocol as used by RabbitMQ, this is found in the amqp0-9-1.extended.xml distribution sharedir (see File::ShareDir).

Normally, you should be able to ignore this. If you want to load an alternative spec, note that (a) this is global, rather than per-instance, (b) it needs to be set before you use this module.

 BEGIN { $Net::Async::AMQP::XML_SPEC = '/tmp/amqp.xml' }
 use Net::Async::AMQP;

Once loaded, this module will not attempt to apply the spec again.

%CONNECTION_DEFAULTS

The default parameters to use for "connect". Changing these values is permitted, but do not attempt to delete from or add any entries to the hash.

Passing parameters directly to "connect" is much safer, please do that instead.

METHODS

configure

Set up variables. Takes the following optional named parameters:

  • heartbeat_interval - (optional) interval between heartbeat messages, default is set by the "HEARTBEAT_INTERVAL" constant

  • max_channels - how many channels to allow on this connection, default is defined by the "MAX_CHANNELS" constant

Returns the new instance.

bus

Event bus. Used for sharing global events such as connection closure.

connect

Takes the following parameters:

  • port - the AMQP port, defaults to 5672, can be a service name if preferred

  • host - host to connect to, defaults to localhost

  • local_host - our local IP to connect from

  • user - which user to connect as, defaults to guest

  • pass - the password for this user, defaults to guest

  • ssl - true if you want to connect over SSL

  • SSL_* - SSL-specific parameters, see IO::Async::SSL and IO::Socket::SSL for details

Returns $self.

on_stream

Called once the underlying TCP connection has been established.

Returns nothing of importance.

on_read

Called whenever there's data available to be read.

on_closed

Called when the TCP connection is closed.

post_connect

Sends initial startup header and applies listener for the Connection::Start message.

Returns $self.

setup_tuning

Applies listener for the Connection::Tune message, used for determining max frame size and heartbeat settings.

Returns $self.

open_connection

Establish a new connection to a vhost - this is called after tuning is complete, and must happen before any channel connections are attempted.

Returns $self.

setup_connection

Applies listener for the Connection::OpenOk message, which triggers the connected event.

Returns $self.

connected

Returns a Future which will resolve when the MQ connection is ready for use.

next_channel

Returns the next available channel ready for "open_channel". Note that whatever it reports will be completely wrong if you've manually specified a channel anywhere, so don't do that.

If channels have been closed on this connection, those IDs will be reused in preference to handing out a new ID.

create_channel

Returns a new ::Channel instance, populating the map of assigned channels in the process. Takes a single parameter:

open_channel

Opens a new channel.

Returns the new Net::Async::AMQP::Channel instance.

close

Close the connection.

Returns a Future which will resolve with $self when the connection is closed.

channel_closed

next_pending

Retrieves the next pending handler for the given incoming frame type (see "amqp_frame_type" in Net::Async::AMQP::Utils), and calls it.

Takes the following parameters:

  • $type - the frame type, such as 'Basic::ConnectOk'

  • $frame - the frame itself

Returns $self.

METHODS - Accessors

host

The current host.

vhost

Virtual host.

port

Port number. Usually 5672.

user

MQ user.

frame_max

Maximum number of bytes allowed in any given frame. This is the value negotiated with the remote server.

channel_max

Maximum number of channels. This is whatever we ended up with after initial negotiation.

last_frame_time

Timestamp of the last frame we received from the remote. Used for handling heartbeats.

stream

Returns the current IO::Async::Stream for the AMQP connection.

incoming_message

Future for the current incoming message (received in two or more parts: the header then all body chunks).

METHODS - Internal

The following methods are intended for internal use. They are documented for completeness but should not normally be needed outside this library.

heartbeat_interval

Current maximum interval between frames.

missed_heartbeats_allowed

How many times we allow the remote to miss the frame-sending deadline in a row before we give up and close the connection. Defined by the protocol, should be 3x heartbeats.

apply_heartbeat_timer

Enable both heartbeat timers.

reset_heartbeat

Resets our side of the heartbeat timer.

This is used to ensure we send data at least once every "heartbeat_interval" seconds.

heartbeat_receive_timer

Timer for tracking frames we've received.

heartbeat_send_timer

Timer for tracking when we're due to send out something.

handle_heartbeat_failure

Called when heartbeats are enabled and we've had no response from the server for 3 heartbeat intervals (see "missed_heartbeats_allowed"). We'd expect some frame from the remote - even if just a heartbeat frame - at least once every heartbeat interval so if this triggers then we're likely dealing with a dead or heavily loaded server.

This will invoke the "heartbeat_failure event" then close the connection.

send_heartbeat

Sends the heartbeat frame.

push_pending

Adds the given handler(s) to the pending handler list for the given type(s).

Takes one or more of the following parameter pairs:

Returns $self .

remove_pending

Removes a coderef from the pending event handler.

Returns $self .

write

Writes data to the server.

Returns a Future which will resolve to an empty list when done.

process_frame

Process a single incoming frame.

Takes the following parameters:

Returns $self.

split_payload

Splits a message into separate frames.

Takes the $payload as a scalar containing byte data, and the following parameters:

  • exchange - where we're sending the message

  • routing_key - other part of message destination

Additionally, the following headers can be passed:

  • content_type

  • content_encoding

  • headers

  • delivery_mode

  • priority

  • correlation_id

  • reply_to

  • expiration

  • message_id

  • timestamp

  • type

  • user_id

  • app_id

  • cluster_id

Returns list of frames suitable for passing to "send_frame".

send_frame

Send a single frame.

Takes the $frame instance followed by these optional named parameters:

  • channel - which channel we should send on

Returns a Future which will resolve to an empty list when the frame has been written (this does not guarantee that the server has received it).

header_bytes

Byte string representing the header bytes we should send on initial TCP connect. Net::AMQP uses AMQP\x01\x01\x09\x01, which does not appear to comply with AMQP 0.9.1 section 4.2.2.

future

Returns a new IO::Async::Future instance.

Supports optional named parameters for setting label etc.

EVENTS

The following events may be raised by this class - use "subscribe_to_event" in Mixin::Event::Dispatch to watch for them:

 $mq->bus->subscribe_to_event(
   heartbeat_failure => sub {
         my ($ev, $last) = @_;
         print "Heartbeat failure detected\n";
   }
 );

connected event

Called after the connection has been opened.

close event

Called after the remote has closed the connection.

heartbeat_failure event

Raised if we receive no data from the remote for more than 3 heartbeat intervals and heartbeats are enabled,

unexpected_frame event

If we receive an unsolicited frame from the server this event will be raised:

 $mq->bus->subscribe_to_event(
  unexpected_frame => sub {
   my ($ev, $type, $frame) = @_;
   warn "Frame type $type received: $frame\n";
  }
 )

ALTERNATIVE AMQP IMPLEMENTATIONS

As usual there's a few other options:

Modules based on librabbitmq-c are probably fine for simple sync tasks, but I wouldn't recommend them for any async work. In contrast, the Net::AMQP protocol module generates all the classes and methods directly from the AMQP spec, so it's an excellent base on which to develop the transport module (as in the case of Net::Async::AMQP).

SEE ALSO

  • Net::AMQP - this does all the hard work of converting the XML protocol specification into appropriate Perl methods and classes.

  • Net::RabbitMQ::Management::API - doesn't do AMQP, but provides sync (LWP-based) access to RabbitMQ's HTTP API

  • Test::Net::RabbitMQ - provides a basic server implementation for testing

INHERITED METHODS

IO::Async::Notifier

add_child, adopt_future, can_event, children, configure_unknown, debug_printf, get_loop, invoke_error, invoke_event, loop, make_event_cb, maybe_invoke_event, maybe_make_event_cb, new, notifier_name, parent, remove_child, remove_from_parent

AUTHOR

Tom Molesworth <TEAM@cpan.org>

LICENSE

Licensed under the same terms as Perl itself, with additional licensing terms for the MQ spec to be found in share/amqp0-9-1.extended.xml ('a worldwide, perpetual, royalty-free, nontransferable, nonexclusive license to (i) copy, display, distribute and implement the Advanced Messaging Queue Protocol ("AMQP") Specification').