File Coverage

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

Criterion	Covered	Total	%
statement	60	276	21.7
branch	3	44	6.8
condition			n/a
subroutine	14	29	48.2
pod	15	15	100.0
total	92	364	25.2

package IO::Iron::IronWorker::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: IronWorker (Online Worker Platform) Client.

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



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

use IO::Iron::IronWorker::Api ();
use IO::Iron::Common ();
require IO::Iron::Connection;
require IO::Iron::IronWorker::Task;

# CONSTANTS for this package

# DEFAULTS
use Const::Fast;

# Service specific!
const my $DEFAULT_API_VERSION => '2';
const my $DEFAULT_HOST => 'worker-aws-us-east-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 = (
			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;
	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('IronWorker Client 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;
}

###############################################
######## FUNCTIONS: CODE PACKAGES #############
###############################################


sub list_code_packages {
	my ($self) = @_;
	$log->tracef('Entering list_code_packages()');

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_LIST_CODE_PACKAGES(), { } );
	$self->{'last_http_status_code'} = $http_status_code;
	my @codes;
	foreach (@{$response_message}) {
		push @codes, $_;
	}
	$log->debugf('Returning %d code packages.', scalar @codes);
	$log->tracef('Exiting list_code_packages: %s', \@codes);
	return @codes;
}


sub update_code_package {
	my $self = shift;
	my %params = validate(
		@_, {
			'name' => { type => SCALAR, }, # Code package name.
			'file' => { type => SCALAR, optional => 1, # The zip archive as a string buffer.
				depends => [ 'file_name', 'runtime' ], },
			'file_name' => { type => SCALAR, optional => 1 },  # Name of the zip file, not
			'runtime' => { type => SCALAR, optional => 1, }, # The runtime type, e.g. sh, perl, ruby.
			'config' => { type => SCALAR, optional => 1, },
			'max_concurrency' => { type => SCALAR, optional => 1, },
			'retries' => { type => SCALAR, optional => 1, },
			'retries_delay' => { type => SCALAR, optional => 1, },
		}
	);
	$log->tracef('Entering update_code_package(%s)', \%params);
	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_UPLOAD_OR_UPDATE_A_CODE_PACKAGE(),
			{
				'body'         => \%params,
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;
	my $id = $response_message->{'id'};
	$log->tracef('Exiting update_code_package: %s', $id);
	return $id;
}


sub get_info_about_code_package {
	my $self = shift;
	my %params = validate(
		@_, {
			'id' => { type => SCALAR, }, # Code package id.
		}
	);
	$log->tracef('Entering get_info_about_code_package(%s)', \%params);
	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_GET_INFO_ABOUT_A_CODE_PACKAGE(),
			{ '{Code ID}' => $params{'id'}, }
		);
	$self->{'last_http_status_code'} = $http_status_code;
	my $info = $response_message;
	$log->tracef('Exiting get_info_about_code_package: %s', $info);
	return $info;
}


sub delete_code_package {
	my $self = shift;
	my %params = validate(
		@_, {
			'id' => { type => SCALAR, }, # Code package id.
		}
	);
	$log->tracef('Entering delete_code_package(%s)', $params{'id'});
	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_DELETE_A_CODE_PACKAGE(),
			{
				'{Code ID}' => $params{'id'},
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;
	$log->tracef('Exiting delete_code_package: %d', 1);
	return 1;
}


sub download_code_package {
	my $self = shift;
	my %params = validate(
		@_, {
			'id' => { type => SCALAR, }, # Code package id.
			'revision' => { type => SCALAR, optional => 1, }, # Code package revision.
		}
	);
	$log->tracef('Entering download_code_package(%s)', \%params);
	my $connection = $self->{'connection'};
	my %query_params;
	$query_params{'{revision}'} = $params{'revision'} if $params{'revision'}; ## no critic (ControlStructures::ProhibitPostfixControls)
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_DOWNLOAD_A_CODE_PACKAGE(),
			{
				'{Code ID}' => $params{'id'},
				%query_params,
			}
		);
	my $code_package = $response_message->{'file'};
	my $file_name = $response_message->{'file_name'};
	$self->{'last_http_status_code'} = $http_status_code;

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


sub list_code_package_revisions {
	my $self = shift;
	my %params = validate(
		@_, {
			'id' => { type => SCALAR, }, # Code package id.
		}
	);
	$log->tracef('Entering list_code_package_revisions(%s)', $params{'id'});

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_LIST_CODE_PACKAGE_REVISIONS(),
			{ '{Code ID}' => $params{'id'}, }
		);
	my @revisions;
	foreach (@{$response_message}) {
		push @revisions, $_;
	}
	$log->debugf('Returning %d code packages.', scalar @revisions);
	$log->tracef('Exiting list_code_package_revisions: %s', \@revisions);
	return @revisions;
}


###############################################
######## FUNCTIONS: TASK ######################
###############################################


