Merge pull request #559 from pvanderlinden/streaming-improvements

Streaming improvements

tweepy/streaming.py

@@ -79,6 +79,10 @@ def on_data(self, raw_data):
         else:
             logging.error("Unknown message type: " + str(raw_data))
 
+    def keep_alive(self):
+        """Called when a keep-alive arrived"""
+        return
+
     def on_status(self, status):
         """Called when a new status arrives"""
         return
@@ -200,13 +204,17 @@ def __init__(self, auth, listener, **options):
         self.verify = options.get("verify", True)
 
         self.api = API()
-        self.session = requests.Session()
-        self.session.headers = options.get("headers") or {}
-        self.session.params = None
+        self.headers = options.get("headers") or {}
+        self.new_session()
         self.body = None
         self.retry_time = self.retry_time_start
         self.snooze_time = self.snooze_time_step
 
+    def new_session(self):
+        self.session = requests.Session()
+        self.session.headers = self.headers
+        self.session.params = None
+
     def _run(self):
         # Authenticate
         url = "https://%s%s" % (self.host, self.url)
@@ -270,12 +278,12 @@ def _run(self):
         if resp:
             resp.close()
 
-        self.session = requests.Session()
+        self.new_session()
 
         if exception:
             # call a handler first so that the exception can be logged.
             self.listener.on_exception(exception)
-            raise exception
+            raise
 
     def _data(self, data):
         if self.listener.on_data(data) is False:
@@ -289,7 +297,7 @@ def _read_loop(self, resp):
             while True:
                 line = buf.read_line().strip()
                 if not line:
-                    pass  # keep-alive new lines are expected
+                    self.listener.keep_alive()  # keep-alive new lines are expected
                 elif line.isdigit():
                     length = int(line)
                     break