File Coverage

blib/lib/IO/Iron/IronMQ/Queue.pm

Criterion	Covered	Total	%
statement	29	236	12.2
branch	0	36	0.0
condition			n/a
subroutine	12	29	41.3
pod	15	15	100.0
total	56	316	17.7

package IO::Iron::IronMQ::Queue;

## no critic (Documentation::RequirePodAtEnd)
## no critic (Documentation::RequirePodSections)
## no critic (Subroutines::RequireArgUnpacking)
## no critic (ControlStructures::ProhibitPostfixControls)

use 5.010_000;
use strict;
use warnings;

# Global creator
BEGIN {
	# Export Nothing
}

# Global destructor
END {
}


# ABSTRACT: IronMQ (Online Message Queue) Client (Queue).

our $VERSION = '0.13'; # VERSION: generated by DZP::OurPkgVersion


use Log::Any qw($log);
use Hash::Util 0.06 qw{lock_keys unlock_keys};
use Carp::Assert::More;
use English '-no_match_vars';
use Params::Validate qw(:all);

use IO::Iron::Common;
use IO::Iron::IronMQ::Api;
require IO::Iron::IronMQ::Message;



sub new {
	my $class = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
			'ironmq_client' => { type => OBJECT, }, # Reference to IronMQ client
			'connection' => { type => OBJECT, },  # Reference to REST client
		}
	);
	$log->tracef( 'Entering new(%s, %s)', $class, \%params );
	my $self;
	my @self_keys = ( ## no critic (CodeLayout::ProhibitQuotedWordLists)
		'ironmq_client',         # Reference to IronMQ client
		'name',                  # Queue name
		'connection',            # Reference to REST client
		'last_http_status_code', # After successfull network operation, the return value is here.
	);
	lock_keys( %{$self}, @self_keys );
	$self->{'ironmq_client'} = $params{'ironmq_client'};
	$self->{'name'} = $params{'name'};
	$self->{'connection'} = $params{'connection'};
	assert_isa( $self->{'connection'}, 'IO::Iron::Connection', 'self->{\'connection\'} is IO::Iron::Connection.' );
	assert_isa( $self->{'ironmq_client'}, 'IO::Iron::IronMQ::Client', 'self->{\'ironmq_client\'} is IO::Iron::IronMQ::Client.' );
	assert_nonblank( $self->{'name'}, 'self->{\'name\'} is defined and is not blank.' );

	unlock_keys( %{$self} );
	my $blessed_ref = bless $self, $class;
	lock_keys( %{$self}, @self_keys );

	$log->tracef( 'Exiting new: %s', $blessed_ref );
	return $blessed_ref;
}


sub size {
	my $self = shift;
	my %params = validate(
		@_, {
			# No parameters
		}
	);
	$log->tracef('Entering size().');

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_GET_QUEUE_INFO(),
		{ '{Queue Name}' => $queue_name, } );
	$self->{'last_http_status_code'} = $http_status_code;
	my $size = $response_message->{'queue'}->{'size'};
	$log->debugf( 'Queue size is %s.', $size );

	$log->tracef( 'Exiting size(): %s', $size );
	return $size;
}