sub create_task {
	my $self = shift;
	my %params = validate_with(
		'params' => \@_,
		'normalize_keys' => sub { return lc shift },
		'spec' => {
		'code_name' => { type => SCALAR, }, # Code package name.
		'payload' => { type => SCALAR, },  # Payload
		},
		'allow_extra' => 1,
    );
	$log->tracef('Entering create_task(%s)', \%params);

	my $connection = $self->{'connection'};

	my $task = IO::Iron::IronWorker::Task->new({
		'ironworker_client' => $self, # Pass a reference to the parent object.
		'connection' => $connection,
		%params,
	});

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


sub tasks {
	my $self = shift;
	my %params = validate(
		@_, {
			'code_name' => { type => SCALAR, }, # Code package name.
			'queued' => { type => SCALAR, 'optional' => 1, },
			'running' => { type => SCALAR, 'optional' => 1, },
			'complete' => { type => SCALAR, 'optional' => 1, },
			'error' => { type => SCALAR, 'optional' => 1, },
			'cancelled' => { type => SCALAR, 'optional' => 1, },
			'killed' => { type => SCALAR, 'optional' => 1, },
			'from_time' => { type => SCALAR, 'optional' => 1, },
			'to_time' => { type => SCALAR, 'optional' => 1, },
		}
	);
	$log->tracef('Entering tasks(%s, %s)', \%params);

	my $code_name = $params{'code_name'};
	delete $params{'code_name'};
	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_LIST_TASKS(), {
				'{code_name}' => $code_name,
				%params,
			} );
	$self->{'last_http_status_code'} = $http_status_code;
	my @tasks;
	foreach (@{$response_message}) {
		$log->debugf('task info:%s', $_);
		push @tasks, $self->create_task(
			'code_name' => $_->{'code_name'},
			'payload' => $_->{'payload'} ? $_->{'payload'} : q{},
			%{$_},
		);
	}
	$log->debugf('Returning %d tasks.', scalar @tasks);
	$log->tracef('Exiting tasks: %s', \@tasks);
	return @tasks;
}


sub queue {
	my $self = shift;
	my %validate_params = (
		'tasks' => { type => OBJECT | ARRAYREF, }, # ref to task.
		);
	my %params = validate(
		@_, {
			%validate_params
		}
	);
	lock_keys(%params, keys %validate_params);
	$log->tracef('Entering queue(%s)', \%params);

	my $connection = $self->{'connection'};
	my @message_tasks;
	if(ref $params{'tasks'} eq 'IO::Iron::IronWorker::Task') {
		$params{'tasks'} = [ $params{'tasks'} ];
	}
	foreach my $task (@{$params{'tasks'}}) {
		assert_isa( $task, 'IO::Iron::IronWorker::Task', 'task is IO::Iron::IronWorker::Task.');
		my %task_body;
		$task_body{'code_name'} = $task->{'code_name'};
		$task_body{'payload'} = $task->{'payload'};
		$task_body{'priority'} = $task->{'priority'} if defined $task->{'priority'};
		$task_body{'timeout'} = $task->{'timeout'} if defined $task->{'timeout'};
		$task_body{'delay'} = $task->{'delay'} if defined $task->{'delay'};
		$task_body{'name'} = $task->{'name'} if defined $task->{'name'};
		push @message_tasks, \%task_body;
	}

	my %message_body = ('tasks' => \@message_tasks);
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_QUEUE_A_TASK(),
			{
				'body'         => \%message_body,
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;

	my @ids;
	my @ret_tasks = ( @{ $response_message->{'tasks'} } ); # tasks.
	foreach my $task (@{$params{'tasks'}}) {
		my $task_info = shift @ret_tasks; # We are using the same task objects.
		push @ids, $task_info->{'id'};
		$task->id( $task_info->{'id'} );
	}
	#assert_is($response_message->{'msg'}, 'Queued up'); # Could be dangerous!
	$log->debugf( 'Queued IronWorker Task(s) (task id(s)=%s).', ( join q{,}, @ids ) );
	if (wantarray) {
		$log->tracef( 'Exiting queue: %s', ( join q{:}, @ids ) );
		return @ids;
	}
	else {
		if ( scalar @{$params{'tasks'}} == 1 ) {
			$log->tracef( 'Exiting queue: %s', $ids[0] );
			return $ids[0];
		}
		else {
			$log->tracef( 'Exiting queue: %s', scalar @ids );
			return scalar @ids;
		}
	}
}


sub get_info_about_task {
	my $self = shift;
	my %params = validate(
		@_, {
			'id' => { type => SCALAR, }, # task id.
		}
	);
	$log->tracef('Entering get_info_about_task(%s)', \%params);

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_GET_INFO_ABOUT_A_TASK(),
			{ '{Task ID}' => $params{'id'}, }
		);
	$self->{'last_http_status_code'} = $http_status_code;
	my $info = $response_message;
	$log->tracef('Exiting get_info_about_task: %s', $info);
	return $info;
}

###############################################
######## FUNCTIONS: SCHEDULED TASK ############
###############################################


