view release on metacpan or  search on metacpan

Server/t/lib/Apache/AppCluster/Client.pm  view on Meta::CPAN

	}
}

sub get_request_status
{
	my $self = shift @_;
	my $index = shift @_;

	if(exists $self->{_requests}->{$index})
	{
		return $self->{_requests}->{$index}->{status}; 
	} else
	{
		return undef; #no such request
	}
}

sub send_requests
{
	my $self = shift @_;
	my $timeout = shift @_; #in seconds - can be a float
	
	my $stime = Time::HiRes::time();

	foreach my $index (keys %{$self->{_requests}})
	{
		warn "Connecting to: " . $self->{_requests}->{$index}->{server} . ":" . $self->{_requests}->{$index}->{port} if($VERBOSE);
		if(! ($self->{_requests}->{$index}->{sock} = new IO::Socket::INET (
			PeerAddr => $self->{_requests}->{$index}->{server},
			PeerPort => $self->{_requests}->{$index}->{port},
			Proto => 'tcp',)) 
		  ) 
		{
			$self->{_requests}->{$index}->{status} = REQ_CONNECT_FAIL;
		} else
		{
			fcntl($self->{_requests}->{$index}->{sock}, F_SETFL(), O_NONBLOCK()); #now you're a non-blocker
		}
	}

	my $connected_sockets = 0;
	
	foreach my $index (keys %{$self->{_requests}})
	{
	
		if( $self->{_requests}->{$index}->{status} == REQ_INCOMPLETE ) #those that didn't fail to connect
		{
			$connected_sockets++;
			my $href = {
				method => $self->{_requests}->{$index}->{method},
				params => $self->{_requests}->{$index}->{params},
				};

			my $data = freeze($href);
			my $digest = md5_hex($data);
			my $send_data = '<frozen>' . $digest . $data . '</frozen>';
			my $content_length = length($send_data);
			my $uri = $self->{_requests}->{$index}->{uri};
			my $host = $self->{_requests}->{$index}->{server};

#I'm guessing that octet-stream is the correct mime type for this sort of thing.
			print {$self->{_requests}->{$index}->{sock}} <<"EOF";
POST $uri HTTP/1.0
Accept: application/octet-stream
Accept: */*
Host: $host
Connection: close
User-Agent: Apache::AppCluster::Client v0.1
Content-Length: $content_length
Content-Type: application/octet-stream;

$send_data
EOF

			$self->{_requests}->{$index}->{sock}->flush();
		}
	}





	my $cutoff_time = Time::HiRes::time() + $timeout;

	my $sockets_pending = $connected_sockets;

	while((Time::HiRes::time() < $cutoff_time) && $sockets_pending)
	{
		my $sockets_finished = 0;
		foreach my $index (keys %{$self->{_requests}})
		{
			if($self->{_requests}->{$index}->{status} == REQ_INCOMPLETE) 
			{
				my $buf;
				my $bytes_read = sysread($self->{_requests}->{$index}->{sock}, $buf, 1024);
				if(defined($bytes_read) )
				{
					if($bytes_read == 0)
					{
						close($self->{_requests}->{$index}->{sock});
						$self->{_requests}->{$index}->{status} = REQ_SUCCESS; #finished
						$sockets_pending--;
					} else
					{
						$self->{_requests}->{$index}->{data} .= $buf;
					}
				} else #no data to read yet
				{
					if($! == EAGAIN()) #socket would have blocked
					{
						#keep going until there is more data on the socket
					} else
					{
						$self->{_requests}->{$index}->{status} = REQ_SUCCESS;
						$sockets_pending--;
					}
				}
			} 
		}
		
	}