File Coverage

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

Criterion	Covered	Total	%
statement	44	261	16.8
branch	0	50	0.0
condition	0	3	0.0
subroutine	16	41	39.0
pod	14	14	100.0
total	74	369	20.0

package IO::Iron::IronMQ::Client;

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

use 5.010_000;
use strict;
use warnings;

# Global creator
BEGIN {
	use parent qw( IO::Iron::ClientBase ); # Inheritance
}

# Global destructor
END {
}


# ABSTRACT: IronMQ (Online Message Queue) Client.

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



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


use IO::Iron::IronMQ::Api;
use IO::Iron::Common;
require IO::Iron::Connection;
require IO::Iron::IronMQ::Queue;

# CONSTANTS for this package

# DEFAULTS
use Const::Fast;

# Service specific!
const my $DEFAULT_API_VERSION => '3';
const my $DEFAULT_HOST => 'mq-aws-us-east-1-1.iron.io';



sub new {
	my $class = shift;
	my %params = validate(
		@_, {
			map { $_ => { type => SCALAR, optional => 1 }, } IO::Iron::Common::IRON_CLIENT_PARAMETERS(), ## no critic (ValuesAndExpressions::ProhibitCommaSeparatedStatements)
		}
	);

	$log->tracef('Entering new(%s, %s)', $class, \%params);
	my $self = IO::Iron::ClientBase->new();
	# Add more keys to the self hash.
	my @self_keys = (
			'queues',        # References to all objects created of class IO::Iron::IronMQ::Queue.
			legal_keys(%{$self}),
	);
	unlock_keys(%{$self});
	lock_keys_plus(%{$self}, @self_keys);
	my $config = IO::Iron::Common::get_config(%params);
	$log->debugf('The config: %s', $config);
	$self->{'project_id'} = defined $config->{'project_id'} ? $config->{'project_id'} : undef;
	$self->{'queues'} = [];
	assert_nonblank( $self->{'project_id'}, 'self->{project_id} is not defined or is blank');

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

	# Set up the connection client
	my $connection = IO::Iron::Connection->new( {
		'project_id' => $config->{'project_id'},
		'token' => $config->{'token'},
		'host' => defined $config->{'host'} ? $config->{'host'} : $DEFAULT_HOST,
		'protocol' => $config->{'protocol'},
		'port' => $config->{'port'},
		'api_version' => defined $config->{'api_version'} ? $config->{'api_version'} : $DEFAULT_API_VERSION,
		'timeout' => $config->{'timeout'},
		'connector' => $params{'connector'},
		}
	);
	$self->{'connection'} = $connection;
	$log->debugf('IronMQ Connection created with config: (project_id=%s; token=%s; host=%s; timeout=%s).', $config->{'project_id'}, $config->{'token'}, $config->{'host'}, $config->{'timeout'});
	$log->tracef('Exiting new: %s', $self);
	return $self;
}


sub get_queue {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
		}
	);
	$log->tracef('Entering get_queue(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_GET_QUEUE_INFO(),
			{ '{Queue Name}' => $params{'name'}, }
		);
	$self->{'last_http_status_code'} = $http_status_code;
    my $new_queue = $response_message->{'queue'};
	my $get_queue_name = $new_queue->{'name'};
	my $queue = IO::Iron::IronMQ::Queue->new({
		'ironmq_client' => $self, # Pass a reference to the parent object.
		'name' => $get_queue_name,
		'connection' => $connection,
	});
	$log->debugf('Created a new IO::Iron::IronMQ::Queue object (queue name=%s.', $get_queue_name);
	$log->tracef('Exiting get_queue: %s', $queue);
	return $queue;
}


sub get_queues {
	my $self = shift;
	my %params = validate(
		@_, {
            'prefix' => { type => SCALAR, optional => 1, },
		}
	);
	$log->tracef('Entering get_queues()');

    my $previous = q{};
    my @queue_names;
    while( my @names = $self->list_queues( 'per_page' => 30, 'previous' => $previous ) ) {
        push @queue_names, @names;
        $previous = $names[-1];
    }
    $log->debugf('Got a list of %d queue names.', scalar @queue_names);

	my @queues;
    foreach my $queue_name (@queue_names) {
        push @queues, $self->get_queue('name' => $queue_name);
    }

	$log->tracef('Exiting get_queues: %s', \@queues);
	return @queues;
}


