File Coverage

blib/lib/AnyEvent/Twitter/Stream.pm
Criterion Covered Total %
statement 30 145 20.6
branch 0 52 0.0
condition 0 39 0.0
subroutine 10 20 50.0
pod 1 1 100.0
total 41 257 15.9


line stmt bran cond sub pod time code
1             package AnyEvent::Twitter::Stream;
2              
3 3     3   21861 use strict;
  3         4  
  3         89  
4 3     3   54 use 5.008_001;
  3         8  
  3         109  
5             our $VERSION = '0.23';
6              
7 3     3   1879 use AnyEvent;
  3         8190  
  3         63  
8 3     3   1880 use AnyEvent::HTTP;
  3         74384  
  3         293  
9 3     3   26 use AnyEvent::Util;
  3         4  
  3         155  
10 3     3   1563 use MIME::Base64;
  3         1718  
  3         146  
11 3     3   2344 use URI;
  3         17565  
  3         80  
12 3     3   19 use URI::Escape;
  3         3  
  3         164  
13 3     3   12 use Carp;
  3         3  
  3         131  
14 3     3   1899 use Compress::Raw::Zlib;
  3         14272  
  3         5029  
15              
16             our $STREAMING_SERVER = 'stream.twitter.com';
17             our $USERSTREAM_SERVER = 'userstream.twitter.com';
18             our $SITESTREAM_SERVER = 'sitestream.twitter.com';
19             our $PROTOCOL = 'https';
20             our $US_PROTOCOL = 'https'; # for testing
21              
22             my %methods = (
23             filter => [ POST => sub { "$PROTOCOL://$STREAMING_SERVER/1.1/statuses/filter.json" } ],
24             sample => [ GET => sub { "$PROTOCOL://$STREAMING_SERVER/1.1/statuses/sample.json" } ],
25             firehose => [ GET => sub { "$PROTOCOL://$STREAMING_SERVER/1.1/statuses/firehose.json" } ],
26             userstream => [ GET => sub { "$US_PROTOCOL://$USERSTREAM_SERVER/1.1/user.json" } ],
27             sitestream => [ GET => sub { "$PROTOCOL://$SITESTREAM_SERVER/1.1/site.json" } ],
28              
29             # DEPRECATED
30             links => [ GET => sub { "$PROTOCOL://$STREAMING_SERVER/1/statuses/links.json" } ],
31             retweet => [ GET => sub { "$PROTOCOL://$STREAMING_SERVER/1/statuses/retweet.json" } ],
32             );
33              
34             sub new {
35 0     0 1   my $class = shift;
36 0           my %args = @_;
37              
38 0           my $username = delete $args{username};
39 0           my $password = delete $args{password};
40 0           my $consumer_key = delete $args{consumer_key};
41 0           my $consumer_secret = delete $args{consumer_secret};
42 0           my $token = delete $args{token};
43 0           my $token_secret = delete $args{token_secret};
44 0           my $method = delete $args{method};
45 0   0 0     my $on_connect = delete $args{on_connect} || sub { };
  0            
46 0           my $on_tweet = delete $args{on_tweet};
47 0   0 0     my $on_error = delete $args{on_error} || sub { die @_ };
  0            
48 0   0 0     my $on_eof = delete $args{on_eof} || sub { };
  0            
49 0   0 0     my $on_keepalive = delete $args{on_keepalive} || sub { };
  0            
50 0           my $on_delete = delete $args{on_delete};
51 0           my $on_friends = delete $args{on_friends};
52 0           my $on_direct_message = delete $args{on_direct_message};
53 0           my $on_event = delete $args{on_event};
54 0           my $timeout = delete $args{timeout};
55              
56 0           my $decode_json;
57 0 0         unless (delete $args{no_decode_json}) {
58 0           require JSON;
59 0           $decode_json = 1;
60             }
61              
62 0           my ($zlib, my $_zstatus);
63 0 0         if (delete $args{use_compression}){
64 0           ($zlib, $_zstatus) = Compress::Raw::Zlib::Inflate->new(
65             -LimitOutput => 1,
66             -AppendOutput => 1,
67             -WindowBits => WANT_GZIP_OR_ZLIB,
68             );
69 0 0         die "Can't make inflator: $_zstatus" unless $zlib;
70             }
71              
72 0 0 0       unless ($methods{$method} || exists $args{api_url} ) {
73 0           $on_error->("Method $method not available.");
74 0           return;
75             }
76              
77 0   0       my $uri = URI->new(delete $args{api_url} || $methods{$method}[1]());
78              
79 0           my $request_body;
80 0   0       my $request_method = delete $args{request_method} || $methods{$method}[0] || 'GET';
81 0 0         if ( $request_method eq 'POST' ) {
82 0           $request_body = join '&', map "$_=" . URI::Escape::uri_escape($args{$_}), keys %args;
83             }else{
84 0           $uri->query_form(%args);
85             }
86              
87 0           my $auth;
88 0 0         if ($consumer_key) {
89 0           eval {require Net::OAuth;};
  0            
90 0 0         die $@ if $@;
91              
92 0 0         my $request = Net::OAuth->request('protected resource')->new(
93             version => '1.0',
94             consumer_key => $consumer_key,
95             consumer_secret => $consumer_secret,
96             token => $token,
97             token_secret => $token_secret,
98             request_method => $request_method,
99             signature_method => 'HMAC-SHA1',
100             timestamp => time,
101             nonce => MIME::Base64::encode( time . $$ . rand ),
102             request_url => $uri,
103             $request_method eq 'POST' ? (extra_params => \%args) : (),
104             );
105 0           $request->sign;
106 0           $auth = $request->to_authorization_header;
107             }else{
108 0           $auth = "Basic ".MIME::Base64::encode("$username:$password", '');
109             }
110              
111 0           my $self = bless {}, $class;
112              
113             {
114 0           Scalar::Util::weaken(my $self = $self);
  0            
115              
116             my $set_timeout = $timeout
117 0     0     ? sub { $self->{timeout} = AE::timer($timeout, 0, sub { $on_error->('timeout') }) }
  0            
118 0 0   0     : sub {};
  0            
119              
120             my $on_json_message = sub {
121 0     0     my ($json) = @_;
122              
123             # Twitter stream returns "\x0a\x0d\x0a" if there's no matched tweets in ~30s.
124 0           $set_timeout->();
125 0 0         if ($json !~ /^\s*$/) {
126 0 0         my $tweet = $decode_json ? JSON::decode_json($json) : $json;
127 0 0 0       if ($on_delete && $tweet->{delete} && $tweet->{delete}->{status}) {
    0 0        
    0 0        
    0 0        
      0        
128 0           $on_delete->($tweet->{delete}->{status}->{id}, $tweet->{delete}->{status}->{user_id});
129             }elsif($on_friends && $tweet->{friends}) {
130 0           $on_friends->($tweet->{friends});
131             }elsif($on_direct_message && $tweet->{direct_message}) {
132 0           $on_direct_message->($tweet->{direct_message});
133             }elsif($on_event && $tweet->{event}) {
134 0           $on_event->($tweet);
135             }else{
136 0           $on_tweet->($tweet);
137             }
138             }
139             else {
140 0           $on_keepalive->();
141             }
142 0           };
143              
144 0           $set_timeout->();
145              
146             $self->{connection_guard} = http_request($request_method, $uri,
147             headers => {
148             Accept => '*/*',
149             ( defined $zlib ? ('Accept-Encoding' => 'deflate, gzip') : ()),
150             Authorization => $auth,
151             ($request_method eq 'POST'
152             ? ('Content-Type' => 'application/x-www-form-urlencoded')
153             : ()
154             ),
155             },
156             body => $request_body,
157             on_header => sub {
158 0     0     my($headers) = @_;
159 0 0         if ($headers->{Status} ne '200') {
160 0           $on_error->("$headers->{Status}: $headers->{Reason}");
161 0           return;
162             }
163 0           return 1;
164             },
165             want_body_handle => 1, # for some reason on_body => sub {} doesn't work :/
166             sub {
167 0     0     my ($handle, $headers) = @_;
168              
169 0 0         return unless $handle;
170 0           my $input;
171             my $chunk_reader = sub {
172 0           my ($handle, $line) = @_;
173              
174 0 0         $line =~ /^([0-9a-fA-F]+)/ or die 'bad chunk (incorrect length)';
175 0           my $len = hex $1;
176              
177             $handle->push_read(chunk => $len, sub {
178 0           my ($handle, $chunk) = @_;
179 0 0         $handle->push_read(line => sub { length $_[1] and die 'bad chunk (missing last empty line)'; });
  0            
180              
181 0 0         unless ($headers->{'content-encoding'}) {
    0          
182 0           $on_json_message->($chunk);
183             } elsif ($headers->{'content-encoding'} =~ 'deflate|gzip') {
184 0           $input .= $chunk;
185 0           my ($message);
186 0   0       do {
187 0           $_zstatus = $zlib->inflate(\$input, \$message);
188 0 0 0       return unless $_zstatus == Z_OK || $_zstatus == Z_BUF_ERROR;
189             } while ( $_zstatus == Z_OK && length $input );
190 0           $on_json_message->($message);
191             } else {
192 0           die "Don't know how to decode $headers->{'content-encoding'}"
193             }
194 0           });
195 0           };
196             my $line_reader = sub {
197 0           my ($handle, $line) = @_;
198              
199 0           $on_json_message->($line);
200 0           };
201              
202             $handle->on_error(sub {
203 0           undef $handle;
204 0           $on_error->($_[2]);
205 0           });
206             $handle->on_eof(sub {
207 0           undef $handle;
208 0           $on_eof->(@_);
209 0           });
210              
211 0 0 0       if (($headers->{'transfer-encoding'} || '') =~ /\bchunked\b/i) {
212             $handle->on_read(sub {
213 0           my ($handle) = @_;
214 0           $handle->push_read(line => $chunk_reader);
215 0           });
216             } else {
217             $handle->on_read(sub {
218 0           my ($handle) = @_;
219 0           $handle->push_read(line => $line_reader);
220 0           });
221             }
222              
223             $self->{guard} = AnyEvent::Util::guard {
224 0 0         $handle->destroy if $handle;
225 0           };
226              
227 0           $on_connect->();
228             }
229 0 0         );
    0          
230             }
231              
232 0           return $self;
233             }
234              
235             1;
236             __END__