view release on metacpan or  search on metacpan

xt/05_multi_channel.t  view on Meta::CPAN

    return $ch;
}

sub declare_queue {
    my ($ch, $queue,) = @_;

    my $done = AnyEvent->condvar;
    $ch->declare_queue(
        queue      => $queue,
        on_success => sub {$done->send(1)},
        on_failure => sub {$done->send()},
    );
    die 'Declare queue failure' if !$done->recv;

    return;
}

sub delete_queue {
    my ($ch, $queue,) = @_;

    my $done = AnyEvent->condvar;
    $ch->delete_queue(
        queue      => $queue,
        on_success => sub {$done->send(1)},
        on_failure => sub {$done->send()},
    );
    die 'Delete queue failure' if !$done->recv;

    return;
}

sub consume {
    my ($ch, $queue, $handle_consume, $handle_cancel,) = @_;

    my $done = AnyEvent->condvar;
    $ch->consume(
        queue      => $queue,
        on_success => sub {$done->send(1)},
        on_failure => sub {$done->send()},
        on_consume => $handle_consume,
        on_cancel  => $handle_cancel,
    );
    die 'Consume failure' if !$done->recv;

    return;
}

sub publish {
    my ($ch, $queue, $message,) = @_;

    $ch->publish(
        routing_key => $queue,
        body        => $message,
        mandatory   => 1,
    );

    return;
}

sub handle_close {
    my $method_frame = shift->method_frame;
    die $method_frame->reply_code, $method_frame->reply_text
      if $method_frame->reply_code;
}