NAME
AnyEvent::RabbitMQ::Simple - Easy to use asynchronous AMQP client
VERSION
version 0.01
SYNOPSIS
use strict;
use warnings;
use AnyEvent::RabbitMQ::Simple;
# create main loop
my $loop = AE::cv;
my $rmq = AnyEvent::RabbitMQ::Simple->new(
host => '127.0.0.1',
port => 5672,
user => 'username',
pass => 'password',
vhost => '/',
timeout => 1,
tls => 0,
verbose => 0,
confirm_publish => 1,
prefetch_count => 10,
failure_cb => sub {
my ($event, $details, $why) = @_;
if ( ref $why ) {
my $method_frame = $why->method_frame;
$why = $method_frame->reply_text;
}
$loop->croak("[ERROR] $event($details): $why" );
},
# routing layout
# [========== exchanges ===================] [===== queues ==============]
# [ (type/routing key) ] [ (routing key) ]
# logger ----------> stats --------------> stats-logs
# |(fanout) (direct) (mail.stats)
# | |
# | | \----------> errors -------------> ftp-error-logs
# | | | (topic:*.error.#) (ftp.error.#)
# | | |
# | | \-------------------> mail-error-logs
# | | (mail.error.#)
# | |
# | \-----------> info ---------------> info-logs
# | (topic:*.info.#) (*.info.#)
# |
# \------------------------------------> debug-queue
# declare exchanges
exchanges => [
'logger' => {
durable => 0,
type => 'fanout',
internal => 0,
auto_delete => 1,
},
'stats' => {
durable => 0,
type => 'direct',
internal => 0,
auto_delete => 1,
},
'errors' => {
durable => 0,
type => 'topic',
internal => 0,
auto_delete => 1,
},
'info' => {
durable => 0,
type => 'topic',
internal => 0,
auto_delete => 1,
},
],
# declare queues
queues => [
'debug-queue' => {
durable => 0,
auto_delete => 1,
},
'stats-logs' => {
durable => 0,
auto_delete => 1,
},
'ftp-error-logs' => {
durable => 0,
auto_delete => 1,
},
'mail-error-logs' => {
durable => 0,
auto_delete => 1,
},
'info-logs' => {
durable => 0,
auto_delete => 1,
},
],
# exchange to exchange bindings, with optional routing key
bind_exchanges => [
{ 'stats' => 'logger' },
{ 'errors' => [ 'logger', '*.error.#' ] },
{ 'info' => [ 'logger', '*.info.#' ] },
],
# queue to exchange bindings, with optional routing key
bind_queues => [
{ 'debug-queue' => 'logger' },
{ 'ftp-error-logs' => [ 'errors', 'ftp.error.#' ] },
{ 'mail-error-logs' => [ 'errors', 'mail.error.#' ] },
{ 'info-logs' => [ 'info', 'info.#' ] },
{ 'stats-logs' => [ 'stats', 'mail.stats' ] },
],
);
# publisher timer
my $t;
# connect and set up channel
my $conn = $rmq->connect();
$conn->cb(
sub {
print "waiting for channel..\n";
my $channel = shift->recv or $loop->croak("Could not open channel");
print "************* consuming\n";
for my $q ( qw( debug-queue ftp-error-logs mail-error-logs info-logs stats-logs ) ) {
consume($channel, $q);
}
print "************* starting publishing\n";
$t = AE::timer 0, 1.0, sub { publish($channel, "message prepared at ". scalar(localtime) ) };
}
);
# consumes from requested queue
sub consume {
my ($channel, $queue) = @_;
my $consumer_tag;
$channel->consume(
queue => $queue,
no_ack => 0,
on_success => sub {
my $frame = shift;
$consumer_tag = $frame->method_frame->consumer_tag;
print "************* consuming from $queue with $consumer_tag\n";
},
on_consume => sub {
my $res = shift;
my $body = $res->{body}->payload;
print "+++++++++++++ consumed($queue): $body\n";
$channel->ack(
delivery_tag => $res->{deliver}->method_frame->delivery_tag
);
},
on_failure => sub {
print "************* failed to consume($queue)\n";
}
);
}
# randomly generates routing key and message body
sub publish {
my ($channel, $msg) = @_;
unless ( $channel->is_open ) {
warn "Cannot publish, channel closed";
return;
}
my @system = qw( mail ftp web );
my @levels = qw( debug info error stats );
my $routing_key = $system[rand @system] .'.'. $levels[ rand @levels ];
$msg = sprintf("[%s] %s", uc($routing_key), $msg);
print "\n------- publishing: $msg\n";
$channel->publish(
routing_key => $routing_key,
exchange => 'logger',
body => $msg,
on_ack => sub {
print "------- published: $msg\n";
},
on_return => sub {
print "************* failed to publish: $msg\n";
}
);
}
# wait forever or die on error
my $done = $loop->recv;
DESCRIPTION
This module is meant to simplify the process of setting up the RabbitMQ
channel, so you can start publishing and/or consuming messages without
chaining "on_success" callbacks.
METHODS
new
my $rmq = AnyEvent::RabbitMQ::Simple->new(
...
);
Returns configured the object using following parameters:
host
my $rmq = AnyEvent::RabbitMQ::Simple->new(
host => '127.0.0.1', # default
...
);
Host IP.
port
my $rmq = AnyEvent::RabbitMQ::Simple->new(
port => 5672, # default
...
);
Port number.
vhost
my $rmq = AnyEvent::RabbitMQ::Simple->new(
vhost => '/', # default
...
);
Virtual host namespace.
user
my $rmq = AnyEvent::RabbitMQ::Simple->new(
user => 'guest', # default
...
);
User name.
pass
my $rmq = AnyEvent::RabbitMQ::Simple->new(
pass => 'guest', # default
...
);
Password.
tune
my $rmq = AnyEvent::RabbitMQ::Simple->new(
tune => {
heartbeat => $connection_heartbeat,
channel_max => $max_channel_number,
frame_max => $max_frame_size
},
...
);
Connection tuning options.
timeout
my $rmq = AnyEvent::RabbitMQ::Simple->new(
timeout => 0, # default
...
);
Connection timeout.
tls
my $rmq = AnyEvent::RabbitMQ::Simple->new(
tls => 0, # default
...
);
Use TLS.
verbose
my $rmq = AnyEvent::RabbitMQ::Simple->new(
verbose => 0, # default
...
);
Turn on protocol debug.
confirm_publish
my $rmq = AnyEvent::RabbitMQ::Simple->new(
confirm_publish => 0, # default
...
);
Turn on confirm mode on channel. If set it enables the "on_ack" callback
of channel's "publish" method.
prefetch_count
my $rmq = AnyEvent::RabbitMQ::Simple->new(
prefetch_count => 0, # default
...
);
Specify the number of prefetched messages when consuming from the
channel.
exchange
my $rmq = AnyEvent::RabbitMQ::Simple->new(
exchange => 'name_of_exchange',
...
);
Optional name of exchange to declare with its default configuration
options.
See "declare_exchange (%args)" in AnyEvent::RabbitMQ::Channel for
details.
exchanges
my $rmq = AnyEvent::RabbitMQ::Simple->new(
exchanges => [
'name_of_exchange' => {
durable => 1,
type => 'fanout',
... # other exchange configuration parameters
},
...
],
...
);
Optional list of exchanges to declare with their configuration options.
See "declare_exchange (%args)" in AnyEvent::RabbitMQ::Channel for
details.
queue
my $rmq = AnyEvent::RabbitMQ::Simple->new(
queue => 'name_of_queue',
...
);
Optional name of queue to declare with its default configuration
options.
If no queues were declared or empty name has been specified a unique
generated queue name will be available:
my $gen_queue = $rmq->gen_queue;
See "declare_queue" in AnyEvent::RabbitMQ::Channel for details.
queues
my $rmq = AnyEvent::RabbitMQ::Simple->new(
queues => [
'name_of_queue' => {
durable => 1,
no_ack => 0,
... # other queue configuration parameters
},
...
],
...
);
Optional list of queues to declare with their configuration options.
See "declare_queue" in AnyEvent::RabbitMQ::Channel for details.
gen_queue
my $gen_queue = $rmq->gen_queue;
Name of the generated queue if no queues were declared (or queue with
empty name has been specified).
bind_exchanges
my $rmq = AnyEvent::RabbitMQ::Simple->new(
bind_exchanges => [
# without routing key
{ 'destination1' => 'source' },
# with routing key
{ 'destination2' => [ 'source', 'routing_key' ] },
...
],
...
);
Optional list of exchange-to-exchange bindings.
See "bind_exchange" in AnyEvent::RabbitMQ::Channel for details.
bind_queues
my $rmq = AnyEvent::RabbitMQ::Simple->new(
bind_queues => [
# without routing key
{ 'queue1' => 'exchange' },
# with routing key
{ 'queue2' => [ 'exchange', 'routing_key' ] },
...
],
...
);
Optional list of queue-to-exchange bindings.
See "bind_queue" in AnyEvent::RabbitMQ::Channel for details.
failure_cb
my $rmq = AnyEvent::RabbitMQ::Simple->new(
failure_cb => sub {
my ($event, $details, $why) = @_;
if ( ref $why ) {
my $method_frame = $why->method_frame;
$why = $method_frame->reply_text;
}
$loop->croak("[ERROR] $event($details): $why" );
},
...
);
Generic error handling callback. The value of $event is one of:
ConnectOnFailure
ConnectOnReadFailure
ConnectOnReturn
ConnectOnClose
OpenChannelOnFailure
OpenChannelOnReturn
OpenChannelOnClose
DeclareExchangeOnFailure
Value of $details has following format: "name:$name_of_exchange".
BindExchangeOnFailure
Value of $details has following format:
"source:$name_of_source_exchange,
destination:$name_of_destination_exchange".
DeclareQueueOnFailure
Value of $details has following format: "name:$name_of_queue".
BindQueueOnFailure
Value of $details has following format: "queue:$name_of_queue,
exchange:$name_of_exchange".
ConfirmChannelOnFailure
QosChannelOnFailure
connect
my $conn = $rmq->connect();
$conn->cb(
sub {
my $channel = shift->recv or $loop->croak("Could not open channel");
...
}
);
Returns the AnyEvent condvar that returns AnyEvent::RabbitMQ::Channel
object after all the configuration steps were successful.
disconnect
$rmq->disconnect();
Disconnects from RabbitMQ server.
SEE ALSO
* AnyEvent::RabbitMQ
* <https://www.rabbitmq.com/>
AUTHOR
Alex J. G. Burzyński <ajgb@cpan.org>
COPYRIGHT AND LICENSE
This software is copyright (c) 2016 by Alex J. G. Burzyński
<ajgb@cpan.org>.
This is free software; you can redistribute it and/or modify it under
the same terms as the Perl 5 programming language system itself.