From 5f11467f2749fead5de6700401674812a62b05f3 Mon Sep 17 00:00:00 2001
From: Manuel Cortez <manuel@manuelcortez.net>
Date: Fri, 2 Jul 2021 09:52:21 -0500
Subject: [PATCH] Switched threads to our own facilities and attempts to
 improve thread management for streaming endpoints

---
 src/controller/mainController.py | 2 ++
 src/sessions/twitter/session.py  | 9 +++++++--
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/src/controller/mainController.py b/src/controller/mainController.py
index 9e75b459..d6021a78 100644
--- a/src/controller/mainController.py
+++ b/src/controller/mainController.py
@@ -659,6 +659,8 @@ class Controller(object):
             sessions.sessions[item].sound.cleaner.cancel()
             log.debug("Saving database for " +    sessions.sessions[item].session_id)
             sessions.sessions[item].save_persistent_data()
+            log.debug("Disconnecting streaming endpoint for session" +    sessions.sessions[item].session_id)
+            sessions.sessions[item].stop_streaming()
         if system == "Windows":
             self.systrayIcon.RemoveIcon()
             pidpath = os.path.join(os.getenv("temp"), "{}.pid".format(application.name))
diff --git a/src/sessions/twitter/session.py b/src/sessions/twitter/session.py
index 0bee2fd7..e14916f1 100644
--- a/src/sessions/twitter/session.py
+++ b/src/sessions/twitter/session.py
@@ -506,8 +506,13 @@ class Session(base.baseSession):
 
     def start_streaming(self):
         self.stream_listener = streaming.StreamListener(twitter_api=self.twitter, user=self.db["user_name"], user_id=self.db["user_id"])
-        self.stream = tweepy.Stream(auth = self.auth, listener=self.stream_listener)
-        self.stream_thread = self.stream.filter(follow=self.stream_listener.users, is_async=True)
+        self.stream = streaming.Stream(auth = self.auth, listener=self.stream_listener, chunk_size=1025)
+        self.stream_thread = call_threaded(self.stream.filter, follow=self.stream_listener.users, stall_warnings=True)
+
+    def stop_streaming(self):
+        if hasattr(self, "stream_thread"):
+            self.stream_thread.join()
+        log.debug("Stopping Streaming Endpoint...")
 
     def handle_new_status(self, status, user):
         """ Handles a new status present in the Streaming API. """