sub post_messages {
	# TODO Limit the total size!
	my $self = shift;
	my %params = validate(
		@_, {
			'messages' => {
				type => ARRAYREF,
				callbacks => {
					'assert_class' => sub {
						foreach my $message (@{$_[0]}) {
							assert_isa( $message, 'IO::Iron::IronMQ::Message',
								'Message is IO::Iron::IronMQ::Message.' );
								# FIXME Do this better!
						}
						return 1;
					}
				}
			}, # one or more objects of class IO::Iron::IronMQ::Message.
		}
	);
	my @messages = @{$params{'messages'}};
	$log->tracef( 'Entering post_messages(%s)', @messages );

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
	my @message_contents;
	foreach my $message (@messages) {
		my ( $msg_body, $msg_delay, $msg_push_headers, ) = (
			$message->body(), $message->delay(), $message->push_headers(),
		);
		my $message_content = {};
		$message_content->{'body'}         = $msg_body;
		$message_content->{'delay'}        = $msg_delay if defined $msg_delay;
		$message_content->{'push_headers'} = $msg_push_headers if defined $msg_push_headers;
		# Gimmick to ensure the proper jsonization of numbers
		# Otherwise numbers might end up as strings.
		$message_content->{'delay'} += 0;

        CORE::push @message_contents, $message_content;
	}
	my %item_body = ( 'messages' => \@message_contents );

	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_POST_MESSAGES(),
		{
			'{Queue Name}' => $queue_name,
			'body'         => \%item_body,
		}
	  );
	$self->{'last_http_status_code'} = $http_status_code;

	my ( @ids, $msg );
	@ids = ( @{ $response_message->{'ids'} } );    # message ids.
	$msg = $response_message->{'msg'};    # Should be "Messages put on queue."
	$log->debugf( 'Pushed IronMQ Message(s) (queue name=%s; message id(s)=%s).',
		$self->{'name'}, ( join q{,}, @ids ) );
	if (wantarray) {
		$log->tracef( 'Exiting post_messages: %s', ( join q{:}, @ids ) );
		return @ids;
	}
	else {
		if ( scalar @messages == 1 ) {
			$log->tracef( 'Exiting post_messages: %s', $ids[0] );
			return $ids[0];
		}
		else {
			$log->tracef( 'Exiting post_messages: %s', scalar @ids );
			return scalar @ids;
		}
	}
}


sub reserve_messages {
	my $self = shift;
	my %params = validate(
		@_, {
			'n' => { type => SCALAR, optional => 1, },    # Number of messages to pull.
			'timeout' => { type => SCALAR, optional => 1, }, # When reading from queue, after timeout (in seconds), item will be placed back onto queue.
			'wait' => { type => SCALAR, optional => 1, }, # Seconds to long poll the queue.
            'delete' => { type => SCALAR, optional => 1, } # Do not put each message back on to the queue after reserving.
		}
	);
	assert_positive(wantarray, 'Method reserve_messages() only works in LIST context!');
	$log->tracef( 'Entering reserve_messages(%s)', \%params );

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
	my %item_body;
	$item_body{'n'}       = $params{'n'}+0        if $params{'n'};
	$item_body{'timeout'} = $params{'timeout'}+0  if $params{'timeout'};
	$item_body{'wait'}    = $params{'wait'}+0     if $params{'wait'};
	$item_body{'delete'}  = $params{'delete'}     if $params{'delete'};
	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_RESERVE_MESSAGES(),
		{
			'{Queue Name}' => $queue_name,
			'body'         => \%item_body,
		}
	  );
	$self->{'last_http_status_code'} = $http_status_code;

	my @pulled_messages;
	my $messages = $response_message->{'messages'};    # messages.
	foreach ( @{$messages} ) {
		my $msg = $_;
		$log->debugf( 'Pulled IronMQ Message (queue name=%s; message id=%s).',
			$self->{'name'}, $msg->{'id'} );
		my $message = IO::Iron::IronMQ::Message->new(
			'body'           => $msg->{'body'},
			'id'             => $msg->{'id'},
			'reserved_count' => $msg->{'reserved_count'},
			'reservation_id' => $msg->{'reservation_id'},
		);
		CORE::push @pulled_messages,
		  $message;    # using CORE routine, not this class' method.
	}
	$log->debugf( 'Reserved %d IronMQ Messages (queue name=%s).',
		scalar @pulled_messages, $self->{'name'} );
	$log->tracef( 'Exiting reserve_messages(): %s',
		@pulled_messages ? @pulled_messages : '[NONE]' );
	return @pulled_messages;
}


