Add StopSendingReceived QUIC event

Section 2.4 of RFC9000 says that an application protocol can
request to be informed of state changes on streams. Even though
the state machines described in section 3 don't explicitly have
state transitions regarding STOP_SENDING, an application
protocol sometimes needs to know the reception of STOP_SENDING.

docs/quic.rst

@@ -44,6 +44,9 @@ Events
     .. autoclass:: PingAcknowledged
         :members:
 
+    .. autoclass:: StopSendingReceived
+        :members:
+
     .. autoclass:: StreamDataReceived
         :members:
 

src/aioquic/quic/connection.py

@@ -2070,6 +2070,10 @@ def _handle_stop_sending_frame(
         stream = self._get_or_create_stream(frame_type, stream_id)
         stream.sender.reset(error_code=QuicErrorCode.NO_ERROR)
 
+        self._events.append(
+            events.StopSendingReceived(error_code=error_code, stream_id=stream_id)
+        )
+
     def _handle_stream_frame(
         self, context: QuicReceiveContext, frame_type: int, buf: Buffer
     ) -> None:

src/aioquic/quic/events.py

@@ -82,6 +82,20 @@ class ProtocolNegotiated(QuicEvent):
     "The protocol which was negotiated using ALPN, or `None`."
 
 
+@dataclass
+class StopSendingReceived(QuicEvent):
+    """
+    The StopSendingReceived event is fired when the remote peer requests
+    stopping data transmission on a stream.
+    """
+
+    error_code: int
+    "The error code that was sent from the peer."
+
+    stream_id: int
+    "The ID of the stream that the peer requested stopping data transmission."
+
+
 @dataclass
 class StreamDataReceived(QuicEvent):
     """

tests/test_connection.py

@@ -1922,6 +1922,19 @@ def test_handle_stop_sending_frame(self):
                 Buffer(data=b"\x00\x11"),
             )
 
+            # check events
+            self.assertEqual(type(client.next_event()), events.ProtocolNegotiated)
+            self.assertEqual(type(client.next_event()), events.HandshakeCompleted)
+            for i in range(7):
+                self.assertEqual(type(client.next_event()), events.ConnectionIdIssued)
+
+            event = client.next_event()
+            self.assertEqual(type(event), events.StopSendingReceived)
+            self.assertEqual(event.stream_id, 0)
+            self.assertEqual(event.error_code, 0x11)
+
+            self.assertIsNone(client.next_event())
+
     def test_handle_stop_sending_frame_receive_only(self):
         with client_and_server() as (client, server):
             # server creates unidirectional stream 3