File Coverage

blib/lib/POE/Component/Client/UserAgent.pm

Criterion	Covered	Total	%
statement	7	9	77.7
branch			n/a
condition			n/a
subroutine	3	3	100.0
pod			n/a
total	10	12	83.3

package POE::Component::Client::UserAgent;
use strict;
use POE;
use LWP::Parallel;

@POE::Component::Client::UserAgent::ISA = 'LWP::Parallel::UserAgent';
$POE::Component::Client::UserAgent::VERSION = '0.08';

my $debuglevel = 0;

sub new
{
	my $class = @_ ? shift : 'POE::Component::Client::UserAgent';
	$class -> spawn (@_);
}

sub spawn
{
	my $class = @_ ? shift : 'POE::Component::Client::UserAgent';
	$class = ref $class || $class;
	my $object = $class -> SUPER::new;
	bless $object, $class;
	$object -> nonblock (0);
	my $argref = @_ & 1 ? pop @_ : { };
	my %args = (@_, %$argref);
	$args{alias} ||= 'useragent';
	LWP::Debug::trace ("Alias=$args{alias}\n\t$object");
	POE::Session -> create (
		object_states => [
			$object => {
				_start => '_pococ_ua_start',
				_stop => '_pococ_ua_stop',
				sigint => '_pococ_ua_sig_int',
				write => '_pococ_ua_write',
				read => '_pococ_ua_read',
				error => '_pococ_ua_error',
				timeout => '_pococ_ua_timeout',
				request => '_pococ_ua_request',
				shutdown => '_pococ_ua_shutdown'
			}
		],
		args => \%args
	);
	my $entry = LWP::Parallel::UserAgent::Entry -> new;
	$$entry{_permitted}{$_} = undef for qw(postback alarm_id alarm_time);
	return $object;
}

sub _pococ_ua_start
{
	my ($object, $kernel, $heap, $args) = @_[OBJECT, KERNEL, HEAP, ARG0];
	my $alias = $$args{alias};
	LWP::Debug::trace ("Alias=$alias\n\t$object\n\t$kernel");
	warn "Session '$alias' started\n" if $debuglevel >= 3;
	$kernel -> alias_set ($alias);
	$$heap{alias} = $alias;
	$object -> $_ ($$args{$_}) for grep exists ($$args{$_}),
		qw(agent from timeout redirect duplicates in_order remember_failures
		env_proxy proxy cookie_jar parse_head max_size max_hosts max_req delay);
	$kernel->sig(INT => 'sigint');
	$kernel->sig(BREAK => 'sigint');
}

sub _pococ_ua_stop
{
	my ($object, $heap) = @_[OBJECT, HEAP];
	LWP::Debug::trace ("Alias=$$heap{alias}\n\t$object");
	warn "Session '$$heap{alias}' stopped\n" if $debuglevel >= 3;
}

sub DESTROY
{
	my ($object) = @_;
	LWP::Debug::trace ("$object");
	warn "$object destroyed\n" if $debuglevel >= 3;
}

sub _pococ_ua_sig_int
{
	my ($object, $signal) = @_[OBJECT, ARG0];
	LWP::Debug::trace ("Signal=$signal\n\t$object");
	warn "Signal '$signal' arrived\n" if $debuglevel >= 3;
	$object -> _pococ_ua_cleanup();
	return 0;
}

