File Coverage

blib/lib/POE/Component/SmokeBox/Uploads/RSS.pm

Criterion	Covered	Total	%
statement	12	12	100.0
branch			n/a
condition			n/a
subroutine	4	4	100.0
pod			n/a
total	16	16	100.0

package POE::Component::SmokeBox::Uploads::RSS;

use strict;
use warnings;
use Carp;
use POE qw(Component::RSSAggregator Component::Client::HTTP);
use HTTP::Request;
use HTML::LinkExtor;
use vars qw($VERSION);

$VERSION = '1.00';

sub spawn {
  my $package = shift;
  my %opts = @_;
  $opts{lc $_} = delete $opts{$_} for keys %opts;
  croak "$package requires an 'event' argument\n" unless $opts{event};
  $opts{feed} = 'http://search.cpan.org/uploads.rdf' unless $opts{feed};
  $opts{name} = 'search-cpan-recent' unless $opts{name};
  $opts{delay} = 1800 unless $opts{delay};
  my $options = delete $opts{options};
  my $self = bless \%opts, $package;
  $self->{session_id} = POE::Session->create(
        object_states => [
	   $self => { shutdown => '_shutdown', },
           $self => [ qw(_start _dispatch _feed_url _handle_feed _real_shutdown) ],
        ],
        heap => $self,
        ( ref($options) eq 'HASH' ? ( options => $options ) : () ),
  )->ID();
  return $self;
}

sub session_id {
  return $_[0]->{session_id};
}

sub shutdown {
  my $self = shift;
  $poe_kernel->post( $self->{session_id}, 'shutdown' );
  return;
}

sub _shutdown {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  $self->{_shutdown} = 1;
  return if $self->{_http_requests};
  $kernel->yield( '_real_shutdown' );
  return;
}

sub _real_shutdown {
  my ($kernel,$self) = @_[KERNEL,OBJECT];
  $kernel->alias_remove( $_ ) for $kernel->alias_list();
  $kernel->refcount_decrement( $self->{session_id}, __PACKAGE__ ) unless $self->{alias};
  $kernel->refcount_decrement( $self->{sender_id}, __PACKAGE__ );
  $kernel->post( $self->{http_id}, 'shutdown' ) unless $self->{http_alias};
  $kernel->post( $self->{rssagg}, 'shutdown' );
  return;
}

sub _start {
  my ($kernel,$session,$sender,$self) = @_[KERNEL,SESSION,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__ );
  $self->{sender_id} = $sender_id;
  if ( $self->{http_alias} ) {
     my $http_ref = $kernel->alias_resolve( $self->{http_alias} );
     $self->{http_id} = $http_ref->ID() if $http_ref;
  }
  unless ( $self->{http_id} ) {
    $self->{http_id} = 'smokeboxrss' . $$ . $self->{session_id};
    POE::Component::Client::HTTP->spawn(
	Alias           => $self->{http_id},
	FollowRedirects => 2,
        Timeout         => 60,
        Agent           => 'Mozilla/5.0 (X11; U; Linux i686; en-US; '
                . 'rv:1.1) Gecko/20020913 Debian/1.1-1',
    );
  }
  $self->{rssagg} = 'rssagg' . $self->{session_id};
  POE::Component::RSSAggregator->new(
            alias    => $self->{rssagg},
            callback => $session->postback('_handle_feed'),
            http_alias => $self->{http_id},
            tmpdir   => $self->{tmpdir} || '.',        # optional caching
  );
  my $feed = {
                url   => $self->{feed},
                name  => $self->{name},
                delay => $self->{delay},
  };
  $kernel->post( $self->{rssagg}, 'add_feed', $feed );
  return;
}

sub _handle_feed {
  my ($kernel,$self,$feed) = (@_[KERNEL,OBJECT], $_[ARG1]->[0]);
  for my $headline ( reverse $feed->late_breaking_news ) {
    $kernel->post(
        $self->{http_id},
        'request',
        '_feed_url',
        HTTP::Request->new( GET => $headline->url ),
        $headline->headline,
    );
    $self->{_http_requests}++;
  }
  return;
}

sub _feed_url {
  my ($kernel,$self,$request_packet,$response_packet) = @_[KERNEL,OBJECT,ARG0,ARG1];
  my $http_resp = $response_packet->[0];
  $self->{_http_requests}--;
  return unless $http_resp and $http_resp->code() == 200;
  my $tag    = $request_packet->[1];
  my $p = HTML::LinkExtor->new();
  $p->parse( $http_resp->content() );
  foreach my $link ( $p->links() ) {
     if ( $link->[0] eq 'a' and $link->[2] =~ /\Q$tag\E/ ) {
        ( my $module = $link->[2] ) =~ s#/CPAN/authors/id/##;
	$kernel->call( $self->{session_id}, '_dispatch', $module );
        last;
     }
  }
  $kernel->yield( '_real_shutdown' ) if $self->{_shutdown} and $self->{_http_requests} == 0;
  return;
}

sub _dispatch {
  my ($kernel,$self,$module) = @_[KERNEL,OBJECT,ARG0];
  $kernel->post( $self->{sender_id}, $self->{event}, $module );
  return;
}

"This town ain't big enough for the both of us";

__END__