sub peek_messages {
	my $self = shift;
	my %params = validate(
		@_, {
			'n' => { type => SCALAR, optional => 1, }, # Number of messages to read.
		}
	);
	assert_positive(wantarray, 'Method peek_messages() only works in LIST context!');
	$log->tracef( 'Entering peek_messages(%s)', \%params );

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
	my %query_params;
	$query_params{'{n}'} = $params{'n'} if $params{'n'};
	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_PEEK_MESSAGES(),
		{
			'{Queue Name}' => $queue_name,
			%query_params
		}
	  );
	$self->{'last_http_status_code'} = $http_status_code;

	my @peeked_messages;
	my $messages = $response_message->{'messages'};    # messages.
	foreach ( @{$messages} ) {
		my $msg = $_;
		$log->debugf( 'peeked IronMQ Message (queue name=%s; message id=%s.',
			$self->{'name'}, $msg->{'id'} );
		my $message = IO::Iron::IronMQ::Message->new(
			'body'    => $msg->{'body'},
			'id'             => $msg->{'id'},
		);
		$message->reserved_count($msg->{'reserved_count'}) if $msg->{'reserved_count'};
		# When peeking, timeout is not returned
		# (it is irrelevent, because peeking does not reserve the message).
		push @peeked_messages, $message;
	}
	$log->tracef( 'Exiting peek_messages(): %s',
		@peeked_messages ? @peeked_messages : '[NONE]' );
	return @peeked_messages;
}


sub delete_message {
	my $self = shift;
	my %params = validate(
		@_, {
			'message' => {
				type => OBJECT,
                isa => 'IO::Iron::IronMQ::Message',
                optional => 0,
			},
            'subscriber_name' => {
                type => SCALAR,
                optional => 1,
            },
		}
	);
	$log->tracef( 'Entering delete(%s)', \%params );

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
    my $message = $params{'message'};
	my %item_body  = ( 'reservation_id' => $message->reservation_id(), );
    $item_body{'subscriber_name'} = $params{'subscriber_name'} if $params{'subscriber_name'};
	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_DELETE_MESSAGE(),
		{
			'{Queue Name}' => $queue_name,
			'{Message ID}' => $message->id(),
			'body'         => \%item_body,
		}
	  );
	$self->{'last_http_status_code'} = $http_status_code;

	my $msg = $response_message->{'msg'};    # Should be 'Deleted'
	$log->debugf( 'Deleted IronMQ Message (queue name=%s; message id=%s.',
		$queue_name, $params{'message'}->id() );
	$log->tracef( 'Exiting delete_message(): %s', 'undef' );
    return;
}


sub delete_messages {
	my $self = shift;
	# my %params = validate(
	# 	@_, {
	# 		'ids' => {
	# 			type => ARRAYREF,
	# 		}, # one or more id strings (alphanum text string).
	# 	}
	# );
    my @messages = validate_pos(@_, ( { type => OBJECT, isa => 'IO::Iron::IronMQ::Message', } ) x scalar @_);
	# my @message_ids = @{$params{'ids'}};
	assert_positive(scalar @messages, 'There is one or more messages.');
	$log->tracef( 'Entering delete_messages(%s)', \@messages );

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
	my %item_body  = ( 'ids' => [ ], );
    my @message_ids;
    foreach my $msg (@messages) {
        CORE::push @{$item_body{'ids'}}, { 'id' => $msg->id(), 'reservation_id' => $msg->reservation_id(), };
        CORE::push @message_ids, $msg->id();
    }

	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_DELETE_MESSAGES(),
		{
			'{Queue Name}' => $queue_name,
			'body'         => \%item_body,
		}
	  );
	$self->{'last_http_status_code'} = $http_status_code;

	my $msg = $response_message->{'msg'};    # Should be 'Deleted'
	$log->debugf( 'Deleted IronMQ Message(s) (queue name=%s; message id(s)=%s.',
		$queue_name, ( join q{,}, @message_ids ) );
	$log->tracef( 'Exiting delete_messages: %s', 'undef' );
    return;
}


sub touch_message {
	my $self = shift;
	my %params = validate(
		@_, {
			'message' => {
				type => OBJECT,
                isa => 'IO::Iron::IronMQ::Message',
                optional => 0,
			},
            'timeout' => {
                type => SCALAR,
                optional => 1,
            },
		}
	);
	$log->tracef( 'Entering touch_message(%s)', \%params );

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
    my $message = $params{'message'};
	my %item_body  = ( 'reservation_id' => $message->reservation_id(), );
    $item_body{'timeout'} = $params{'timeout'} if $params{'timeout'};
	my ( $http_status_code, $response_message ) = $connection->perform_iron_action(
        IO::Iron::IronMQ::Api::IRONMQ_TOUCH_MESSAGE(),
        {
            '{Queue Name}' => $queue_name,
            '{Message ID}' => $message->id(),
            'body'         => \%item_body,
        }
    );
	$self->{'last_http_status_code'} = $http_status_code;
    $message->reservation_id($response_message->{'reservation_id'});
	$log->debugf( 'Touched IronMQ Message (queue name=%s; message id=%s.',
		$queue_name, $message->id() );

	$log->tracef( 'Exiting touch_message(): %s', 'undef' );
	return;
}


