Add a StreamFatalError event

This event is sent when the stream is about to stop for good due to an
error (this event is NOT sent when the stream stops due to normal stream
termination).

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

RELEASE_NOTES.md

@@ -21,6 +21,8 @@
                 print("Stream started")
             case StreamStopped(delay, error):
                 print(f"Stream stopped, reason {error}, retry in {delay}")
+            case StreamFatalError(error):
+                print(f"Stream will stop because of a fatal error: {error}")
             case int() as output:
                 print(f"Received message: {output}")
     ```

src/frequenz/client/base/streaming.py

@@ -42,7 +42,15 @@ class StreamStopped:
     """The exception that caused the stream to stop, if any."""
 
 
-StreamEvent: TypeAlias = StreamStarted | StreamStopped
+@dataclass(frozen=True)
+class StreamFatalError:
+    """Event indicating that the stream has stopped due to an unrecoverable error."""
+
+    exception: Exception
+    """The exception that caused the stream to stop."""
+
+
+StreamEvent: TypeAlias = StreamStarted | StreamStopped | StreamFatalError
 """Type alias for the events that can be sent over the stream."""
 
 
@@ -83,6 +91,8 @@ def async_range() -> AsyncIterable[int]:
                     print("Stream started")
                 case StreamStopped(delay, error):
                     print(f"Stream stopped, reason {error}, retry in {delay}")
+                case StreamFatalError(error):
+                    print(f"Stream will stop because of a fatal error: {error}")
                 case int() as output:
                     print(f"Received message: {output}")
         ```
@@ -197,6 +207,8 @@ async def _run(self) -> None:
                     self._retry_strategy.get_progress(),
                     error_str,
                 )
+                if error is not None:
+                    await sender.send(StreamFatalError(error))
                 await self._channel.close()
                 break
             _logger.warning(

tests/streaming/test_grpc_stream_broadcaster.py

@@ -17,6 +17,7 @@
 from frequenz.client.base import retry, streaming
 from frequenz.client.base.streaming import (
     StreamEvent,
+    StreamFatalError,
     StreamStarted,
     StreamStopped,
 )
@@ -95,7 +96,7 @@ async def _split_message(
     events: list[StreamEvent] = []
     async for item in receiver:
         match item:
-            case StreamStarted() | StreamStopped() as item:
+            case StreamStarted() | StreamStopped() | StreamFatalError():
                 events.append(item)
             case str():
                 items.append(item)
@@ -343,5 +344,6 @@ async def test_messages_on_retry(
             ),
             StreamStarted(),
             StreamStopped(exception=mock_error(), retry_time=None),
+            StreamFatalError(mock_error()),
         ]
     ]