NAME

Mojo::RabbitMQ::Client - Mojo::IOLoop based RabbitMQ client

SYNOPSIS

use Mojo::RabbitMQ::Client;

# Supply URL according to (https://www.rabbitmq.com/uri-spec.html)
my $client = Mojo::RabbitMQ::Client->new(
  url => 'amqp://guest:guest@127.0.0.1:5672/');

# Catch all client related errors
$client->catch(sub { warn "Some error caught in client"; });

# When connection is in Open state, open new channel
$client->on(
  open => sub {
    my ($client) = @_;

    # Create a new channel with auto-assigned id
    my $channel = Mojo::RabbitMQ::Client::Channel->new();

    $channel->catch(sub { warn "Error on channel received"; });

    $channel->on(
      open => sub {
        my ($channel) = @_;
        $channel->qos(prefetch_count => 1)->deliver;

        # Publish some example message to test_queue
        my $publish = $channel->publish(
          exchange    => 'test',
          routing_key => 'test_queue',
          body        => 'Test message',
          mandatory   => 0,
          immediate   => 0,
          header      => {}
        );
        # Deliver this message to server
        $publish->deliver;

        # Start consuming messages from test_queue
        my $consumer = $channel->consume(queue => 'test_queue');
        $consumer->on(message => sub { say "Got a message" });
        $consumer->deliver;
      }
    );
    $channel->on(close => sub { $log->error('Channel closed') });

    $client->open_channel($channel);
  }
);

# Start connection
$client->connect();

# Start Mojo::IOLoop if not running already
Mojo::IOLoop->start unless Mojo::IOLoop->is_running;

CONSUMER

use Mojo::RabbitMQ::Client;
my $consumer = Mojo::RabbitMQ::Client->consumer(
  url      => 'amqp://guest:guest@127.0.0.1:5672/?exchange=mojo&queue=mojo',
  defaults => {
    qos      => {prefetch_count => 1},
    queue    => {durable        => 1},
    consumer => {no_ack         => 0},
  }
);

$consumer->catch(sub { die "Some error caught in Consumer" } );
$consumer->on('success' => sub { say "Consumer ready" });
$consumer->on(
  'message' => sub {
    my ($consumer, $message) = @_;

    $consumer->channel->ack($message)->deliver;
  }
);
$consumer->start();

Mojo::IOLoop->start unless Mojo::IOLoop->is_running;

PUBLISHER

use Mojo::RabbitMQ::Client;
my $publisher = Mojo::RabbitMQ::Client->publisher(
  url => 'amqp://guest:guest@127.0.0.1:5672/?exchange=mojo&routing_key=mojo'
);

$publisher->publish('plain text');

$publisher->publish(
  {encode => { to => 'json'}},
  routing_key => 'mojo_mq'
)->then(sub {
  say "Message published";
})->catch(sub {
  die "Publishing failed"
})->wait;

DESCRIPTION

Mojo::RabbitMQ::Client is a rewrite of AnyEvent::RabbitMQ to work on top of Mojo::IOLoop.

EVENTS

Mojo::RabbitMQ::Client inherits all events from Mojo::EventEmitter and can emit the following new ones.

connect

$client->on(connect => sub {
  my ($client, $stream) = @_;
  ...
});

Emitted when TCP/IP connection with RabbitMQ server is established.

open

$client->on(open => sub {
  my ($client) = @_;
  ...
});

Emitted AMQP protocol Connection.Open-Ok method is received.

close

$client->on(close => sub {
  my ($client) = @_;
  ...
});

Emitted on reception of Connection.Close-Ok method.

disconnect

$client->on(close => sub {
  my ($client) = @_;
  ...
});

Emitted when TCP/IP connection gets disconnected.

ATTRIBUTES

Mojo::RabbitMQ::Client has following attributes.

tls

my $tls = $client->tls;
$client = $client->tls(1)

Force secure connection. Default is disabled (0).

user

my $user = $client->user;
$client  = $client->user('guest')

Sets username for authorization, by default it's not defined.

pass

my $pass = $client->pass;
$client  = $client->pass('secret')

Sets user password for authorization, by default it's not defined.

host

my $host = $client->host;
$client  = $client->host('localhost')

Hostname or IP address of RabbitMQ server. Defaults to localhost.

port

my $port = $client->port;
$client  = $client->port(1234)

Port on which RabbitMQ server listens for new connections. Defaults to 5672, which is standard RabbitMQ server listen port.

vhost

my $vhost = $client->vhost;
$client  = $client->vhost('/')

RabbitMQ virtual server to user. Default is /.

params

my $params = $client->params;
$client  = $client->params(Mojo::Parameters->new('verify=1'))

Sets additional parameters for connection. Default is not defined.

For list of supported parameters see "SUPPORTED QUERY PARAMETERS".

url

my $url = $client->url;
$client = $client->url('amqp://...');

Sets all connection parameters in one string, according to specification from https://www.rabbitmq.com/uri-spec.html.

