TwitterStreamingAPI.pm

package Net::TwitterStreamingAPI;
use utf8;
use strict;
use warnings;

use JSON;
use LWP::UserAgent;
use HTTP::Request;
use POE::Filter::Line;

use Data::Dumper;

use base qw(Class::Accessor::Fast);
__PACKAGE__->mk_accessors(qw/
    ua
    filter
    uri
    postdata
    content_cb
/);

sub new {
    my $class = shift;
    my $self  = $class->SUPER::new();
    $self->_setup(@_);
    return $self;
}

sub _setup {
    my $self = shift;
    my %opt = (ref $_[0]) ? %{$_[0]} : @_;

    $self->ua($opt{ua} || LWP::UserAgent->new);
    $self->uri(
        (defined $opt{api})
            ? ($opt{api} =~ m|^http://|)
                ? URI->new($opt{api})
                : _streamapi_url(\%opt)
            : die 'specify api or url.'
    );
    $self->postdata($opt{query}->{post}) if ($opt{query} && ref $opt{query});
    $self->content_cb($opt{content_cb});

    $self->filter(POE::Filter::Line->new);
    $self->ua->credentials(
        $self->uri->host_port,
        'Firehose',
        $opt{username},
        $opt{password},
    );

    return $self;
}

sub _streamapi_url {
    my $opt = shift;

    my $api_name = $opt->{api} or die 'specify api name.';
    my $uri = URI->new('http://stream.twitter.com/'.$api_name.'.json');

    my $query = $opt->{query}->{get};
    if ($opt->{query} && ref $opt->{query} && ref $opt->{query}->{get}) {
        $uri->query_form($opt->{query}->{get});
    }

    return $uri;
}

sub get {
    my $self = shift;
    my %opt = (ref $_[0]) ? %{$_[0]} : @_;

    my $callback = $opt{callback} || sub {
        print $_[1]."\n";
    };
    
    my $req;
    if ($self->postdata) {
        $req = HTTP::Request->new('post', $self->uri);
        $req->content_type('application/x-www-form-urlencoded');
        $req->content($self->postdata);
    }
    else {
        $req = HTTP::Request->new('get', $self->uri);
    }
    
    $self->ua->request(
        $req,
        $self->content_cb || sub {
            my $chunk = shift;
            my $lines = $self->filter->get([ $chunk ]);

            for my $line (@$lines) {
                if (my $data = from_json($line)) {
                    &{$callback}($data, $line);
                }
            }
        }
    ) if ($req);
}

1;