A Simpler Fix to the Streaming Code due to Changes from Twitter on Jan. 13, 2014.

.gitignore

@@ -8,3 +8,4 @@ twitter3.egg-info
 *~
 dist
 build
+.idea

twitter/api.py

@@ -128,7 +128,7 @@ class TwitterCall(object):
 
     def __init__(
         self, auth, format, domain, callable_cls, uri="",
-        uriparts=None, secure=True, timeout=None):
+        uriparts=None, secure=True, timeout=None, gzip=False):
         self.auth = auth
         self.format = format
         self.domain = domain
@@ -137,6 +137,7 @@ def __init__(
         self.uriparts = uriparts
         self.secure = secure
         self.timeout = timeout
+        self.gzip = gzip
 
     def __getattr__(self, k):
         try:
@@ -145,9 +146,9 @@ def __getattr__(self, k):
             def extend_call(arg):
                 return self.callable_cls(
                     auth=self.auth, format=self.format, domain=self.domain,
-                    callable_cls=self.callable_cls, timeout=self.timeout, uriparts=self.uriparts \
-                        + (arg,),
-                    secure=self.secure)
+                    callable_cls=self.callable_cls, timeout=self.timeout,
+                    secure=self.secure, gzip=self.gzip,
+                    uriparts=self.uriparts + (arg,))
             if k == "_":
                 return extend_call
             else:
@@ -194,7 +195,7 @@ def __call__(self, **kwargs):
         uriBase = "http%s://%s/%s%s%s" %(
                     secure_str, self.domain, uri, dot, self.format)
 
-        headers = {'Accept-Encoding': 'gzip'}
+        headers = {'Accept-Encoding': 'gzip'} if self.gzip else dict()
         if self.auth:
             headers.update(self.auth.generate_headers())
             arg_data = self.auth.encode_params(uriBase, method, kwargs)

twitter/stream.py

@@ -17,6 +17,8 @@ class TwitterJSONIter(object):
     def __init__(self, handle, uri, arg_data, block=True, timeout=None):
         self.decoder = json.JSONDecoder()
         self.handle = handle
+        self.uri = uri
+        self.arg_data = arg_data
         self.buf = b""
         self.block = block
         self.timeout = timeout
@@ -34,6 +36,9 @@ def __iter__(self):
         while True:
             try:
                 utf8_buf = self.buf.decode('utf8').lstrip()
+                if utf8_buf and utf8_buf[0] != '{':  # Remove the hex delimiter length and extra whitespace.
+                    utf8_buf = utf8_buf.lstrip('0123456789abcdefABCDEF')
+                    utf8_buf = utf8_buf.lstrip()
                 res, ptr = self.decoder.raw_decode(utf8_buf)
                 self.buf = utf8_buf[ptr:].encode('utf8')
                 yield wrap_response(res, self.handle.headers)
@@ -44,8 +49,8 @@ def __iter__(self):
                     pass
                 else:
                     yield None
-            except urllib_error.HTTPError as e:
-                raise TwitterHTTPError(e, uri, self.format, arg_data)
+            except urllib_error.HTTPError as e:  # Probably unnecessary, no dynamic url calls in the try block.
+                raise TwitterHTTPError(e, self.uri, 'json', self.arg_data)
             # this is a non-blocking read (ie, it will return if any data is available)
             try:
                 if self.timeout:
@@ -57,13 +62,15 @@ def __iter__(self):
                     else:
                         yield {"timeout":True}
                 else:
-                    self.buf += sock.recv(1024)
+                    self.buf += sock.recv(1024)  # As tweets are typically longer than 1KB, consider increasing this size.
             except SSLError as e:
                 if (not self.block or self.timeout) and (e.errno == 2):
                     # Apparently this means there was nothing in the socket buf
                     pass
                 else:
                     raise
+            except urllib_error.HTTPError as e:
+                raise TwitterHTTPError(e, self.uri, 'json', self.arg_data)
 
 def handle_stream_response(req, uri, arg_data, block, timeout=None):
     handle = urllib_request.urlopen(req,)
@@ -119,4 +126,4 @@ def __init__(
         TwitterStreamCall.__init__(
             self, auth=auth, format="json", domain=domain,
             callable_cls=call_cls,
-            secure=secure, uriparts=uriparts, timeout=timeout)
+            secure=secure, uriparts=uriparts, timeout=timeout, gzip=False)

twitter/stream_example.py

@@ -4,25 +4,39 @@
 
 USAGE
 
-  twitter-stream-example <username> <password>
+  stream-example -t <token> -ts <token_secret> -ck <consumer_key> -cs <consumer_secret>
 
 """
 
 from __future__ import print_function
 
-import sys
+import argparse
 
-from .stream import TwitterStream
-from .auth import UserPassAuth
-from .util import printNicely
+from twitter.stream import TwitterStream
+from twitter.oauth import OAuth
+from twitter.util import printNicely
 
-def main(args=sys.argv[1:]):
-    if not args[1:]:
-        print(__doc__)
-        return 1
 
-    # When using twitter stream you must authorize. UserPass or OAuth.
-    stream = TwitterStream(auth=UserPassAuth(args[0], args[1]))
+def parse_arguments():
+
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument('-t',  '--token', help='The Twitter Access Token.')
+    parser.add_argument('-ts', '--token_secret', help='The Twitter Access Token Secret.')
+    parser.add_argument('-ck', '--consumer_key', help='The Twitter Consumer Key.')
+    parser.add_argument('-cs', '--consumer_secret', help='The Twitter Consumer Secret.')
+
+    return parser.parse_args()
+
+##  parse_arguments()
+
+
+def main():
+
+    args = parse_arguments()
+
+    # When using twitter stream you must authorize.
+    stream = TwitterStream(auth=OAuth(args.token, args.token_secret, args.consumer_key, args.consumer_secret))
 
     # Iterate over the sample stream.
     tweet_iter = stream.statuses.sample()
@@ -31,3 +45,8 @@ def main(args=sys.argv[1:]):
         # or data message.
         if tweet.get('text'):
             printNicely(tweet['text'])
+
+##  main()
+
+if __name__ == '__main__':
+    main()