Add capacity for self-closing messages

Author: blast-hardcheese

tests/v2/datagrams.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
 from typing import (
     Any,
+    Literal,
     NewType,
     TypeAlias,
 )
@@ -30,13 +31,17 @@ class ValueSet:
     create_alias: StreamAlias | None = None
     stream_alias: StreamAlias | None = None
     payload: Datagram | None = None
+    stream_closed: Literal[True] | None = None
 
 
 @dataclass(frozen=True)
 class FromClient:
     handshake_request: tuple[ClientId, ServerId, SessionId] | ValueSet | None = None
-    stream_open: tuple[ClientId, ServerId, str, str, StreamId] | ValueSet | None = None
+    stream_open: (
+        tuple[ClientId, ServerId, str, str, StreamId, Datagram] | ValueSet | None
+    ) = None
     stream_frame: tuple[ClientId, ServerId, int, int, Datagram] | ValueSet | None = None
+    stream_closed: Literal[True] | None = None
 
 
 @dataclass(frozen=True)
@@ -75,7 +80,12 @@ def decode_FromClient(datagram: dict[str, Any]) -> FromClient:
                 datagram["serviceName"],
                 datagram["procedureName"],
                 StreamId(datagram["streamId"]),
+                datagram["payload"],
+            ),
+            stream_closed=(
+                datagram["controlFlags"] & 0b01000 > 0  # STREAM_CLOSED_BIT
             )
+            or None,
         )
     elif datagram:
         return FromClient(

tests/v2/interpreter.py

@@ -73,7 +73,7 @@ def from_client_interpreter(
                 payload=_build_handshake_resp(session_id),
             )
         elif isinstance(received.stream_open, tuple):
-            (from_, to, service_name, procedure_name, stream_id) = received.stream_open
+            (from_, to, service_name, procedure_name, stream_id, payload) = received.stream_open
             assert isinstance(expected.stream_open, ValueSet), "Expected ValueSet 2"
             assert expected.stream_open.from_ == to, (
                 f"Expected {expected.stream_open.from_} == {to}"
@@ -90,6 +90,13 @@ def from_client_interpreter(
             assert expected.stream_open.create_alias, (
                 "Expected create_alias to be a StreamAlias"
             )
+            if expected.stream_open.payload is not None and payload is not None:
+                assert expected.stream_open.payload == payload, (
+                    f"Expected {expected.stream_open.payload} == {payload}"
+                )
+            assert expected.stream_open.stream_closed or not received.stream_closed, (
+                    f"Are we self-closing? {expected.stream_open.stream_closed} or not {received.stream_closed}"
+                )
             # Do it all again because mypy can't infer correctly
             alias_mapping: tuple[ClientId, ServerId, StreamId] = (
                 ClientId(from_),