File Coverage

blib/lib/POE/Component/SmokeBox/Recent.pm

Criterion	Covered	Total	%
statement	133	156	85.2
branch	46	64	71.8
condition	2	6	33.3
subroutine	19	23	82.6
pod	1	1	100.0
total	201	250	80.4

package POE::Component::SmokeBox::Recent;
$POE::Component::SmokeBox::Recent::VERSION = '1.50';
#ABSTRACT: A POE component to retrieve recent CPAN uploads.

use strict;
use warnings;
use Carp;
use POE qw(Component::SmokeBox::Recent::HTTP Component::SmokeBox::Recent::FTP Wheel::Run);
use URI;
use HTTP::Request;
use File::Spec;

sub recent {
  my $package = shift;
  my %opts = @_;
  $opts{lc $_} = delete $opts{$_} for keys %opts;
  croak "$package requires a 'url' argument\n" unless $opts{url};
  croak "$package requires an 'event' argument\n" unless $opts{event};
  $opts{rss} = 0 unless $opts{rss};
  my $options = delete $opts{options};
  my $self = bless \%opts, $package;
  $self->{recent} = [];
  $self->{uri} = URI->new( $self->{url} );
  croak "url provided is of an unsupported scheme\n"
	unless $self->{uri}->scheme and $self->{uri}->scheme =~ /^(ht|f)tp|file$/;
  $self->{session_id} = POE::Session->create(
	object_states => [
	   $self => [ qw(_start _process_http _process_ftp _process_file _recent _sig_child _epoch _epoch_fail) ],
	   $self => {
		      http_sockerr  => '_get_connect_error',
		      http_timeout  => '_get_connect_error',
		      http_response => '_http_response',
		      ftp_sockerr   => '_get_connect_error',
		      ftp_error     => '_get_error',
		      ftp_data      => '_get_data',
		      ftp_done      => '_get_done', },
	],
	heap => $self,
	( ref($options) eq 'HASH' ? ( options => $options ) : () ),
  )->ID();
  return $self;
}

sub _start {
  my ($kernel,$sender,$self) = @_[KERNEL,SENDER,OBJECT];
  $self->{session_id} = $_[SESSION]->ID();
  if ( $kernel == $sender and !$self->{session} ) {
	croak "Not called from another POE session and 'session' wasn't set\n";
  }
  my $sender_id;
  if ( $self->{session} ) {
    if ( my $ref = $kernel->alias_resolve( $self->{session} ) ) {
	$sender_id = $ref->ID();
    }
    else {
	croak "Could not resolve 'session' to a valid POE session\n";
    }
  }
  else {
    $sender_id = $sender->ID();
  }
  $kernel->refcount_increment( $sender_id, __PACKAGE__ );
  $kernel->detach_myself();
  $self->{sender_id} = $sender_id;
  if ( $self->{epoch} ) {
    $kernel->yield( '_epoch' );
    return;
  }
  $kernel->yield( '_process_' . $self->{uri}->scheme );
  return;
}

sub _recent {
  my ($kernel,$self,$type) = @_[KERNEL,OBJECT,ARG0];
  my $target = delete $self->{sender_id};
  my %reply;
  $reply{recent} = delete $self->{recent} if $self->{recent};
  $reply{error} = delete $self->{error} if $self->{error};
  $reply{context} = delete $self->{context} if $self->{context};
  $reply{url} = delete $self->{url};
  @{ $reply{recent} } = grep { my @parts = split m!/!; $parts[3] !~ m!^perl6$!i } @{ $reply{recent} };
  my $event = delete $self->{event};
  $kernel->post( $target, $event, \%reply );
  $kernel->refcount_decrement( $target, __PACKAGE__ );
  return;
}

sub _process_http {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  my @path = $self->{rss} ? ( 'modules', '01modules.mtime.rss' ) : ( 'RECENT' );
  $self->{uri}->path( File::Spec::Unix->catfile( $self->{uri}->path(), @path ) );
  POE::Component::SmokeBox::Recent::HTTP->spawn(
	uri => $self->{uri},
  );
  return;
}

