view release on metacpan or  search on metacpan

lib/App/cdnget/Worker.pm  view on Meta::CPAN

sub worker
{
	my $self = shift;
	my ($req) = @_;
	my ($in, $out, $err) = $req->GetHandles();
	my $env = $req->GetEnvironment();

	my $id = $env->{CDNGET_ID};
	$self->throw("Invalid ID") unless defined($id);
	$id = ($id =~ /^(.*)/)[0];
	$id =~ s/^\s+|\s+$//g;
	$self->throw("Invalid ID") unless $id =~ /^\w+$/i;

	my $origin = $env->{CDNGET_ORIGIN};
	$self->throw("Invalid origin") unless defined($origin);
	$origin = ($origin =~ /^(.*)/)[0];
	$origin =~ s/^\s+|\s+$//g;
	$origin = URI->new($origin);
	$self->throw("Invalid origin scheme") unless $origin->scheme =~ /^http|https$/i;
	$origin->path(substr($origin->path, 0, length($origin->path)-1)) while $origin->path and substr($origin->path, -1) eq "/";

	my $uri = $env->{CDNGET_URI};
	$self->throw("Invalid URI") unless defined($uri);
	$uri = ($uri =~ /^(.*)/)[0];
	$uri =~ s/^\s+|\s+$//g;
	$uri = "/$uri" unless $uri and substr($uri, 0, 1) eq "/";

	my $hook = $env->{CDNGET_HOOK};
	$hook = "" unless defined($hook);
	$hook = ($hook =~ /^(.*)/)[0];
	$hook =~ s/^\s+|\s+$//g;

	my $url = $origin->scheme."://".$origin->host_port.$origin->path.$uri;
	my $digest = Digest::MD5::md5_hex("$url $hook");
	my $uid = "$id/$digest";
	my $path = "$cachePath/$id";
	mkdir($path);
	my @dirs = $digest =~ /(..)(.)$/;
	my $file = $digest;
	for (reverse @dirs)
	{
		$path .= "/$_";
		mkdir($path);
	}
	$self->throw("Cache directory not exists") unless -d $path;
	$path .= "/$file";

	my $fh;
	my $downloader;
	do
	{
		lock(%App::cdnget::Downloader::uids);
		$fh = FileHandle->new($path, "<");
		unless ($fh)
		{
			return unless App::cdnget::Downloader->new($uid, $path, $url, $hook);
			$fh = FileHandle->new($path, "<") or $self->throw($!);
		}
		$downloader = $App::cdnget::Downloader::uids{$uid};
	};
	$fh->binmode(":bytes") or $self->throw($!);

	do
	{
		local ($/, $\) = ("\r\n")x2;
		my $line;
		my $buf;
		my $empty = 1;
		while (not $self->terminating)
		{
			threads->yield();
			my $downloaderTerminated = ! $downloader || $downloader->terminated;
			$line = $fh->getline;
			unless (defined($line))
			{
				$self->throw($!) if $fh->error;
				return if $downloaderTerminated;
				my $pos = $fh->tell;
				usleep(1*1000);
				$fh->seek($pos, 0) or $self->throw($!);
				next;
			}
			chomp $line;
			unless ($line =~ /^(Client\-)/i)
			{
				if (not $out->print("$line\r\n"))
				{
					not $! or $!{EPIPE} or $!{ECONNRESET} or $!{EPROTOTYPE} or $self->throw($!);
					return;
				}
				$empty = 0;
			}
			last unless $line;
		}
		while (not $self->terminating)
		{
			threads->yield();
			my $downloaderTerminated = ! $downloader || $downloader->terminated;
			my $len = $fh->read($buf, $App::cdnget::CHUNK_SIZE);
			$self->throw($!) unless defined($len);
			if ($len == 0)
			{
				return if $downloaderTerminated;
				my $pos = $fh->tell;
				usleep(1*1000);
				$fh->seek($pos, 0) or $self->throw($!);
				next;
			}
			if (not $out->write($buf, $len))
			{
				not $! or $!{EPIPE} or $!{ECONNRESET} or $!{EPROTOTYPE} or $self->throw($!);
				return;
			}
			$empty = 0;
		}
		if ($empty)
		{
			if (not $out->print("Status: 404\r\n"))
			{
				not $! or $!{EPIPE} or $!{ECONNRESET} or $!{EPROTOTYPE} or $self->throw($!);
				return;