sub scheduled_tasks {
	my ($self) = @_;
	$log->tracef('Entering scheduled_tasks()');

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_LIST_SCHEDULED_TASKS(), { } );
	$self->{'last_http_status_code'} = $http_status_code;
	my @tasks;
	foreach (@{$response_message}) {
		$log->debugf('task info:%s', $_);
		push @tasks, $self->create_task(
			'code_name' => $_->{'code_name'},
			'payload' => $_->{'payload'} ? $_->{'payload'} : q{},
			%{$_},
		);
	}
	$log->debugf('Returning %d tasks.', scalar @tasks);
	$log->tracef('Exiting scheduled_tasks: %s', \@tasks);
	return @tasks;
}


sub schedule {
	my $self = shift;
	my %validate_params = (
		'tasks' => { type => OBJECT | ARRAYREF, }, # ref to task.
		);
	my %params = validate(
		@_, {
			%validate_params
		}
	);
	lock_keys(%params, keys %validate_params);
	$log->tracef('Entering schedule(%s)', \%params);

	my $connection = $self->{'connection'};
	my @message_tasks;
	if(ref $params{'tasks'} eq 'IO::Iron::IronWorker::Task') {
		$log->debugf('The parameter is a single object.');
		$params{'tasks'} = [ $params{'tasks'} ];
	}
	foreach my $task (@{$params{'tasks'}}) {
		assert_isa( $task, 'IO::Iron::IronWorker::Task', 'task is IO::Iron::IronWorker::Task.');
		my %task_body;
		$task_body{'code_name'} = $task->{'code_name'};
		$task_body{'payload'} = $task->{'payload'};
		$task_body{'run_every'} = $task->{'run_every'} if defined $task->{'run_every'};
		$task_body{'end_at'} = $task->{'end_at'} if defined $task->{'end_at'};
		$task_body{'run_times'} = $task->{'run_times'} if defined $task->{'run_times'};
		$task_body{'priority'} = $task->{'priority'} if defined $task->{'priority'};
		$task_body{'start_at'} = $task->{'start_at'} if defined $task->{'start_at'};
		$task_body{'name'} = $task->{'name'} if defined $task->{'name'}; # Hm... documents do not mention but example does...
		push @message_tasks, \%task_body;
	}

	my %message_body = ('schedules' => \@message_tasks);
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_SCHEDULE_A_TASK(),
			{
				'body'         => \%message_body,
			}
		);
	$self->{'last_http_status_code'} = $http_status_code;

	my ( @ids, $msg ); ## no critic (Variables::ProhibitUnusedVariables)
	my @ret_tasks = ( @{ $response_message->{'schedules'} } );    # scheduled tasks.
	foreach my $task (@{$params{'tasks'}}) {
		my $task_info = shift @ret_tasks;
		push @ids, $task_info->{'id'};
		$task->id( $task_info->{'id'} );
	}
        #assert_is($response_message->{'msg'}, 'Scheduled'); # Could be dangerous!
        ## It was dangerous! Failing discovered 2014-08-10, response message has dropped the field "msg"!
	$log->debugf( 'Scheduled IronWorker Task(s) (task id(s)=%s).', ( join q{,}, @ids ) );
	if (wantarray) {
		$log->tracef( 'Exiting schedule: %s', ( join q{:}, @ids ) );
		return @ids;
	}
	else {
		if ( scalar @{$params{'tasks'}} == 1 ) {
			$log->tracef( 'Exiting schedule: %s', $ids[0] );
			return $ids[0];
		}
		else {
			$log->tracef( 'Exiting schedule: %s', scalar @ids );
			return scalar @ids;
		}
	}
}


sub get_info_about_scheduled_task {
	my $self = shift;
	my %params = validate(
		@_, {
			'id' => { type => SCALAR, }, # task id.
		}
	);
	$log->tracef('Entering get_info_about_scheduled_task(%s)', \%params);

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_GET_INFO_ABOUT_A_SCHEDULED_TASK(),
			{ '{Schedule ID}' => $params{'id'}, }
		);
	$self->{'last_http_status_code'} = $http_status_code;
	my $info = $response_message;
	$log->tracef('Exiting get_info_about_scheduled_task: %s', $info);
	return $info;
}

###############################################
########### FUNCTIONS: STACKS #################
###############################################


sub list_available_stacks {
	my ($self) = @_;
	$log->tracef('Entering list_available_stacks()');

	my $connection = $self->{'connection'};
	my ($http_status_code, $response_message) = $connection->perform_iron_action(
			IO::Iron::IronWorker::Api::IRONWORKER_LIST_OF_AVAILABLE_STACKS(), { } );
	$self->{'last_http_status_code'} = $http_status_code;
	my @stacks;
	foreach (@{$response_message}) {
		push @stacks, $_;
	}
	$log->debugf('Returning %d stacks.', scalar @stacks);
	$log->tracef('Exiting list_available_stacks(): %s', \@stacks);
	return @stacks;
}

1;

__END__