view release on metacpan or  search on metacpan

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

                        } else {
                            die "Don't know how to decode $headers->{'content-encoding'}"
                        }
                        $handle->push_read(line => $chunk_reader);
                    }; # chunk_part_reader
                    $handle->push_read(chunk => $len + 2, $chunk_part_reader);
                };

                my $line_reader = sub {
                    my ($handle, $line) = @_;

                    $on_json_message->($line);
                };

                $handle->on_error(sub {
                    undef $handle;
                    $on_error->($_[2]);
                });
                $handle->on_eof(sub {
                    undef $handle;
                    $on_eof->(@_);
                });

                if (($headers->{'transfer-encoding'} || '') =~ /\bchunked\b/i) {
                    $handle->on_read(sub {
                        my ($handle) = @_;
                        $handle->push_read(line => $chunk_reader);
                    });
                } else {
                    $handle->on_read(sub {
                        my ($handle) = @_;
                        $handle->push_read(line => $line_reader);
                    });
                }

                $self->{guard} = AnyEvent::Util::guard {
                    $handle->destroy if $handle;
                };

                $on_connect->();
            }
        );
    }

    return $self;
}

1;
__END__

=encoding utf-8

=for stopwords
API AnyEvent

=for test_synopsis
my($user, $password, @following_ids, $consumer_key, $consumer_secret, $token, $token_secret);

=head1 NAME

AnyEvent::Twitter::Stream - Receive Twitter streaming API in an event loop

=head1 SYNOPSIS

  use AnyEvent::Twitter::Stream;
  
  my $done = AE::cv;

  # receive updates from @following_ids
  my $listener = AnyEvent::Twitter::Stream->new(
      username => $user,
      password => $password,
      method   => "filter",  # "firehose" for everything, "sample" for sample timeline
      follow   => join(",", @following_ids), # numeric IDs
      on_tweet => sub {
          my $tweet = shift;
          warn "$tweet->{user}{screen_name}: $tweet->{text}\n";
      },
      on_keepalive => sub {
          warn "ping\n";
      },
      on_delete => sub {
          my ($tweet_id, $user_id) = @_; # callback executed when twitter send a delete notification
          ...
      },
      timeout => 45,
  );

  # track keywords
  my $guard = AnyEvent::Twitter::Stream->new(
      username => $user,
      password => $password,
      method   => "filter",
      track    => "Perl,Test,Music",
      on_tweet => sub { },
  );

  # to use OAuth authentication
  my $listener = AnyEvent::Twitter::Stream->new(
      consumer_key    => $consumer_key,
      consumer_secret => $consumer_secret,
      token           => $token,
      token_secret    => $token_secret,
      method          => "filter",
      track           => "...",
      on_tweet        => sub { ... },
  );
  
  $done->recv;

=head1 DESCRIPTION

AnyEvent::Twitter::Stream is an AnyEvent user to receive Twitter streaming
API, available at L<http://dev.twitter.com/pages/streaming_api> and
L<http://dev.twitter.com/pages/user_streams>.

See L<eg/track.pl> for more client code example.

=head1 METHODS

=head2 my $streamer = AnyEvent::Twitter::Stream->new(%args);

=over 4

=item B<username> B<password>

These arguments are used for basic authentication.

=item B<consumer_key> B<consumer_secret> B<token> B<token_secret>

If you want to use the OAuth authentication mechanism, you need to set these arguments

=item B<method>

The name of the method you want to use on the stream. Currently, any one of :

=over 2

=item B<firehose>

=item B<sample>

=item B<userstream>

To use this method, you need to use the OAuth mechanism.

=item B<filter>

With this method you can specify what you want to filter amongst B<track>, B<follow> and B<locations>.

See L<https://dev.twitter.com/docs/api/1.1/post/statuses/filter> for the details of the parameters.

=back

=item B<api_url>

Pass this to override the default URL for the API endpoint.

=item B<request_method>

Pass this to override the default HTTP request method.

=item B<timeout>

Set the timeout value.

=item B<on_connect>

Callback to execute when a stream is connected.

=item B<on_tweet>

Callback to execute when a new tweet is received. The argument is the tweet, a hashref documented at
L<https://dev.twitter.com/docs/api/1/get/statuses/show/%3Aid>.