convert events.StreamReset into a dataclass

Author: mhils

src/h2/events.py

@@ -431,6 +431,7 @@ def __repr__(self) -> str:
  return f"<StreamEnded stream_id:{self.stream_id}>"
 
 
+@dataclass(**kw_only)
 class StreamReset(Event):
  """
  The StreamReset event is fired in two situations. The first is when the
@@ -442,16 +443,20 @@ class StreamReset(Event):
  This event is now fired when h2 automatically resets a stream.
  """
 
- def __init__(self) -> None:
- #: The Stream ID of the stream that was reset.
- self.stream_id: int | None = None
+ stream_id: int
+ """
+ The Stream ID of the stream that was reset.
+ """
 
- #: The error code given. Either one of :class:`ErrorCodes
- #: <h2.errors.ErrorCodes>` or ``int``
- self.error_code: ErrorCodes | int | None = None
+ error_code: ErrorCodes | int = _LAZY_INIT
+ """
+ The error code given.
+ """
 
- #: Whether the remote peer sent a RST_STREAM or we did.
- self.remote_reset = True
+ remote_reset: bool = True
+ """
+ Whether the remote peer sent a RST_STREAM or we did.
+ """
 
  def __repr__(self) -> str:
  return f"<StreamReset stream_id:{self.stream_id}, error_code:{self.error_code!s}, remote_reset:{self.remote_reset}>"

src/h2/stream.py

@@ -257,9 +257,7 @@ def stream_reset(self, previous_state: StreamState) -> list[Event]:
  Fired when a stream is forcefully reset.
  """
  self.stream_closed_by = StreamClosedBy.RECV_RST_STREAM
- event = StreamReset()
- event.stream_id = self.stream_id
- return [event]
+ return [StreamReset(stream_id=self.stream_id)]
 
  def send_new_pushed_stream(self, previous_state: StreamState) -> list[Event]:
  """
@@ -342,12 +340,13 @@ def reset_stream_on_error(self, previous_state: StreamState) -> list[Event]:
  self.stream_closed_by = StreamClosedBy.SEND_RST_STREAM
 
  error = StreamClosedError(self.stream_id)
-
- event = StreamReset()
- event.stream_id = self.stream_id
- event.error_code = ErrorCodes.STREAM_CLOSED
- event.remote_reset = False
- error._events = [event]
+ error._events = [
+ StreamReset(
+ stream_id=self.stream_id,
+ error_code=ErrorCodes.STREAM_CLOSED,
+ remote_reset=False,
+ ),
+ ]
  raise error
 
  def recv_on_closed_stream(self, previous_state: StreamState) -> list[Event]:
@@ -1160,13 +1159,14 @@ def receive_window_update(self, increment: int) -> tuple[list[Frame], list[Event
  except FlowControlError:
  # Ok, this is bad. We're going to need to perform a local
  # reset.
- event = StreamReset()
- event.stream_id = self.stream_id
- event.error_code = ErrorCodes.FLOW_CONTROL_ERROR
- event.remote_reset = False
-
- events = [event]
- frames = self.reset_stream(event.error_code)
+ events = [
+ StreamReset(
+ stream_id=self.stream_id,
+ error_code=ErrorCodes.FLOW_CONTROL_ERROR,
+ remote_reset=False,
+ ),
+ ]
+ frames = self.reset_stream(ErrorCodes.FLOW_CONTROL_ERROR)
 
  return frames, events
 

tests/test_events.py

@@ -246,10 +246,11 @@ def test_streamreset_repr(self) -> None:
  """
  StreamEnded has a useful debug representation.
  """
- e = h2.events.StreamReset()
- e.stream_id = 919
- e.error_code = h2.errors.ErrorCodes.ENHANCE_YOUR_CALM
- e.remote_reset = False
+ e = h2.events.StreamReset(
+ stream_id=919,
+ error_code=h2.errors.ErrorCodes.ENHANCE_YOUR_CALM,
+ remote_reset=False,
+ )
 
  if sys.version_info >= (3, 11):
  assert repr(e) == (