Update services/registry.py

services/registry.py

@@ -107,14 +107,11 @@ class RepoConnection:
                             if not self._running or not self._metrics_active:
                                 break
                             decoded = line.decode("utf-8").strip()
-                            logger.info(f"Started connections for {decoded}")
-                            if decoded.startswith("data:"):
-                                data_str = decoded[5:].strip()
-                                try:
-                                    data = json.loads(data_str)
-                                    storage.add_metric(self.repo_id, data)
-                                except json.JSONDecodeError:
-                                    continue
+                            field, value = parse_sse_line(decoded)
+                            if field == 'data':
+                                data = json.loads(value)
+                                logger.info(f"Started connections for {data}")
+                                storage.add_metric(self.repo_id, data)
             except asyncio.CancelledError:
                 break
             except Exception as e:
@@ -127,6 +124,13 @@ class RepoConnection:
         delay = min(1.0 * (2 ** self._retry_count), 30.0)
         self._retry_count = min(self._retry_count + 1, self._max_retries)
         return delay
+        
+    async def parse_sse_line(line):
+        """Parse a single SSE line into field and value."""
+        if ':' in line:
+            field, _, value = line.partition(':')
+            return field, value.lstrip(' ')
+        return line, ''
 
 
 class ConnectionRegistry: