MyApp::Model

gistfile1.pl

package MyApp::Model;
use Moose;
use JSON::XS;
use AnyEvent::RabbitMQ;
use MooseX::ClassAttribute;
class_has queue => ( is => 'ro', builder => '_build_queue' );
has _channel => ( is => 'rw' );

sub _build_queue {
    my $self = shift;
    my $cv   = AE::cv;
    my $ar   = AnyEvent::RabbitMQ->new->load_xml_spec->connect(
        host       => 'localhost',
        port       => 5672,
        user       => 'guest',
        pass       => 'guest',
        vhost      => '/',
        on_success => sub {
            shift->open_channel(
                on_success => sub {
                    shift->declare_queue(
                        queue      => 'myapp.jobs',
                        durable    => 1,
                        on_success => $cv
                    );
                },
            );
        },
    );
    $cv->recv;
    return $ar;
}

sub channel {
    my ( $self, $cb ) = @_;
    return $cb->( $self->_channel ) if ( $self->_channel );
    $self->queue->open_channel(
        on_success => sub { $self->_channel(@_); $cb->(@_) } );
}

sub publish {
    my ( $self, $body ) = @_;
    $self->channel(
        sub {
            shift->publish(
                routing_key => 'myapp.jobs',
                body        => encode_json($body),
            );
        }
    );
}

sub new_queue {
    my $self  = shift;
    my $cv    = AE::cv;
    my $queue = $self->channel(
        sub { shift->declare_queue( durable => 1, on_success => $cv ) } );
    return $cv;
}

sub consume {
    my ( $self, $queue ) = @_;
    my $cv  = AE::cv;
    my $res = $self->channel(
        sub {
            shift->consume(
                queue      => $queue,
                on_consume => $cv,
                no_ack     => 1,
            );
        }
    );
    return $cv;
}