<html><head><meta name="color-scheme" content="light dark"></head><body><pre style="word-wrap: break-word; white-space: pre-wrap;">NAME
    AnyEvent::RabbitMQ::Simple - Easy to use asynchronous AMQP client

VERSION
    version 0.02

SYNOPSIS
        use strict;
        use warnings;
        use AnyEvent::RabbitMQ::Simple;

        # create main loop
        my $loop = AE::cv;

        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            host       =&gt; '127.0.0.1',
            port       =&gt; 5672,
            user       =&gt; 'username',
            pass       =&gt; 'password',
            vhost      =&gt; '/',
            timeout    =&gt; 1,
            tls        =&gt; 0,
            verbose    =&gt; 0,
            confirm_publish =&gt; 1,
            prefetch_count =&gt; 10,

            failure_cb =&gt; sub {
                my ($event, $details, $why) = @_;
                if ( ref $why ) {
                    my $method_frame = $why-&gt;method_frame;
                    $why = $method_frame-&gt;reply_text;
                }
                $loop-&gt;croak("[ERROR] $event($details): $why" );
            },

            # routing layout
            # [========== exchanges ===================] [===== queues ==============]
            # [             (type/routing key)         ] [        (routing key) ]
            #  logger ----------&gt; stats --------------&gt;   stats-logs
            #   |(fanout)           (direct)                (mail.stats)
            #   |  |
            #   |  | \----------&gt; errors -------------&gt;   ftp-error-logs
            #   |  |              | (topic:*.error.#)       (ftp.error.#)
            #   |  |              |
            #   |  |              \-------------------&gt;   mail-error-logs
            #   |  |                                        (mail.error.#)
            #   |  |
            #   |   \-----------&gt; info ---------------&gt;   info-logs
            #   |                   (topic:*.info.#)        (*.info.#)
            #   |
            #    \------------------------------------&gt;   debug-queue


            # declare exchanges
            exchanges =&gt; [
                'logger' =&gt; {
                    durable =&gt; 0,
                    type =&gt; 'fanout',
                    internal =&gt; 0,
                    auto_delete =&gt; 1,
                },
                'stats' =&gt; {
                    durable =&gt; 0,
                    type =&gt; 'direct',
                    internal =&gt; 0,
                    auto_delete =&gt; 1,
                },
                'errors' =&gt; {
                    durable =&gt; 0,
                    type =&gt; 'topic',
                    internal =&gt; 0,
                    auto_delete =&gt; 1,
                },
                'info' =&gt; {
                    durable =&gt; 0,
                    type =&gt; 'topic',
                    internal =&gt; 0,
                    auto_delete =&gt; 1,
                },
            ],

            # declare queues
            queues =&gt; [
                'debug-queue' =&gt; {
                    durable =&gt; 0,
                    auto_delete =&gt; 1,
                },
                'stats-logs' =&gt; {
                    durable =&gt; 0,
                    auto_delete =&gt; 1,
                },
                'ftp-error-logs' =&gt; {
                    durable =&gt; 0,
                    auto_delete =&gt; 1,
                },
                'mail-error-logs' =&gt; {
                    durable =&gt; 0,
                    auto_delete =&gt; 1,
                },
                'info-logs' =&gt; {
                    durable =&gt; 0,
                    auto_delete =&gt; 1,
                },
            ],

            # exchange to exchange bindings, with optional routing key
            bind_exchanges =&gt; [
                { 'stats'   =&gt;   'logger'                 },
                { 'errors'  =&gt; [ 'logger', '*.error.#' ]  },
                { 'info'    =&gt; [ 'logger', '*.info.#'  ]  },
            ],


            # queue to exchange bindings, with optional routing key
            bind_queues =&gt; [
                { 'debug-queue'     =&gt;   'logger'                   },
                { 'ftp-error-logs'  =&gt; [ 'errors', 'ftp.error.#'  ] },
                { 'mail-error-logs' =&gt; [ 'errors', 'mail.error.#' ] },
                { 'info-logs'       =&gt; [ 'info',   'info.#'       ] },
                { 'stats-logs'      =&gt; [ 'stats',  'mail.stats'   ] },
            ],

        );

        # publisher timer
        my $t;

        # connect and set up channel
        my $conn = $rmq-&gt;connect();
        $conn-&gt;cb(
            sub {
                print "waiting for channel..\n";
                my $channel = shift-&gt;recv or $loop-&gt;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-&gt;consume(
                queue =&gt; $queue,
                no_ack =&gt; 0,
                on_success =&gt; sub {
                    my $frame = shift;
                    $consumer_tag = $frame-&gt;method_frame-&gt;consumer_tag;
                    print "************* consuming from $queue with $consumer_tag\n";
                },
                on_consume =&gt; sub {
                    my $res = shift;
                    my $body = $res-&gt;{body}-&gt;payload;
                    print "+++++++++++++ consumed($queue): $body\n";
                    $channel-&gt;ack(
                        delivery_tag =&gt; $res-&gt;{deliver}-&gt;method_frame-&gt;delivery_tag
                    );
                },
                on_failure =&gt; sub {
                    print "************* failed to consume($queue)\n";
                }
            );
        }

        # randomly generates routing key and message body
        sub publish {
            my ($channel, $msg) = @_;

            unless ( $channel-&gt;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-&gt;publish(
                routing_key =&gt; $routing_key,
                exchange =&gt; 'logger',
                body =&gt; $msg,
                on_ack =&gt; sub {
                    print "------- published: $msg\n";
                },
                on_return =&gt; sub {
                    print "************* failed to publish: $msg\n";
                }
            );
        }

        # wait forever or die on error
        my $done = $loop-&gt;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-&gt;new(
            ...
        );

    Returns configured object using following parameters:

   host
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            host =&gt; '127.0.0.1', # default
            ...
        );

    Host IP.

   port
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            port =&gt; 5672, # default
            ...
        );

    Port number.

   vhost
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            vhost =&gt; '/', # default
            ...
        );

    Virtual host namespace.

   user
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            user =&gt; 'guest', # default
            ...
        );

    User name.

   pass
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            pass =&gt; 'guest', # default
            ...
        );

    Password.

   tune
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            tune =&gt; {
                heartbeat =&gt; $connection_heartbeat,
                channel_max =&gt; $max_channel_number,
                frame_max =&gt; $max_frame_size
            },
            ...
        );

    Optional connection tuning options.

   timeout
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            timeout =&gt; 0, # default
            ...
        );

    Connection timeout.

   tls
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            tls =&gt; 0, # default
            ...
        );

    Use TLS.

   verbose
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            verbose =&gt; 0, # default
            ...
        );

    Turn on protocol debug.

   confirm_publish
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            confirm_publish =&gt; 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-&gt;new(
            prefetch_count =&gt; 0, # default
            ...
        );

    Specify the number of prefetched messages when consuming from the
    channel.

   exchange
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            exchange =&gt; '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-&gt;new(
            exchanges =&gt; [
                'name_of_exchange' =&gt; {
                    durable =&gt; 1,
                    type =&gt; '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-&gt;new(
            queue =&gt; '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-&gt;gen_queue;

    See "declare_queue" in AnyEvent::RabbitMQ::Channel for details.

   queues
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            queues =&gt; [
                'name_of_queue' =&gt; {
                    durable =&gt; 1,
                    no_ack =&gt; 0,
                    ... # other queue configuration parameters
                },
                ...
            ],
            ...
        );

    Optional list of queues to declare with their configuration options.

    See "declare_queue" in AnyEvent::RabbitMQ::Channel for details.

   bind_exchanges
        my $rmq = AnyEvent::RabbitMQ::Simple-&gt;new(
            bind_exchanges =&gt; [
                # without routing key
                { 'destination1' =&gt; 'source' },

                # with routing key
                { 'destination2'  =&gt; [ '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-&gt;new(
            bind_queues =&gt; [
                # without routing key
                { 'queue1' =&gt; 'exchange' },

                # with routing key
                { 'queue2'  =&gt; [ '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-&gt;new(
            failure_cb =&gt; sub {
                my ($event, $details, $why) = @_;
                if ( ref $why ) {
                    my $method_frame = $why-&gt;method_frame;
                    $why = $method_frame-&gt;reply_text;
                }
                $loop-&gt;croak("[ERROR] $event($details): $why" );
            },
            ...
        );

    Required catch-all 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-&gt;connect();
        $conn-&gt;cb(
            sub {
                my $channel = shift-&gt;recv or $loop-&gt;croak("Could not open channel");

                ...
            }
        );

    Returns the AnyEvent condvar that returns AnyEvent::RabbitMQ::Channel
    object after all the configuration steps were successful.

  disconnect
        $rmq-&gt;disconnect();

    Disconnects from RabbitMQ server.

  gen_queue
        my $gen_queue = $rmq-&gt;gen_queue;

    Name of the generated queue if no queues were declared (or queue with
    empty name has been specified).

SEE ALSO
    *   AnyEvent::RabbitMQ

    *   &lt;https://www.rabbitmq.com/&gt;

AUTHOR
    Alex J. G. BurzyĹ„ski &lt;ajgb@cpan.org&gt;

COPYRIGHT AND LICENSE
    This software is copyright (c) 2016 by Alex J. G. BurzyĹ„ski
    &lt;ajgb@cpan.org&gt;.

    This is free software; you can redistribute it and/or modify it under
    the same terms as the Perl 5 programming language system itself.

</pre></body></html>