sub _http_response {
  my ($kernel,$self,$response) = @_[KERNEL,OBJECT,ARG0];
  if ( $response->code() == 200 ) {
    if ( $self->{rss} ) {
      for ( split /\n/, $response->content() ) {
        next unless m#(.+?)#i;
        next unless m#by-authors#i;
        my ($link) = $_ =~ m#id/(.+?)\s*$#i;
        next unless $link;
        unshift @{ $self->{recent} }, $link;
      }
    }
    else {
      for ( split /\n/, $response->content() ) {
        next unless /^authors/;
        next unless /\.(tar\.gz|tgz|tar\.bz2|zip)$/;
        s!authors/id/!!;
        push @{ $self->{recent} }, $_;
      }
    }
  }
  else {
    $self->{error} = $response->as_string();
  }
  $kernel->yield( '_recent', 'http' );
  return;
}

sub _process_ftp {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  my @path = $self->{rss} ? ( 'modules', '01modules.mtime.rss' ) : ( 'RECENT' );
  POE::Component::SmokeBox::Recent::FTP->spawn(
        Username => 'anonymous',
        Password => '[e-mail]',
        address  => $self->{uri}->host,
	      port	   => $self->{uri}->port,
	      path     => File::Spec::Unix->catfile( $self->{uri}->path, @path ),
  );
  return;
}

sub _get_connect_error {
  my ($kernel,$self,@args) = @_[KERNEL,OBJECT,ARG0..$#_];
  $self->{error} = join ' ', @args;
  $kernel->yield( '_recent', 'ftp' );
  return;
}

sub _get_error {
  my ($kernel,$self,$sender,@args) = @_[KERNEL,OBJECT,SENDER,ARG0..$#_];
  $self->{error} = join ' ', @args;
  $kernel->yield( '_recent', 'ftp' );
  return;
}

sub _get_data {
  my ($kernel,$self,$data) = @_[KERNEL,OBJECT,ARG0];
  $data =~ s![\x0D\x0A]+$!!g;
  if ( $self->{rss} ) {
    return unless $data =~ m#(.+?)#i;
    return unless $data =~ m#by-authors#i;
    my ($link) = $data =~ m#id/(.+?)\s*$#i;
    return unless $link;
    unshift @{ $self->{recent} }, $link;
  }
  elsif ( $self->{epoch} ) {
    push @{ $self->{recent} }, $data;
  }
  else {
    return unless $data =~ /^authors/i;
    return unless $data =~ /\.(tar\.gz|tgz|tar\.bz2|zip)$/;
    $data =~ s!authors/id/!!;
    push @{ $self->{recent} }, $data;
  }
  return;
}

sub _get_done {
  my ($kernel,$self,$sender) = @_[KERNEL,OBJECT,SENDER];
  $kernel->yield( '_recent', 'ftp' );
  return;
}

sub _process_file {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  delete $self->{_epoch_fail};
  {
    my @segs = $self->{uri}->path_segments;
    pop @segs unless $segs[-1];
    push @segs, 'RECENT';
    $self->{uri}->path_segments( @segs );
  }
  $self->{wheel} = POE::Wheel::Run->new(
      Program => sub {
        my $path = shift;
        open my $fh, '<', $path or die "$!\n";
        while (<$fh>) {
          print STDOUT $_;
        }
        close $fh;
      },
      ProgramArgs => [ $self->{uri}->file ],
      StdoutEvent => 'ftp_data',
  );
  $kernel->sig_child( $self->{wheel}->PID(), '_sig_child' );
  return;
}

sub _epoch {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  require CPAN::Recent::Uploads;
  $self->{wheel} = POE::Wheel::Run->new(
      Program => sub {
        my $epoch  = shift;
        my $mirror = shift;
        print STDOUT $_, "\n" for
          CPAN::Recent::Uploads->recent( $epoch, $mirror );
      },
      ProgramArgs => [ $self->{epoch}, $self->{uri}->as_string ],
      StdoutEvent => 'ftp_data',
      StderrEvent => '_epoch_fail',
  );
  $kernel->sig_child( $self->{wheel}->PID(), '_sig_child' );
  return;
}

sub _epoch_fail {
  my ($kernel,$self,$data) = @_[KERNEL,OBJECT,ARG0];
  # Anything on STDERR means an error
  return if $self->{_epoch_fail};
  $self->{_epoch_fail} = 1;
  $kernel->yield( '_process_' . $self->{uri}->scheme );
  return;
}

sub _sig_child {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  delete $self->{wheel};
  $kernel->yield( '_recent', 'file' ) unless $self->{_epoch_fail};
  $kernel->sig_handled();
}

qq[What's the road on the street?];

__END__