Allow base-client to be updated to 0.11.0

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

Author: Marenz

RELEASE_NOTES.md

@@ -6,11 +6,13 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+* When receiving streaming data for components, you now need to handle the receiving of the types `StreamStarted`, `StreamRetrying`, `StreamFatalError`.
+
+    If you don't care about the new events and just want the old behavior your can always use `Receiver.filter()` to ignore them, for example: `client.meter_data().filter(lambda m: isinstance(m, MeterData))`.
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* Using the latest streaming client, when using `stream_sensor_data()` you will now get stream notification events, such as `StreamStarted`, `StreamRetrying` and `StreamFatalError`, which can be used to monitor the state of the stream.
 
 ## Bug Fixes
 

pyproject.toml

@@ -38,7 +38,7 @@ requires-python = ">= 3.11, < 4"
 dependencies = [
   "frequenz-api-microgrid >= 0.15.3, < 0.16.0",
   "frequenz-channels >= 1.0.0-rc1, < 2.0.0",
-  "frequenz-client-base >= 0.8.0, < 0.12.0",
+  "frequenz-client-base >= 0.11.0, < 0.12.0",
   "grpcio >= 1.59.0, < 2",
   "protobuf >= 4.21.6, < 7",
   "timezonefinder >= 6.2.0, < 7",

tests/test_client.py

@@ -17,6 +17,7 @@
 from frequenz.api.common import components_pb2, metrics_pb2
 from frequenz.api.microgrid import grid_pb2, inverter_pb2, microgrid_pb2, sensor_pb2
 from frequenz.client.base import conversion, retry
+from frequenz.client.base.streaming import StreamRetrying, StreamStarted
 from google.protobuf.empty_pb2 import Empty
 
 from frequenz.client.microgrid import (
@@ -688,6 +689,7 @@ async def stream_data(
 
     client.mock_stub.StreamComponentData.side_effect = stream_data
     receiver = await getattr(client, method)(component_id)
+    assert isinstance(await receiver.receive(), StreamStarted)
     latest = await receiver.receive()
     assert isinstance(latest, component_class)
     assert latest.component_id == component_id
@@ -736,14 +738,28 @@ async def stream_data(
 
     client.mock_stub.StreamComponentData.side_effect = stream_data
     receiver = await getattr(client, method)(component_id)
+    assert isinstance(await receiver.receive(), StreamStarted)
+    assert isinstance(await receiver.receive(), StreamRetrying)
+    assert isinstance(await receiver.receive(), StreamStarted)
+
     latest = await receiver.receive()
     assert isinstance(latest, component_class)
     assert latest.component_id == component_id
 
+    assert isinstance(await receiver.receive(), StreamRetrying)
+    assert isinstance(await receiver.receive(), StreamStarted)
+    assert isinstance(await receiver.receive(), StreamRetrying)
+    assert isinstance(await receiver.receive(), StreamStarted)
+
     latest = await receiver.receive()
     assert isinstance(latest, component_class)
     assert latest.component_id == component_id
 
+    assert isinstance(await receiver.receive(), StreamRetrying)
+    assert isinstance(await receiver.receive(), StreamStarted)
+    assert isinstance(await receiver.receive(), StreamRetrying)
+    assert isinstance(await receiver.receive(), StreamStarted)
+
     latest = await receiver.receive()
     assert isinstance(latest, component_class)
     assert latest.component_id == component_id
@@ -947,6 +963,8 @@ async def stream_data_impl(
     receiver = client.stream_sensor_data(
         SensorId(sensor201.id), [SensorMetric.TEMPERATURE]
     )
+
+    assert isinstance(await receiver.receive(), StreamStarted)
     sample = await receiver.receive()
 
     assert isinstance(sample, SensorDataSamples)
@@ -997,6 +1015,8 @@ async def stream_data_impl(
 
     client.mock_stub.StreamComponentData.side_effect = stream_data_impl
     receiver = client.stream_sensor_data(SensorId(sensor201.id))
+
+    assert isinstance(await receiver.receive(), StreamStarted)
     sample = await receiver.receive()
 
     assert isinstance(sample, SensorDataSamples)
@@ -1046,7 +1066,12 @@ async def stream_data_error_impl(
         receiver = client.stream_sensor_data(
             SensorId(sensor201.id), [SensorMetric.TEMPERATURE]
         )
-        sample = await receiver.receive()  # Should succeed after retries
+        assert isinstance(await receiver.receive(), StreamStarted)
+        assert isinstance(await receiver.receive(), StreamRetrying)
+        assert isinstance(await receiver.receive(), StreamStarted)
+        assert isinstance(await receiver.receive(), StreamRetrying)
+        assert isinstance(await receiver.receive(), StreamStarted)
+        sample = await receiver.receive()  # Get the actual sample
 
     assert isinstance(sample, SensorDataSamples)
     assert int(sample.sensor_id) == sensor201.id