2 import urllib
.request
as urllib_request
3 import urllib
.error
as urllib_error
6 import urllib2
as urllib_request
7 import urllib2
as urllib_error
9 from ssl
import SSLError
12 from .api
import TwitterCall
, wrap_response
14 class TwitterJSONIter(object):
16 def __init__(self
, handle
, uri
, arg_data
, block
=True):
17 self
.decoder
= json
.JSONDecoder()
23 sock
= self
.handle
.fp
._sock
.fp
._sock
24 sock
.setsockopt(socket
.SOL_SOCKET
, socket
.SO_KEEPALIVE
, 1)
26 sock
.setblocking(False)
29 utf8_buf
= self
.buf
.decode('utf8').lstrip()
30 res
, ptr
= self
.decoder
.raw_decode(utf8_buf
)
31 self
.buf
= utf8_buf
[ptr
:].encode('utf8')
32 yield wrap_response(res
, self
.handle
.headers
)
34 except ValueError as e
:
39 except urllib_error
.HTTPError
as e
:
40 raise TwitterHTTPError(e
, uri
, self
.format
, arg_data
)
41 # this is a non-blocking read (ie, it will return if any data is available)
43 self
.buf
+= sock
.recv(1024)
45 if (not self
.block
) and (e
.errno
== 2):
46 # Apparently this means there was nothing in the socket buf
51 def handle_stream_response(req
, uri
, arg_data
, block
):
52 handle
= urllib_request
.urlopen(req
,)
53 return iter(TwitterJSONIter(handle
, uri
, arg_data
, block
))
55 class TwitterStreamCall(TwitterCall
):
56 def _handle_response(self
, req
, uri
, arg_data
, _timeout
=None):
57 return handle_stream_response(req
, uri
, arg_data
, block
=True)
59 class TwitterStreamCallNonBlocking(TwitterCall
):
60 def _handle_response(self
, req
, uri
, arg_data
, _timeout
=None):
61 return handle_stream_response(req
, uri
, arg_data
, block
=False)
63 class TwitterStream(TwitterStreamCall
):
65 The TwitterStream object is an interface to the Twitter Stream API
66 (stream.twitter.com). This can be used pretty much the same as the
67 Twitter class except the result of calling a method will be an
68 iterator that yields objects decoded from the stream. For
71 twitter_stream = TwitterStream(auth=UserPassAuth('joe', 'joespassword'))
72 iterator = twitter_stream.statuses.sample()
74 for tweet in iterator:
75 ...do something with this tweet...
77 The iterator will yield tweets forever and ever (until the stream
78 breaks at which point it raises a TwitterHTTPError.)
80 The `block` parameter controls if the stream is blocking. Default
81 is blocking (True). When set to False, the iterator will
82 occasionally yield None when there is no available message.
85 self
, domain
="stream.twitter.com", secure
=True, auth
=None,
86 api_version
='1', block
=True):
88 uriparts
+= (str(api_version
),)
91 call_cls
= TwitterStreamCall
93 call_cls
= TwitterStreamCallNonBlocking
95 TwitterStreamCall
.__init
__(
96 self
, auth
=auth
, format
="json", domain
=domain
,
97 callable_cls
=call_cls
,
98 secure
=secure
, uriparts
=uriparts
)