Fix stream() not working with base-client 0.11

And fix test to catch it next time.

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

Author: Marenz

src/frequenz/client/dispatch/_client.py

@@ -242,9 +242,7 @@ def _get_stream(
     ) -> GrpcStreamBroadcaster[StreamMicrogridDispatchesResponse, DispatchEvent]:
         """Get an instance to the streaming helper."""
         broadcaster = self._streams.get(microgrid_id)
-        # pylint: disable=protected-access
-        if broadcaster is not None and broadcaster._channel.is_closed:
-            # pylint: enable=protected-access
+        if broadcaster is not None and not broadcaster.is_running:
             del self._streams[microgrid_id]
             broadcaster = None
         if broadcaster is None:

tests/test_client.py

@@ -309,6 +309,9 @@ async def test_dispatch_stream(client: FakeClient, sample: Dispatch) -> None:
 
     stream = client.stream(microgrid_id)
 
+    # Call function again to test behavior if stream already exists
+    _ = client.stream(microgrid_id)
+
     async def expect(dispatch: Dispatch, event: Event) -> None:
         message = await stream.receive()
         assert message.dispatch == dispatch