BUG: handle IncompleteRead exceptions

This should handle both cases of incomplete read catched by requests, or
catched by tweepy.

This resolves tweepy#237, resolves tweepy#448, resolves tweepy#536, resolves tweepy#650,
resolves tweepy#691, resolves tweepy#798.

Similar to tweepy#498.

tweepy/streaming.py

@@ -14,6 +14,7 @@
 from time import sleep
 
 import six
+from six.moves import http_client as httplib
 
 import ssl
 
@@ -159,9 +160,8 @@ def read_len(self, length):
         """Read the data stream until `length` characters been received
 
         :param length: The number of characters to read from the stream(int)
-        :return: The str of the data read with `length` characters, or None
-            if the connection was closed before (str in python 2, bytes in
-            python 3)
+        :return: The str of the data read with `length` characters (str in
+            python 2, bytes in python 3)
         """
         while not self._stream.closed:
             if len(self._buffer) >= length:
@@ -170,15 +170,16 @@ def read_len(self, length):
             self._buffer += self._stream.read(read_len)
 
         # if stream is closed before enough characters were received
-        return None
+        raise httplib.IncompleteRead('%d bytes read, %d more expected'
+            % (len(self._buffer), length - len(self._buffer)))
 
     def read_line(self, sep=six.b('\n')):
         """Read the data stream until a given separator is found (default \n)
 
         :param sep: Separator to read until. Must by of the bytes type (str in
             python 2, bytes in python 3)
-        :return: The str of the data read until sep, or None if the connection
-            was closed before (str in python 2, bytes in python 3)
+        :return: The str of the data read until sep (str in python 2, bytes in
+            python 3)
         """
         start = 0
         while not self._stream.closed:
@@ -190,7 +191,7 @@ def read_line(self, sep=six.b('\n')):
             self._buffer += self._stream.read(self._chunk_size)
 
         # if stream is closed before a seperator was received
-        return None
+        raise httplib.IncompleteRead('"%s" sep was not received' % (str(sep),))
 
     def _pop(self, length):
         r = self._buffer[:length]
@@ -275,7 +276,7 @@ def _run(self):
                     self.snooze_time = self.snooze_time_step
                     self.listener.on_connect()
                     self._read_loop(resp)
-            except (Timeout, ssl.SSLError) as exc:
+            except (Timeout, ssl.SSLError, httplib.HTTPError) as exc:
                 # This is still necessary, as a SSLError can actually be
                 # thrown when using Requests
                 # If it's not time out treat it like any other exception
@@ -324,10 +325,7 @@ def _read_loop(self, resp):
         while self.running and not resp.raw.closed:
             length = 0
             while not resp.raw.closed:
-                line = buf.read_line()
-                if line is None:  # the stream is closed
-                    break
-                line = line.strip()
+                line = buf.read_line().strip()
                 if not line:
                     self.listener.keep_alive()  # keep-alive new lines are expected
                 elif line.isdigit():
@@ -337,8 +335,6 @@ def _read_loop(self, resp):
                     raise TweepError('Expecting length, unexpected value found')
 
             next_status_obj = buf.read_len(length)
-            if next_status_obj is None:  # the stream is closed
-                break
             if self.running:
                 self._data(next_status_obj)