sub create_and_get_queue {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, callbacks => {
                    'RFC 3986 reserved character check' => sub { return ! IO::Iron::Common::contains_rfc_3986_res_chars(shift) },
                }}, # queue name.
			'message_timeout' => { type => SCALAR, optional => 1, },
			'message_expiration' => { type => SCALAR, optional => 1, },
			'type' => { type => SCALAR, optional => 1,
                regex => qr/^(?:multicast|unicast|pull)$/msx, ## no critic (Variables::ProhibitPunctuationVars)
            },
			'push' => { type => HASHREF, optional => 1, },
			'dead_letter' => { type => HASHREF, optional => 1, },
		}
	);
	$log->tracef('Entering create_queue(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

	my $connection = $self->{'connection'};
	my %item_body;
	$item_body{'message_timeout'} = $params{'message_timeout'} if ($params{'message_timeout'});
	$item_body{'message_expiration'} = $params{'message_expiration'} if ($params{'message_expiration'});
	$item_body{'type'} = $params{'type'} if ($params{'type'});
	$item_body{'push'} = $params{'push'} if ($params{'push'});
	$item_body{'dead_letter'} = $params{'dead_letter'} if ($params{'dead_letter'});
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_CREATE_QUEUE(),
			{
				'{Queue Name}' => $params{'name'},
				'body'         => \%item_body,
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;
    my $queue = IO::Iron::IronMQ::Queue->new({
            'ironmq_client' => $self, # Pass a reference to the parent object.
            'connection' => $connection,
            'name' => $params{'name'},
    });
    $log->debugf('Created a new IO::Iron::IronMQ::Queue object (queue name=%s).', $queue->name());
	$log->tracef('Exiting create_queue: %s', $queue);
	return $queue;
}


sub create_queue {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, callbacks => {
                    'RFC 3986 reserved character check' => sub { return ! IO::Iron::Common::contains_rfc_3986_res_chars(shift) },
                }}, # queue name.
			'message_timeout' => { type => SCALAR, optional => 1, },
			'message_expiration' => { type => SCALAR, optional => 1, },
			'type' => { type => SCALAR, optional => 1,
                regex => qr/^(?:multicast|unicast|pull)$/msx, ## no critic (Variables::ProhibitPunctuationVars)
            },
			'push' => { type => HASHREF, optional => 1, },
			'dead_letter' => { type => HASHREF, optional => 1, },
		}
	);
	$log->tracef('Entering create_queue(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

	my $connection = $self->{'connection'};
	my %item_body;
	$item_body{'message_timeout'} = $params{'message_timeout'} if ($params{'message_timeout'});
	$item_body{'message_expiration'} = $params{'message_expiration'} if ($params{'message_expiration'});
	$item_body{'type'} = $params{'type'} if ($params{'type'});
	$item_body{'push'} = $params{'push'} if ($params{'push'});
	$item_body{'dead_letter'} = $params{'dead_letter'} if ($params{'dead_letter'});
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_CREATE_QUEUE(),
			{
				'{Queue Name}' => $params{'name'},
				'body'         => \%item_body,
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;
	$log->tracef('Exiting create_queue: %s', undef);
	return;
}


sub get_queue_info {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
		}
	);
	$log->tracef('Entering get_queue_info(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_GET_QUEUE_INFO(),
			{ '{Queue Name}' => $params{'name'}, }
		);
	$self->{'last_http_status_code'} = $http_status_code;
	my $info = $response_message;
	# {"id":"51be[...]","name":"Log_Test_Queue","size":0,"total_messages":3,"project_id":"51bd[...]"}
	$log->debugf('Returned info about queue %s.', $params{'name'});
	$log->tracef('Exiting get_queue_info: %s', $info);
	return $info;
}


sub update_queue {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, callbacks => {
                    'RFC 3986 reserved character check' => sub { return ! IO::Iron::Common::contains_rfc_3986_res_chars(shift) },
                }}, # queue name.
			'message_timeout' => { type => SCALAR, optional => 1, },
			'message_expiration' => { type => SCALAR, optional => 1, },
			'push' => { type => HASHREF, optional => 1, },
			'dead_letter' => { type => HASHREF, optional => 1, },
		}
	);
	$log->tracef('Entering update_queue(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

	my $connection = $self->{'connection'};
	my %item_body;
	$item_body{'message_timeout'} = $params{'message_timeout'} if ($params{'message_timeout'});
	$item_body{'message_expiration'} = $params{'message_expiration'} if ($params{'message_expiration'});
	$item_body{'push'} = $params{'push'} if ($params{'push'});
	$item_body{'dead_letter'} = $params{'dead_letter'} if ($params{'dead_letter'});
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_UPDATE_QUEUE(),
			{
				'{Queue Name}' => $params{'name'},
				'body'         => \%item_body,
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;
	$log->tracef('Exiting update_queue: %s', undef);
	return;
}


sub delete_queue {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, callbacks => {
                    'RFC 3986 reserved character check' => sub { return ! IO::Iron::Common::contains_rfc_3986_res_chars(shift) },
                    'Is longer than zero characters' => sub { return length shift },
                }}, # queue name.
		}
	);
	$log->tracef('Entering delete_queue(%s)', \%params);

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_DELETE_QUEUE(),
			{
				'{Queue Name}' => $params{'name'},
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;
	$log->debugf('Deleted queue (queue name=%s).', $params{'name'});
	$log->tracef('Exiting delete_queue: %d', undef);
	return;
}


