view release on metacpan or  search on metacpan

lib/Amazon/SQS/Consumer.pm  view on Meta::CPAN

=item queue (required)

The URL of the queue to receive messages from.

=item wait_seconds (optional)

The number of seconds to wait for a new message when the queue is empty.

=item debug (optional)

A flag to turn on debugging. It is turned off by default.

=back

=cut

sub new {
	my $class = shift;
	my %args = @_;

	my $me = \%args;
	bless $me, $class;
	$me->initialize;
	return $me;
}

sub initialize {
	my $me = shift;

	$me->{n_messages} ||= DEFAULT_N_MESSAGES;
	$me->{wait_seconds} ||= DEFAULT_WAIT_SECONDS;
	$me->SUPER::initialize;
}

=head2 next()

This will receive a message from this Publisher's queue. When the queue is empty it will wait a new message for wait_seconds seconds.

=cut

sub next {
	my $me = shift;

	# If we're done with the previous message, delete it
	$me->delete_previous();

	if ( @ARGV ) {
		$me->{messages} = [ map { MessageId => undef, Body => $_ }, @ARGV ];
		undef @ARGV;
		$me->{no_loop} = 't';
	}

	my $seconds_to_wait = $me->{wait_seconds};
	do {

		# If there no messages in the cache, get some from the queue
		$me->{messages} = $me->receive_messages(
			Queue => $me->{queue},
			MaxNumberOfMessages => $me->{n_messages},
			defined $me->{timeout} ? ( VisibilityTimeout => $me->{timeout} ) : ()
		) unless defined $me->{messages} && @{$me->{messages}} or $me->{no_loop};

		# If there's a message in the cache, return it
		if ( my $message = shift @{$me->{messages}} ) {
			$me->{DeleteMessageHandle} = $message->{ReceiptHandle};
			my $object;
			eval {
				my $body = $message->{Body};
				$body = encode_utf8( $body ) if is_utf8( $body );
				$object = decode_json $body;
			};
			if ( $@ ) {
				say "left bad message in queue; could not decode JSON from $message->{Body}: $@";
			} else {
				return $object;
			}
		} elsif ( $me->{no_loop} ) {
			$seconds_to_wait = 0;
		} else {
			# Otherwise, wait a few seconds and try again
			say "waiting $seconds_to_wait seconds for new messages"
				if $seconds_to_wait == $me->{wait_seconds};
			sleep SECONDS_BETWEEN_TRIES;
			$seconds_to_wait -= SECONDS_BETWEEN_TRIES;
		}

	} while ( $me->{forever} or $seconds_to_wait > 0 );

	# If we've retried for a while and gotten no messages, give up
	return undef;

}

sub delete_previous {
	my $me = shift;

	if ( $me->{DeleteMessageHandle} ) {
		say "deleting message $me->{DeleteMessageHandle}" if $me->{debug};
		$me->delete_message( Queue => $me->{queue}, ReceiptHandle => $me->{DeleteMessageHandle} );
	}
}

sub defer { delete $_[0]->{DeleteMessageHandle} }


=head1 AUTHOR

Nic Wolff, <nic@angel.net>

=head1 BUGS

Please report any bugs or feature requests to C<bug-amazon-sqs-producerconsumer at rt.cpan.org>, or through
the web interface at L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=Amazon-SQS-ProducerConsumer>.  I will be notified, and then you'll
automatically be notified of progress on your bug as I make changes.




=head1 SUPPORT