sub release_message {
	my $self = shift;
	my %params = validate(
		@_, {
			'message' => {
				type => OBJECT,
                isa => 'IO::Iron::IronMQ::Message',
                optional => 0,
			},
			'delay' => { type => SCALAR, optional => 1, }, # Delay before releasing.
		}
	);
	assert_nonnegative_integer( $params{'delay'} ? $params{'delay'} : 0, 'Parameter delay is a non negative integer.' );
	$log->tracef( 'Entering release_message(%s)', \%params );

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
    my $message = $params{'message'};
	my %item_body  = ( 'reservation_id' => $message->reservation_id(), );
	$item_body{'delay'} = $params{'delay'} if $params{'delay'};
	# We do not give delay a default value (0); we let IronMQ use internal default values!
	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_RELEASE_MESSAGE(),
		{
			'{Queue Name}' => $queue_name,
			'{Message ID}'  => $message->id(),
			'body'         => \%item_body,
		}
	  );
	$self->{'last_http_status_code'} = $http_status_code;
	$log->debugf(
		'Released IronMQ Message(s) (queue name=%s; message id=%s; delay=%d)',
		$queue_name, $params{'id'}, $params{'delay'} ? $params{'delay'} : 0 );

	$log->tracef( 'Exiting release_message: %s', 1 );
	return 1;
}


sub clear_messages {
	my $self = shift;
	my %params = validate(
		@_, {
			# No parameters
		}
	);
	$log->tracef('Entering clear_messages()');

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
	my %item_body;
	my ( $http_status_code, $response_message ) =
	  $connection->perform_iron_action(
		IO::Iron::IronMQ::Api::IRONMQ_CLEAR_MESSAGES(),
		{
			'{Queue Name}' => $queue_name,
			'body'         => \%item_body,    # Empty body.
		}
	  );
	$self->{'last_http_status_code'} = $http_status_code;
	my $msg = $response_message->{'msg'};    # Should be 'Cleared'
	$log->debugf( 'Cleared IronMQ Message queue %s.', $queue_name );
	$log->tracef( 'Exiting clear_messages: %s', 'undef' );
	return;
}


sub get_push_statuses {
	my $self = shift;
	my %params = validate(
		@_, {
			'id' => { type => SCALAR, }, # message id.
		}
	);
	assert_positive(wantarray == 0, 'Method get_push_statuses() only works in SCALAR context!');
	assert_nonblank( $params{'id'}, 'Parameter id is a non null string.');
	$log->tracef('Entering get_push_statuses(%s)', \%params);

	my $queue_name = $self->name();
	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_GET_PUSH_STATUSES_FOR_A_MESSAGE(),
			{
				'{Queue Name}' => $queue_name,
				'{Message ID}'  => $params{'id'},
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;
	my $info = $response_message;
	$log->debugf('Returned push status for message %s.', $params{'id'});

	$log->tracef('Exiting get_push_statuses: %s', $info);
	return $info;
}


sub ironmq_client { return $_[0]->_access_internal('ironmq_client', $_[1]); }
sub name { return $_[0]->_access_internal('name', $_[1]); }
sub connection { return $_[0]->_access_internal('connection', $_[1]); }
sub last_http_status_code { return $_[0]->_access_internal('last_http_status_code', $_[1]); }

# TODO Move _access_internal() to IO::Iron::Common.

sub _access_internal {
	my ($self, $var_name, $var_value) = @_;
	$log->tracef('_access_internal(%s, %s)', $var_name, $var_value);
	if( defined $var_value ) {
		$self->{$var_name} = $var_value;
		return $self;
	}
	else {
		return $self->{$var_name};
	}
}

1;

__END__