sub _pococ_ua_shutdown
{
	my ($object, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
	LWP::Debug::trace ("Alias=$$heap{alias}\n\t$object\n\t$kernel");
	warn "Removing '$$heap{alias}' alias\n" if $debuglevel >= 3;
	$kernel -> alias_remove ($$heap{alias});
}

sub _pococ_ua_cleanup
{
	my ($object) = @_;
	LWP::Debug::trace ("$object");
	warn "Cleaning up\n" if $debuglevel >= 3;
	$object -> _remove_all_sockets;
	$poe_kernel -> alarm ('timeout');
}

sub _pococ_ua_request
{
	my ($object, @args) = @_[OBJECT, ARG0 .. $#_];
	my $argref = @args & 1 ? pop @args : { };
	my %args = (@args, %$argref);
	my ($request, $filename, $callback, $chunksize, $redirect) =
		@args{qw(request filename callback chunksize redirect)};
	LWP::Debug::trace ("$object\n\t$request");
	warn 'Request for ' . $request -> url -> as_string . "\n" if $debuglevel >= 3;
	my $register = $object -> register ($request,
		$filename || $callback, $chunksize, $redirect);
	$$object{entries_by_requests}{$request} -> postback ($args{response});
	$object -> _make_connections;
}

sub _pococ_ua_set_timeout
{
	my ($object, $entry) = @_;
	my $timeout = $object -> timeout;
	return unless defined $timeout;
	my $alarm_id = $poe_kernel -> delay_set (timeout => $timeout, $entry);
	LWP::Debug::trace ("$object\n\t$entry\n\tTimeout: $timeout\n\tAlarm ID: "
		. (defined $alarm_id ? $alarm_id : '[undef]'));
	$entry -> alarm_id ($alarm_id);
	$entry -> alarm_time (defined $alarm_id ? time() + $timeout : undef);
}

sub _pococ_ua_adjust_timeout
{
	my ($object, $entry) = @_;
	$object -> _pococ_ua_remove_timeout ($entry);
	$object -> _pococ_ua_set_timeout ($entry);
}

# alarm_adjust causes problems in POE 0.1402
#sub _pococ_ua_adjust_timeout
#{
#	my ($object, $entry) = @_;
#	my $timeout = $object -> timeout;
#	return unless defined $timeout;
#	my $alarm_id = $entry -> alarm_id;
#	return unless defined $alarm_id;	# Couldn't set alarm? Should never happen.
#	my $previous_alarm_time = $entry -> alarm_time;
#	my $new_alarm_time = time() + $timeout;
#	return if $new_alarm_time == $previous_alarm_time;
#	LWP::Debug::trace ("$object\n\t$entry\n\tTimeout: $timeout\n"
#		. "\tAlarm ID: $alarm_id\n\tPrevious Alarm Time: $previous_alarm_time\n"
#		. "\tNew Alarm Time: $new_alarm_time");
#	$poe_kernel -> alarm_adjust ($alarm_id, $new_alarm_time - $previous_alarm_time);
#	$entry -> alarm_id ($alarm_id);
#	$entry -> alarm_time ($new_alarm_time);
#}

sub _pococ_ua_remove_timeout
{
	my ($object, $entry) = @_;
	my $alarm_id = $entry -> alarm_id;
	return unless defined $alarm_id;
	LWP::Debug::trace ("$object\n\t$entry\n\tAlarm ID: $alarm_id");
	$poe_kernel -> alarm_remove ($alarm_id);
	$entry -> alarm_id (undef);
	$entry -> alarm_time (undef);
}

sub _connect
{
	my ($object, $entry) = @_;
	LWP::Debug::trace ("$object\n\t$entry\n\t" . $entry -> request -> url);
	warn 'Connecting ' . $entry -> request -> url -> as_string . "\n"
		if $debuglevel >= 3;
	my $result = $object -> SUPER::_connect ($entry);
	return $result if defined $result;
	$object -> _pococ_ua_set_timeout ($entry);
	return undef;
}

sub _add_out_socket
{
	my ($object, $socket) = @_;
	LWP::Debug::trace ("$object\n\t$socket");
	$poe_kernel -> select_write ($socket => 'write');
	$poe_kernel -> select_expedite ($socket => 'error')
		unless -f $socket;
}

sub _add_in_socket
{
	my ($object, $socket) = @_;
	LWP::Debug::trace ("$object\n\t$socket");
	$poe_kernel -> select_read ($socket => 'read');
	$poe_kernel -> select_expedite ($socket => 'error')
		unless -f $socket;
}

sub _remove_out_socket
{
	my ($object, $socket) = @_;
	LWP::Debug::trace ("$object\n\t$socket");
	$poe_kernel -> select_write ($socket);
	$poe_kernel -> select_expedite ($socket)
		unless -f $socket;
}

sub _remove_in_socket
{
	my ($object, $socket) = @_;
	LWP::Debug::trace ("$object\n\t$socket");
	$poe_kernel -> select_read ($socket);
	$poe_kernel -> select_expedite ($socket)
		unless -f $socket;
}

sub _remove_all_sockets
{
	my ($object) = @_;
	LWP::Debug::trace ("$object");
	my ($socket, $entry);
	$object -> _remove_entry_sockets ($entry)
		while ($socket, $entry) = each %{$$object{entries_by_sockets}};
	$object -> initialize;
}

sub _remove_entry_sockets
{
	my ($object, $entry) = @_;
	LWP::Debug::trace ("$object\n\t$entry");
	my $socket = $entry -> cmd_socket;
	if ( defined $socket )
	{
		$object -> _remove_out_socket ($socket);
		$entry -> cmd_socket (undef);
	}
	$socket = $entry -> listen_socket;
	if ( defined $socket )
	{
		$object -> _remove_in_socket ($socket);
		$entry -> listen_socket (undef);
	}
}

sub _pococ_ua_write
{
	my ($object, $socket) = @_[OBJECT, ARG0];
	my $entry = $$object{entries_by_sockets}{$socket};
	LWP::Debug::trace ("$object\n\t$socket\n\t$entry\n\t"
		. $entry -> request -> url);
	warn 'Writing ' . $entry -> request -> url -> as_string . "\n"
		if $debuglevel >= 3;
	$object -> _pococ_ua_adjust_timeout ($entry);
	$object -> _perform_write ($socket);
}

sub _pococ_ua_read
{
	my ($object, $socket) = @_[OBJECT, ARG0];
	my $entry = $$object{entries_by_sockets}{$socket};
	LWP::Debug::trace ("$object\n\t$socket\n\t$entry\n\t"
		. $entry -> request -> url);
	warn 'Reading ' . $entry -> request -> url -> as_string . "\n"
		if $debuglevel >= 3;
	$object -> _pococ_ua_adjust_timeout ($entry);
	$object -> _perform_read ($socket);
}

sub _pococ_ua_error
{
	my ($object, $kernel, $socket) = @_[OBJECT, KERNEL, ARG0];
	my $entry = $$object{entries_by_sockets}{$socket};
	my $request = $entry -> request;
	LWP::Debug::trace ("$object\n\t$kernel\n\t$socket\n\t$entry\n\t$request\n\t"
		. $request -> url);
	warn 'Error on ' . $request -> url -> as_string . "\n"
		if $debuglevel >= 3;
	my $response = HTTP::Response -> new (&HTTP::Status::RC_INTERNAL_SERVER_ERROR,
		'Connection was reset');
	$response -> request ($request);
	$entry -> response ($response);
	$object -> on_failure ($request, $response, $entry);
	LWP::Debug::trace ('Error while processing request ' . $request -> url);
	$object -> _remove_entry_sockets ($entry);
	$object -> _remove_current_connection ($entry);
}

sub _pococ_ua_timeout
{
	my ($object, $kernel, $entry) = @_[OBJECT, KERNEL, ARG0];
	$entry -> alarm_id (undef);
	$entry -> alarm_time (undef);
	my $request = $entry -> request;
	LWP::Debug::trace ("$object\n\t$kernel\n\t$entry\n\t$request\n\t"
		. $request -> url);
	warn 'Timeout on ' . $request -> url -> as_string . "\n"
		if $debuglevel >= 3;
	my $response = HTTP::Response -> new (&HTTP::Status::RC_REQUEST_TIMEOUT,
		'Request timeout (I/O inactivity)');
	$response -> request ($request);
	$entry -> response ($response);
	$object -> on_failure ($request, $response, $entry);
	LWP::Debug::trace ('Request timeout ' . $request -> url -> as_string);
	$object -> _remove_entry_sockets ($entry);
	$object -> _remove_current_connection ($entry);
}

sub _pococ_ua_postback
{
	my ($object, $request, $response, $entry) = @_;
	$object -> _pococ_ua_remove_timeout ($entry);
	$entry -> postback -> ($request, $response, $entry);
	if ( $entry -> redirect_ok )
	{
		# We need to skip cleanup if the response is a redirect.
		# See LWP::Parallel::UserAgent::handle_response for details.
		my $code = $response -> code;
		if ( $code == HTTP::Status::RC_MOVED_PERMANENTLY
				or $code == HTTP::Status::RC_MOVED_TEMPORARILY
				or $code == HTTP::Status::RC_FOUND
				or $code == HTTP::Status::RC_SEE_OTHER
				or $code == HTTP::Status::RC_TEMPORARY_REDIRECT
		) {
			$code = $response -> header ('Client-Warning');
			return unless defined ($code) and $code eq 'Redirect loop detected';
		}
	}
	$object -> discard_entry ($entry);
	# if the entry doesn't get discarded for whatever reason, the postback
	# may create a circular reference, depending on what the user passed
	# to Session::postback(), so we'd better break it here.
	$entry -> postback (undef);
}

sub on_return
{
	my ($object, $request, $response, $entry) = @_;
	LWP::Debug::trace ("$object\n\t$request\n\t$response\n\t$entry\n\t" .
		join "\n\t", $request -> url -> as_string,
		$response -> code, $response -> message);
	warn 'Response returned ' . $request -> url -> as_string . "\n"
		if $debuglevel >= 3;
	$object -> _pococ_ua_postback ($request, $response, $entry);
	return 0;
}

sub on_failure
{
	my ($object, $request, $response, $entry) = @_;
	LWP::Debug::trace ("$object\n\t$request\n\t$response\n\t$entry\n\t" .
		join "\n\t", $request -> url -> as_string,
		$response -> code, $response -> message);
	warn 'Request failed ' . $request -> url -> as_string . "\n"
		if $debuglevel >= 3;
	$object -> _pococ_ua_postback ($request, $response, $entry);
	return 0;
}

sub debug
{
	my $level = shift;
	$level = shift if ref $level;
	return unless defined $level;
	$debuglevel = $level;
	LWP::Debug::level '+debug' if $debuglevel >= 5;
	LWP::Debug::level '+trace' if $debuglevel >= 7;
	LWP::Debug::level '+conns' if $debuglevel >= 9;
	my $filename = shift;
	return unless $debuglevel > 0 and defined $filename;
	close STDERR;
	open STDERR, ">$filename";
}

no warnings 'redefine';

sub LWP::Debug::_log
{
	my $msg = shift;
	$msg .= "\n" unless $msg =~ /\n$/;
	my $sub = (caller (2)) [3];
	warn "$sub\n\t$msg";
}

1;

__END__