logo
Free, unlimited AI code reviews that run on commit
git-lrc git-lrc GitHub Install Now We'd appreciate a star git-lrc - Free, unlimited AI code reviews that run on commit | Product Hunt git-lrc - Free, unlimited AI code reviews that run on commit | Product Hunt

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

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: Maximumnumberofchannelsreached.

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.

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

Name

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

See Also

       Mojo::RabbitMQ::Client::Channel, Mojo::RabbitMQ::Client::Consumer, Mojo::RabbitMQ::Client::Publisher

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.

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
       Sets the AMQP authentication mechanism. Defaults to AMQPLAIN. AMQPLAIN and EXTERNAL are supported;
       EXTERNAL will only work if Mojo::RabbitMQ::Client does not need to do anything beyond passing along a
       username and password if specified.

   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.

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;

See Also