amqp_URI       = "amqp[s]://" amqp_authority [ "/" vhost ] [ "?" query ]

amqp_authority = [ amqp_userinfo "@" ] host [ ":" port ]

amqp_userinfo  = username [ ":" password ]

username       = *( unreserved / pct-encoded / sub-delims )

password       = *( unreserved / pct-encoded / sub-delims )

vhost          = segment

heartbeat_timeout

my $timeout = $client->heartbeat_timeout;
$client     = $client->heartbeat_timeout(180);

Heartbeats are use to monitor peer reachability in AMQP. Default value is 60 seconds, if set to 0 no heartbeats will be sent.

connect_timeout

my $timeout = $client->connect_timeout;
$client     = $client->connect_timeout(5);

Connection timeout used by Mojo::IOLoop::Client. Defaults to environment variable MOJO_CONNECT_TIMEOUT or 10 seconds if nothing else is set.

max_channels

my $max_channels = $client->max_channels;
$client          = $client->max_channels(10);

Maximum number of channels allowed to be active. Defaults to 0 which means no implicit limit.

When you try to call add_channel over limit an error will be emitted on channel saying that: Maximum number of channels reached.

STATIC METHODS

consumer

my $client = Mojo::RabbitMQ::Client->consumer(...)

Shortcut for creating Mojo::RabbitMQ::Client::Consumer.

publisher

my $client = Mojo::RabbitMQ::Client->publisher(...)

Shortcut for creating Mojo::RabbitMQ::Client::Publisher.

METHODS

Mojo::RabbitMQ::Client inherits all methods from Mojo::EventEmitter and implements the following new ones.

connect

$client->connect();

Tries to connect to RabbitMQ server and negotiate AMQP protocol.

close

$client->close();

param

my $param = $client->param('name');
$client   = $client->param(name => 'value');

add_channel

my $channel = Mojo::RabbitMQ::Client::Channel->new();
...
$channel    = $client->add_channel($channel);
$channel->open;

open_channel

my $channel = Mojo::RabbitMQ::Client::Channel->new();
...
$client->open_channel($channel);

delete_channel

my $removed = $client->delete_channel($channel->id);

SUPPORTED QUERY PARAMETERS

There's no formal specification, nevertheless a list of common parameters recognized by officially supported RabbitMQ clients is maintained here: https://www.rabbitmq.com/uri-query-parameters.html.

Some shortcuts are also supported, you'll find them in parenthesis.

Aliases are less significant, so when both are specified only primary value will be used.

cacertfile (ca)

Path to Certificate Authority file for TLS.

certfile (cert)

Path to the client certificate file for TLS.

keyfile (key)

Path to the client certificate private key file for TLS.

fail_if_no_peer_cert (verify)

TLS verification mode, defaults to 0x01 on the client-side if a certificate authority file has been provided, or 0x00 otherwise.

auth_mechanism

Currently only AMQPLAIN is supported, so this parameter is ignored.

heartbeat

Sets requested heartbeat timeout, just like heartbeat_timeout attribute.

connection_timeout (timeout)

Sets connection timeout - see connection_timeout attribute.

channel_max

Sets maximum number of channels - see max_channels attribute.

COPYRIGHT AND LICENSE

This program is free software, you can redistribute it and/or modify it under the terms of the Artistic License version 2.0.

To install Mojo::RabbitMQ::Client, copy and paste the appropriate command in to your terminal.

cpanm

cpanm Mojo::RabbitMQ::Client

CPAN shell

perl -MCPAN -e shell
install Mojo::RabbitMQ::Client

For more information on module installation, please visit the detailed CPAN module installation guide.

	Global
`s`	Focus search bar
`?`	Bring up this help dialog

	GitHub
`g` `p`	Go to pull requests
`g` `i`	go to github issues (only if github is preferred repository)

	POD
`g` `a`	Go to author
`g` `c`	Go to changes
`g` `i`	Go to issues
`g` `d`	Go to dist
`g` `r`	Go to repository/SCM
`g` `s`	Go to source
`g` `b`	Go to file browse

	Search terms
module: (e.g. module:Plugin)
distribution: (e.g. distribution:Dancer auth)
author: (e.g. author:SONGMU Redis)
version: (e.g. version:1.00)

NAME

SYNOPSIS

CONSUMER

PUBLISHER

DESCRIPTION

EVENTS

connect

open

close

disconnect

ATTRIBUTES

tls

user

pass

host

port

vhost

params

url

heartbeat_timeout

connect_timeout

max_channels

STATIC METHODS

consumer

publisher

METHODS

connect

close

param

add_channel

open_channel

delete_channel

SUPPORTED QUERY PARAMETERS

cacertfile (ca)

certfile (cert)

keyfile (key)

fail_if_no_peer_cert (verify)

auth_mechanism

heartbeat

connection_timeout (timeout)

channel_max

SEE ALSO

COPYRIGHT AND LICENSE

Module Install Instructions