view release on metacpan or  search on metacpan

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

package AnyEvent::RabbitMQ;

use strict;
use warnings;
use Carp qw(confess croak);
use Scalar::Util qw(refaddr);
use List::MoreUtils qw(none);
use Devel::GlobalDestruction;
use File::ShareDir;
use Readonly;
use Scalar::Util qw/ weaken /;

require Data::Dumper;
sub Dumper {
    local $Data::Dumper::Terse = 1;
    local $Data::Dumper::Indent = 1;
    local $Data::Dumper::Useqq = 1;
    local $Data::Dumper::Deparse = 1;
    local $Data::Dumper::Quotekeys = 0;
    local $Data::Dumper::Sortkeys = 1;
    &Data::Dumper::Dumper
}

use AnyEvent::Handle;
use AnyEvent::Socket;

use Net::AMQP 0.06;
use Net::AMQP::Common qw(:all);

use AnyEvent::RabbitMQ::Channel;
use AnyEvent::RabbitMQ::LocalQueue;

use namespace::clean;

our $VERSION = '1.22'; # VERSION

use constant {
    _ST_CLOSED => 0,
    _ST_OPENING => 1,
    _ST_OPEN => 2,
    _ST_CLOSING => 3,
};

Readonly my $DEFAULT_AMQP_SPEC
    => File::ShareDir::dist_dir("AnyEvent-RabbitMQ") . '/fixed_amqp0-9-1.xml';

Readonly my $DEFAULT_CHANNEL_MAX => 2**16-1;

sub new {
    my $class = shift;
    return bless {
        verbose            => 0,
        @_,
        _state             => _ST_CLOSED,
        _queue             => AnyEvent::RabbitMQ::LocalQueue->new,
        _last_chan_id      => 0,
        _channels          => {},
        _login_user        => '',
        _server_properties => {},
        _frame_max         => undef,
        _body_max          => undef,
        _channel_max       => undef,
    }, $class;
}

sub verbose {
    my $self = shift;
    @_ ? ($self->{verbose} = shift) : $self->{verbose}
}

sub is_open {
    my $self = shift;
    $self->{_state} == _ST_OPEN
}

sub channels {
    my $self = shift;
    return $self->{_channels};
}

sub _delete_channel {