Fix: Make streams local, not global

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

RELEASE_NOTES.md

@@ -2,6 +2,7 @@
 
 ## Bug Fixes
 
-* Fix missing dependency in last release
+* Fix missing dependency in last release.
 * The `FakeClient.set_dispatches()` method now correctly updates `FakeService._last_id` which is used to generate unique dispatch IDs.
+* Fix that streams were globally shared between all clients.
 

src/frequenz/client/dispatch/_client.py

@@ -55,11 +55,6 @@
 class Client(BaseApiClient[dispatch_pb2_grpc.MicrogridDispatchServiceStub]):
     """Dispatch API client."""
 
-    streams: dict[
-        int, GrpcStreamBroadcaster[StreamMicrogridDispatchesResponse, DispatchEvent]
-    ] = {}
-    """A dictionary of streamers, keyed by microgrid_id."""
-
     def __init__(
         self,
         *,
@@ -91,6 +86,10 @@ def __init__(
             ),
         )
         self._metadata = (("key", key),)
+        self._streams: dict[
+            int, GrpcStreamBroadcaster[StreamMicrogridDispatchesResponse, DispatchEvent]
+        ] = {}
+        """A dictionary of streamers, keyed by microgrid_id."""
 
     @property
     def stub(self) -> dispatch_pb2_grpc.MicrogridDispatchServiceAsyncStub:
@@ -225,11 +224,11 @@ def _get_stream(
         self, microgrid_id: int
     ) -> GrpcStreamBroadcaster[StreamMicrogridDispatchesResponse, DispatchEvent]:
         """Get an instance to the streaming helper."""
-        broadcaster = self.streams.get(microgrid_id)
+        broadcaster = self._streams.get(microgrid_id)
         # pylint: disable=protected-access
         if broadcaster is not None and broadcaster._channel.is_closed:
             # pylint: enable=protected-access
-            del self.streams[microgrid_id]
+            del self._streams[microgrid_id]
             broadcaster = None
         if broadcaster is None:
             request = StreamMicrogridDispatchesRequest(microgrid_id=microgrid_id)
@@ -244,7 +243,7 @@ def _get_stream(
                 transform=DispatchEvent.from_protobuf,
                 retry_strategy=LinearBackoff(interval=1, limit=0),
             )
-            self.streams[microgrid_id] = broadcaster
+            self._streams[microgrid_id] = broadcaster
 
         return broadcaster