view release on metacpan or  search on metacpan

lib/AnyEvent/RabbitMQ/Simple.pm  view on Meta::CPAN

        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;

=head1 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
C<on_success> callbacks.

=head1 METHODS

=head2 new

    my $rmq = AnyEvent::RabbitMQ::Simple->new(
        ...
    );

Returns configured object using following parameters:

=head3 host

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

Host IP.

=head3 port

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

Port number.

=head3 vhost

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

Virtual host namespace.

=head3 user

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

User name.

=head3 pass

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

Password.

=head3 tune

    my $rmq = AnyEvent::RabbitMQ::Simple->new(
        tune => {