Merge pull request thisbejim#112 from thisbejim/simplify-streams

fix: simplify stream re-auth

pyrebase/pyrebase.py

@@ -249,7 +249,8 @@ def build_request_url(self, token):
     def build_headers(self, token=None):
         headers = {"content-type": "application/json; charset=UTF-8"}
         if not token and self.credentials:
-            headers['Authorization'] = 'Bearer ' + self.credentials.get_access_token().access_token
+            access_token = self.credentials.get_access_token().access_token
+            headers['Authorization'] = 'Bearer ' + access_token
         return headers
 
     def get(self, token=None):
@@ -443,6 +444,7 @@ def raise_detailed_error(request_object):
         request_object.raise_for_status()
     except HTTPError as e:
         # raise detailed error message
+        # TODO: Check if we get a { "error" : "Permission denied." } and handle automatically
         raise HTTPError(e, request_object.text)
 
 
@@ -553,9 +555,8 @@ def start(self):
     def start_stream(self):
         self.sse = ClosableSSEClient(self.url, session=self.make_session(), build_headers=self.build_headers)
         for msg in self.sse:
-            msg_data = json.loads(msg.data)
-            # don't return initial data
-            if msg_data:
+            if msg:
+                msg_data = json.loads(msg.data)
                 msg_data["event"] = msg.event
                 if self.stream_id:
                     msg_data["stream_id"] = self.stream_id

sseclient/sseclient.py

@@ -53,22 +53,6 @@ def _connect(self):
         # attribute on Events like the Javascript spec requires.
         self.resp.raise_for_status()
 
-        if not self.start_time:
-            self.thread = threading.Thread(target=self.connect)
-            self.thread.start()
-
-    def connect(self):
-        while self.running:
-            if self.start_time:
-                half_hour_in_seconds = 1800
-                # if it has been a half hour since we started the stream, restart
-                if time.time() - self.start_time > half_hour_in_seconds:
-                    self.start_time = time.time()
-                    self._connect()
-            else:
-                self.start_time = time.time()
-            time.sleep(0.1)
-
     def _event_complete(self):
         return re.search(end_of_field, self.buf) is not None
 
@@ -97,6 +81,13 @@ def __next__(self):
         self.buf = tail
         msg = Event.parse(head)
 
+        if msg.data == "credential is no longer valid":
+            self._connect()
+            return None
+
+        if msg.data == 'null':
+            return None
+
         # If the server requests a specific retry delay, we need to honor it.
         if msg.retry:
             self.retry = msg.retry