sub list_queues {
	my $self = shift;
	my %params = validate(
		@_, {
			'per_page' => { type => SCALAR, optional => 0,
                regex => qr/^[[:digit:]]{1,}$/msx, ## no critic (Variables::ProhibitPunctuationVars)
            },
            'previous' => { type => SCALAR, optional => 0, }, # Can be empty string.
            'prefix' => { type => SCALAR, optional => 1, },
		}
	);
	$log->tracef('Entering list_queues()');

	my @queues;
	my $connection = $self->{'connection'};
	my %query_params;
    $query_params{'{per_page}'} = $params{'per_page'} if $params{'per_page'};
    $query_params{'{previous}'} = $params{'previous'} if $params{'previous'};
    $query_params{'{prefix}'} = $params{'prefix'} if $params{'prefix'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronMQ::Api::IRONMQ_LIST_QUEUES(),
			{
				%query_params
			} );
	$self->{'last_http_status_code'} = $http_status_code;
	foreach my $queue_info (@{$response_message->{'queues'}}) {
		my $queue_name = $queue_info->{'name'};
		push @queues, $queue_name;
	}

	$log->tracef('Exiting list_queues: %s', \@queues);
	return @queues;
}


sub add_subscribers {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
			'subscribers' => { type => ARRAYREF, optional => 1 }, # array of subscriber hashes containing a required "url" field and an optional "headers" map for custom headers.
		}
	);
	$log->tracef('Entering add_subscribers(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

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

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


# TODO Inform bug in documentation: does not return the queue info, returns: "{msg => 'Updated'}".

sub delete_subscribers {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
			'subscribers' => { type => ARRAYREF, optional => 1 }, # array of subscriber hashes containing a required "url" field and an optional "headers" map for custom headers.
		}
	);
	$log->tracef('Entering delete_subscribers(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

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

	$log->debugf('Deleted subscribers (queue name=%s).', $params{'name'});
	$log->tracef('Exiting delete_subscribers: %d', 1);
	return 1;
}


sub add_alerts {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
			'alerts' => { type => ARRAYREF,
				callbacks => {
					'Assert item content' => sub {
						foreach my $alert (@{$_[0]}) {
							# TODO New function: Carp::Assert::More::assert_allowed(). Allowed keys in hash.
							assert_exists($alert, [ 'type', 'queue', 'trigger' ], 'Hash alert contains keys \'type\', \'queue\' and \'trigger\'.');
						}
						return 1;
					}
				}
			}, # An array of alert hashes containing required "type", "queue", "trigger", and optional "direction", "snooze" fields. Maximum number of alerts is 5.
		}
	);
	$log->tracef('Entering add_alerts(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

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

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


sub replace_alerts {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
			'alerts' => { type => ARRAYREF,
				callbacks => {
					'Assert item content' => sub {
						foreach my $alert (@{$_[0]}) {
							assert_exists($alert, [ 'type', 'queue', 'trigger' ], 'Hash alert contains keys \'type\', \'queue\' and \'trigger\'.');
						}
						return 1;
					}
				}
			}, # An array of alert hashes containing required "type", "queue", "trigger", and optional "direction", "snooze" fields. Maximum number of alerts is 5.
		}
	);
	$log->tracef('Entering replace_alerts(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');

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

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


sub delete_alerts {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # queue name.
			'alerts' => { type => ARRAYREF, optional => 1,
				callbacks => {
					'Either parameter \'alerts\' or \'id\'' => sub {
						return exists$_[1]->{'id'} ? 0 : 1;
					},
					'Assert item content' => sub {
						foreach my $alert (@{$_[0]}) {
							assert_exists($alert, [ 'id' ], 'Hash alert contains key \'id\'.');
						}
						return 1;
					}
				},
			}, # An array of alerts hashes containing "id" field.
			'id' => { type => SCALAR, optional => 1,
				callbacks => {
					'Either parameter \'alerts\' or \'id\'' => sub {
						return exists$_[1]->{'alerts'} ? 0 : 1;
					},
				},
			}, # alert id.
			# TODO New function: Params::Validate::validate(), mutually exclusive parameters.
		}
	);
	$log->tracef('Entering delete_alerts(%s)', \%params);
	assert_nonblank( $params{'name'}, 'Parameter \'name\' is a non blank string');
	assert($params{'alerts'} || $params{'id'}, 'Have either parameter \'alerts\' or \'id\'.');

	my $connection = $self->{'connection'};
	if($params{'alerts'}) {
		my %item_body;
		$item_body{'alerts'} = $params{'alerts'};
		my ($http_status_code, $response_message) = $connection->perform_iron_action(
				IO::Iron::IronMQ::Api::IRONMQ_REMOVE_ALERTS_FROM_A_QUEUE(),
				{
					'{Queue Name}' => $params{'name'},
					'body'         => \%item_body,
				}
			);
		$self->{'last_http_status_code'} = $http_status_code;
	}
	else {
		my ($http_status_code, $response_message) = $connection->perform_iron_action(
				IO::Iron::IronMQ::Api::IRONMQ_REMOVE_ALERTS_FROM_A_QUEUE_BY_ID(),
				{
					'{Queue Name}' => $params{'name'},
					'{Alert ID}'   => $params{'id'},
				}
			);
		$self->{'last_http_status_code'} = $http_status_code;
	}

	$log->debugf('Deleted alerts (queue name=%s).', $params{'name'});
	$log->tracef('Exiting delete_alerts: %d', 1);
	return 1;
}

1;

__END__