view release on metacpan or  search on metacpan

lib/AnyMQ/Queue.pm  view on Meta::CPAN

=for stopwords

=head1 NAME

AnyMQ::Queue - AnyMQ Message Queue

=head1 SYNOPSIS

  my $channel = AnyMQ->topic('Foo');
  my $client = AnyMQ->new_listener($channel);

  $client->poll_once(sub {
      my @messages = @_;
      # ...
  });

  $client->poll(sub {
      my @messages = @_;
      # ...
  });

=head1 DESCRIPTION

An AnyMQ::Queue instance is a queue, each message put into the queue
can be consumed exactly once.  It's used as the client (or the
subscriber in terms of pub/sub) in L<AnyMQ>.  An AnyMQ::Queue can
subscribe to multiple L<AnyMQ::Topic>.

=head1 METHODS

=head2 subscribe($topic)

Subscribe to a L<AnyMQ::Topic> object.

=head2 poll($code_ref)

This is the event-driven poll mechanism, which accepts a callback.
Messages are streamed to C<$code_ref> passed in.

=head2 unpoll

Cancels a running L</poll>, which will result in L</on_timeout> being
called.

=head2 poll_once($code_ref, $timeout)

This method returns all messages since the last poll to C<$code_ref>.
It blocks for C<$timeout> seconds if there's currently no messages
available.

=head2 destroyed(BOOL)

Marking the current queue as destroyed or not.

=head2 timeout($seconds)

Timeout value for this queue.  Default is 55.

=head2 on_error(sub { my ($queue, $error, @msg) = @_; ... })

Sets the error handler invoked when C<poll> or C<poll_once> callbacks
fail.  By default the queue is marked as destroyed.  If you register
this error handler, you should call C<< $queue->destroyed(1) >> should you
wish to mark the queue as destroyed and reclaim resources.

Note that for queues that are currently C<poll>'ed, you may unset the
C<persistent> attribute to avoid the queue from being destroyed, and
can be used for further C<poll> or C<poll_once> calls.  In this case,
C<on_timeout> will be triggered if C<poll> or C<poll_once> is not
called after C<< $self->timeout >> seconds.

=head2 on_timeout(sub { my ($queue, $error) = @_; ... })

If a queue is not currently polled, this callback will be triggered
after C<< $self->timeout >> seconds.  The default behaviour is marking the
queue as destroyed.

=head2 append(@messages)

Append messages directly to the queue.  You probably want to use
C<publish> method of L<AnyMQ::Topic>

=head1 SEE ALSO

L<AnyMQ> L<AnyMQ::Topic>

=cut