view release on metacpan or  search on metacpan

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

=head1 SYNOPSIS

  use Amazon::SQS::Producer;

  my $out_queue = new Amazon::SQS::Producer
    AWSAccessKeyId => 'PUBLIC_KEY_HERE',
    SecretAccessKey => 'SECRET_KEY_HERE',
    queue => 'YourOutputQueue',
    consumer => 'ConsumerForOutputQueue';

  $out_queue->publish(
    $existingObjectRef,
    url => $enclosure_URL,
    pubdate => $pubDate,
    title => $title,
    description => $description,
    rss_guid => $guid,
  );

=head1 METHODS

=head2 new(%params)

This is the constructor, it will return you an Amazon::SQS::Producer object to work with.  It takes these parameters:

=over

=item AWSAccessKeyId (required)

Your AWS access key.

=item SecretAccessKey (required)

Your secret key, WARNING! don't give this out or someone will be able to use your account and incur charges on your behalf.

=item queue (required)

The URL of the queue to publish messages to.

=item consumer (optional)

The name of an executable that will consume messages from the queue we're publishing to. An instance will be launched after the each message is published, up to the maximum set by...

=item start_consumers (optional)

The maximum number of consumer instance to launch.

=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->{sleep_after_starting_consumer} = 2 if not exists $me->{sleep_after_starting_consumer};
	$me->SUPER::initialize;
}

=head2 publish(%params)

This will publish a message to this Publisher's queue, and start a consumer if this is the first message this Publisher has published. The message body will be a JSON representaton of the method's argument hash. If the first argument is a reference t...

=cut

sub publish {
	if ( ref $_[0] and ! $_[0]->{queue} ) { goto &fork_consumer }
	if ( ref $_[1] and $_[1]->{_chain_consumers} ) { goto &fork_consumer }

	my $me = shift;
	my $old_data = shift if ref $_[0];
	my $data = { %$old_data, @_ };
	my $encoded_data = encode_json $data;

	say "Queueing message: $encoded_data" if $data->{_debug};
	return if $data->{_test};

	my $retries;
	my $message_id;
	until (
		$message_id = $me->send_message(
			Queue => $me->{queue},
			MessageBody => $encoded_data,
		)
	) {
		say "couldn't queue message: ", $me->error;
		if ( $retries++ < MAX_RETRIES ) {
			say "trying again in $retries seconds";
			sleep $retries;
		} else {
			say "giving up trying to publish to queue $me->{queue} with message body: $encoded_data",
			return;
		}
	}

	if ( $me->{consumer} and $me->{started_consumers}++ < $me->{start_consumers} ) {
		my $pid = fork;
		if ( not defined $pid ) {
			say "couldn't fork";
		} elsif ( not $pid ) {
			close STDIN; open STDIN, '/dev/null';
			close STDOUT; open STDOUT, '/dev/null';
			close STDERR; open STDERR, '>>/tmp/getfeeds.log';
			sleep $me->{sleep_after_starting_consumer};
			exec $me->{consumer};
		} else {
			say "started consumer $me->{consumer} with PID $pid for queue $me->{queue}";
		}