AnyEvent-Twitter-Stream

 view release on metacpan or  search on metacpan

eg/track.pl  view on Meta::CPAN

        my $error = shift;
        warn "ERROR: $error";
        $done->send;
    },
    on_eof   => sub {
        $done->send;
    },
);

# uncomment to test undef $streamer
# my $t = AE::timer 1, 0, sub { undef $streamer };

$done->recv;

lib/AnyEvent/Twitter/Stream.pm  view on Meta::CPAN

    }else{
        $auth = "Basic ".MIME::Base64::encode("$username:$password", '');
    }

    my $self = bless {}, $class;

    {
        Scalar::Util::weaken(my $self = $self);

        my $set_timeout = $timeout
            ? sub { $self->{timeout} = AE::timer($timeout, 0, sub { $on_error->('timeout') }) }
            : sub {};

        my $on_json_message = sub {
            my ($json) = @_;

            # Twitter stream returns "\x0a\x0d\x0a" if there's no matched tweets in ~30s.
            $set_timeout->();
            if ($json !~ /^\s*$/) {
                my $tweet = $decode_json ? JSON::decode_json($json) : $json;
                if ($on_delete && $tweet->{delete} && $tweet->{delete}->{status}) {

t/streaming.t  view on Meta::CPAN

            $writer->write(encode_json({
                hello => 1,
                path => $req->path,
                request_method => $req->method,
                user => $env->{REMOTE_USER},
                param => $req->parameters->mixed,
            }) . "\x0D\x0A");

            my $count = 1;
            my $t; $t = AE::timer(0, 0.2, sub {
                try {
                    $writer->write(encode_json({
                        body => 'x' x 500,
                        count => $count++,
                    }) . "\x0D\x0A");
                } catch {
                    undef $t;
                };
                if ($req->path =~ /sample/ && $count == 2) {
                    try {

t/streaming.t  view on Meta::CPAN

            my $respond = shift;

            my $writer = $respond->([200, [
                'Content-Type' => 'application/json',
                'Server' => 'Jetty(6.1.17)',
            ]]);
            $writer->write(encode_json({
                friends => [qw/1 2 3/],
            }) . "\x0D\x0A");

            my $t; $t = AE::timer(0, 0.2, sub {
                try {
                    $writer->write(encode_json({
                        event => {foo => 'bar'},
                    }) . "\x0D\x0A");
                }catch{
                    undef $t;
                };
            });
        };
    };

t/timeout.t  view on Meta::CPAN

            my $send_tweet = sub {
                my ($tweet) = @_;

                try {
                    $writer->write(encode_json({count => $count++, rand => rand}) . "\x0D\x0A");
                } catch {
                    note($_);
                };
            };

            my $t1; $t1 = AE::timer $wait_a, 0, sub {
                $writer = $respond->([200, [
                    'Content-Type' => 'application/json'
                ]]);

                undef $t1;
            };

            my $t2; $t2 = AE::timer $wait_a + $wait_b, 0, sub {
                $send_tweet->();

                undef $t2;
            };

            my $t3; $t3 = AE::timer $wait_a + $wait_b + $wait_c, 0.5, sub {
                $send_tweet->();

                $t3;
            };
        };
    };


    my $app = builder {
        enable 'Auth::Basic', realm => 'Firehose', authenticator => sub {