Port RUDP reliability core to Python
This commit is contained in:
Meta Vens
parent
c71a1bc5da
commit
343802dacf
4 changed files with 398 additions and 12 deletions
|
|
@ -26,9 +26,9 @@ CultNet owns cross-runtime transport semantics:
|
|||
| --- | --- | --- | --- |
|
||||
| C# `GameCult.Networking` | LiteNetLib UDP `NetManager` / `NetPeer`; sends legacy union messages and schema-v0 messages with `DeliveryMethod.ReliableOrdered` | LiteNetLib connection requests and app-level peer/catalog surfaces | C# owns the production-shaped realtime semantics |
|
||||
| C# interop peer | TCP stream with 4-byte length-prefixed MessagePack frames | UDP multicast probe/announce | Test harness only |
|
||||
| TypeScript `cultnet-ts` | `CultNetPeer` over any Node `Duplex`; interop uses TCP | UDP multicast probe/announce in the interop peer | Byte-stream abstraction; no UDP reliability owner |
|
||||
| Rust `cultnet-rs` | Interop example uses TCP length-prefixed MessagePack frames | UDP multicast probe/announce | Example harness; no UDP reliability owner |
|
||||
| Python `cultcache-py` | TCP sockets with 4-byte length-prefixed MessagePack frames for local CultMesh/CultNet server and client | Endpoint lists and CultMesh peer/Verse catalogs | Local framed service body; no UDP reliability owner |
|
||||
| TypeScript `cultnet-ts` | `CultNetPeer` over any Node `Duplex`; interop uses TCP | UDP multicast probe/announce in the interop peer | Byte-stream abstraction plus socket-free RUDP reliability owner; no UDP socket binding yet |
|
||||
| Rust `cultnet-rs` | Interop example uses TCP length-prefixed MessagePack frames | UDP multicast probe/announce | Example harness plus socket-free RUDP reliability owner; no UDP socket binding yet |
|
||||
| Python `cultcache-py` | TCP sockets with 4-byte length-prefixed MessagePack frames for local CultMesh/CultNet server and client | Endpoint lists and CultMesh peer/Verse catalogs | Local framed service body plus socket-free RUDP reliability owner; no UDP socket binding yet |
|
||||
| Kotlin `cultmesh-kotlin` | Minimal WebSocket-like lane over TCP socket | None found in the live package | Thin client surface |
|
||||
|
||||
The live split is therefore: payload language is converging, transport language
|
||||
|
|
@ -160,15 +160,15 @@ Current progress:
|
|||
the RUDP runtime yet; it is the binary packet language the runtimes must
|
||||
converge on before resend loops, windows, and timeout behavior are allowed to
|
||||
claim parity.
|
||||
- TypeScript, C#, and Rust now share the first deterministic RUDP reliability state
|
||||
machine: connect/accept handshake, packet-level ack/ack-mask accounting,
|
||||
reliable resend scheduling, duplicate suppression, and reliable ordered
|
||||
channel delivery. It is in-memory and socket-free so the behavior can be
|
||||
ported before any runtime binds it to UDP I/O.
|
||||
- TypeScript, C#, Rust, and Python now share the first deterministic RUDP
|
||||
reliability state machine: connect/accept handshake, packet-level
|
||||
ack/ack-mask accounting, reliable resend scheduling, duplicate suppression,
|
||||
and reliable ordered channel delivery. It is in-memory and socket-free so the
|
||||
behavior can be tested before any runtime binds it to UDP I/O.
|
||||
- The remaining parity work is to add the equivalent port to Kotlin, deepen
|
||||
Python's server-side use of its port, and move each runtime's existing
|
||||
TCP/LiteNetLib/WebSocket bodies behind those ports, port the RUDP reliability
|
||||
state machine into Python, then bind it to UDP sockets.
|
||||
TCP/LiteNetLib/WebSocket bodies behind those ports, then bind the shared RUDP
|
||||
reliability state machine to UDP sockets.
|
||||
|
||||
## RUDP Packet Contract V0
|
||||
|
||||
|
|
|
|||
|
|
@ -66,8 +66,13 @@ from .shard_replication import (
|
|||
from .snapshot import CultNetRawDocumentRecord, CultNetRawSnapshotResponse
|
||||
from .subscription import CultNetDatabaseChange
|
||||
from .transport import (
|
||||
CultNetRudpDeliveredFrame,
|
||||
CultNetRudpPacket,
|
||||
CultNetRudpPacketType,
|
||||
CultNetRudpReceiveResult,
|
||||
CultNetRudpSendOptions,
|
||||
CultNetRudpSession,
|
||||
CultNetRudpSessionOptions,
|
||||
CultNetTransportFrame,
|
||||
CultNetTransportStats,
|
||||
TcpFramedTransportConnection,
|
||||
|
|
@ -123,8 +128,13 @@ __all__ = [
|
|||
"CultNetSimulationConsensusOptions",
|
||||
"CultNetSimulationObservation",
|
||||
"CultNetSimulationObservationHub",
|
||||
"CultNetRudpDeliveredFrame",
|
||||
"CultNetRudpPacket",
|
||||
"CultNetRudpPacketType",
|
||||
"CultNetRudpReceiveResult",
|
||||
"CultNetRudpSendOptions",
|
||||
"CultNetRudpSession",
|
||||
"CultNetRudpSessionOptions",
|
||||
"CultNetTransportFrame",
|
||||
"CultNetTransportStats",
|
||||
"TcpFramedTransportConnection",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from typing import Any, BinaryIO
|
||||
|
||||
|
|
@ -90,6 +90,40 @@ class CultNetRudpPacket:
|
|||
payload: bytes = b""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CultNetRudpDeliveredFrame:
|
||||
channel_id: str
|
||||
payload: bytes
|
||||
sequence: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CultNetRudpReceiveResult:
|
||||
delivered: tuple[CultNetRudpDeliveredFrame, ...] = field(default_factory=tuple)
|
||||
reply: CultNetRudpPacket | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CultNetRudpSessionOptions:
|
||||
connection_id: int
|
||||
initial_sequence: int = 1
|
||||
resend_delay_ms: int = 250
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CultNetRudpSendOptions:
|
||||
reliable: bool = False
|
||||
ordered: bool = False
|
||||
sequenced: bool = False
|
||||
now_ms: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class _PendingReliablePacket:
|
||||
packet: CultNetRudpPacket
|
||||
last_sent_at_ms: int
|
||||
|
||||
|
||||
_RUDP_PACKET_TYPE_TO_CODE = {
|
||||
CultNetRudpPacketType.CONNECT: 1,
|
||||
CultNetRudpPacketType.ACCEPT: 2,
|
||||
|
|
@ -145,6 +179,230 @@ def create_rudp_transport_profile(
|
|||
}
|
||||
|
||||
|
||||
class CultNetRudpSession:
|
||||
def __init__(self, options: CultNetRudpSessionOptions) -> None:
|
||||
self.connection_id = _uint32(options.connection_id, "connection_id")
|
||||
self.resend_delay_ms = options.resend_delay_ms
|
||||
self._next_sequence = _uint32(options.initial_sequence, "initial_sequence")
|
||||
self._connected = False
|
||||
self._highest_received_sequence: int | None = None
|
||||
self._received_sequences: set[int] = set()
|
||||
self._pending_reliable: dict[int, _PendingReliablePacket] = {}
|
||||
self._ordered_next_sequence_by_channel: dict[str, int] = {}
|
||||
self._ordered_buffers: dict[str, dict[int, CultNetRudpDeliveredFrame]] = {}
|
||||
|
||||
@property
|
||||
def connected(self) -> bool:
|
||||
return self._connected
|
||||
|
||||
@property
|
||||
def pending_reliable_sequences(self) -> tuple[int, ...]:
|
||||
return tuple(sorted(self._pending_reliable))
|
||||
|
||||
def create_connect(self, now_ms: int = 0, payload: bytes = b"") -> CultNetRudpPacket:
|
||||
packet = self._create_packet(
|
||||
CultNetRudpPacketType.CONNECT,
|
||||
"control",
|
||||
payload,
|
||||
reliable=True,
|
||||
ordered=True,
|
||||
)
|
||||
self._track_reliable(packet, now_ms)
|
||||
return packet
|
||||
|
||||
def accept_connect(
|
||||
self,
|
||||
packet: CultNetRudpPacket,
|
||||
now_ms: int = 0,
|
||||
payload: bytes = b"",
|
||||
) -> CultNetRudpPacket:
|
||||
self._require_connection(packet)
|
||||
if packet.packet_type != CultNetRudpPacketType.CONNECT:
|
||||
raise ValueError(f"Expected RUDP connect packet, got {packet.packet_type.value}")
|
||||
|
||||
self._remember_received(packet.sequence)
|
||||
self._connected = True
|
||||
response = self._create_packet(
|
||||
CultNetRudpPacketType.ACCEPT,
|
||||
"control",
|
||||
payload,
|
||||
reliable=True,
|
||||
ordered=True,
|
||||
)
|
||||
self._track_reliable(response, now_ms)
|
||||
return response
|
||||
|
||||
def send(
|
||||
self,
|
||||
channel_id: str,
|
||||
payload: bytes,
|
||||
options: CultNetRudpSendOptions | None = None,
|
||||
) -> CultNetRudpPacket:
|
||||
if not self._connected:
|
||||
raise ValueError("Cannot send RUDP data before the session is connected")
|
||||
|
||||
options = options or CultNetRudpSendOptions()
|
||||
packet = self._create_packet(
|
||||
CultNetRudpPacketType.DATA,
|
||||
channel_id,
|
||||
payload,
|
||||
reliable=options.reliable,
|
||||
ordered=options.ordered,
|
||||
sequenced=options.sequenced,
|
||||
)
|
||||
if packet.reliable:
|
||||
self._track_reliable(packet, options.now_ms)
|
||||
return packet
|
||||
|
||||
def receive(self, packet: CultNetRudpPacket, now_ms: int = 0) -> CultNetRudpReceiveResult:
|
||||
del now_ms
|
||||
self._require_connection(packet)
|
||||
self._apply_acknowledgements(packet)
|
||||
|
||||
if packet.packet_type == CultNetRudpPacketType.ACCEPT:
|
||||
self._remember_received(packet.sequence)
|
||||
self._connected = True
|
||||
return CultNetRudpReceiveResult()
|
||||
|
||||
if packet.packet_type == CultNetRudpPacketType.PING:
|
||||
self._remember_received(packet.sequence)
|
||||
return CultNetRudpReceiveResult(
|
||||
reply=self._create_packet(CultNetRudpPacketType.PONG, "control", packet.payload)
|
||||
)
|
||||
|
||||
if packet.packet_type in (CultNetRudpPacketType.ACK, CultNetRudpPacketType.PONG):
|
||||
self._remember_received(packet.sequence)
|
||||
return CultNetRudpReceiveResult()
|
||||
|
||||
if packet.packet_type != CultNetRudpPacketType.DATA:
|
||||
return CultNetRudpReceiveResult()
|
||||
|
||||
duplicate = packet.sequence in self._received_sequences
|
||||
self._remember_received(packet.sequence)
|
||||
if duplicate:
|
||||
return CultNetRudpReceiveResult()
|
||||
|
||||
frame = CultNetRudpDeliveredFrame(
|
||||
channel_id=packet.channel_id,
|
||||
payload=bytes(packet.payload),
|
||||
sequence=packet.sequence,
|
||||
)
|
||||
if not packet.ordered:
|
||||
return CultNetRudpReceiveResult(delivered=(frame,))
|
||||
return CultNetRudpReceiveResult(delivered=tuple(self._deliver_ordered(frame)))
|
||||
|
||||
def create_ack(self) -> CultNetRudpPacket:
|
||||
return self._create_packet(CultNetRudpPacketType.ACK, "control", b"")
|
||||
|
||||
def due_resends(self, now_ms: int) -> tuple[CultNetRudpPacket, ...]:
|
||||
due: list[CultNetRudpPacket] = []
|
||||
for pending in self._pending_reliable.values():
|
||||
if now_ms - pending.last_sent_at_ms >= self.resend_delay_ms:
|
||||
pending.last_sent_at_ms = now_ms
|
||||
due.append(pending.packet)
|
||||
return tuple(sorted(due, key=lambda packet: packet.sequence))
|
||||
|
||||
def _create_packet(
|
||||
self,
|
||||
packet_type: CultNetRudpPacketType,
|
||||
channel_id: str,
|
||||
payload: bytes,
|
||||
*,
|
||||
reliable: bool = False,
|
||||
ordered: bool = False,
|
||||
sequenced: bool = False,
|
||||
) -> CultNetRudpPacket:
|
||||
sequence = self._next_sequence
|
||||
self._next_sequence = _uint32(self._next_sequence + 1, "sequence")
|
||||
ack, ack_mask = self._ack_state()
|
||||
return CultNetRudpPacket(
|
||||
packet_type=packet_type,
|
||||
connection_id=self.connection_id,
|
||||
sequence=sequence,
|
||||
ack=ack,
|
||||
ack_mask=ack_mask,
|
||||
channel_id=channel_id,
|
||||
reliable=reliable,
|
||||
ordered=ordered,
|
||||
sequenced=sequenced,
|
||||
payload=bytes(payload),
|
||||
)
|
||||
|
||||
def _track_reliable(self, packet: CultNetRudpPacket, now_ms: int) -> None:
|
||||
self._pending_reliable[packet.sequence] = _PendingReliablePacket(
|
||||
packet=CultNetRudpPacket(
|
||||
packet_type=packet.packet_type,
|
||||
connection_id=packet.connection_id,
|
||||
sequence=packet.sequence,
|
||||
ack=packet.ack,
|
||||
ack_mask=packet.ack_mask,
|
||||
channel_id=packet.channel_id,
|
||||
reliable=packet.reliable,
|
||||
ordered=packet.ordered,
|
||||
sequenced=packet.sequenced,
|
||||
fragment_id=packet.fragment_id,
|
||||
fragment_index=packet.fragment_index,
|
||||
fragment_count=packet.fragment_count,
|
||||
payload=bytes(packet.payload),
|
||||
),
|
||||
last_sent_at_ms=now_ms,
|
||||
)
|
||||
|
||||
def _apply_acknowledgements(self, packet: CultNetRudpPacket) -> None:
|
||||
self._pending_reliable.pop(packet.ack, None)
|
||||
for bit in range(32):
|
||||
if packet.ack_mask & (1 << bit):
|
||||
self._pending_reliable.pop(packet.ack - bit - 1, None)
|
||||
|
||||
def _remember_received(self, sequence: int) -> None:
|
||||
self._received_sequences.add(sequence)
|
||||
if self._highest_received_sequence is None or sequence > self._highest_received_sequence:
|
||||
self._highest_received_sequence = sequence
|
||||
|
||||
def _ack_state(self) -> tuple[int, int]:
|
||||
ack = self._highest_received_sequence or 0
|
||||
ack_mask = 0
|
||||
for bit in range(32):
|
||||
if ack > bit and (ack - bit - 1) in self._received_sequences:
|
||||
ack_mask |= 1 << bit
|
||||
return ack, ack_mask
|
||||
|
||||
def _deliver_ordered(self, frame: CultNetRudpDeliveredFrame) -> list[CultNetRudpDeliveredFrame]:
|
||||
next_sequence = self._ordered_next_sequence_by_channel.get(frame.channel_id)
|
||||
if next_sequence is None:
|
||||
self._ordered_next_sequence_by_channel[frame.channel_id] = frame.sequence + 1
|
||||
return [frame, *self._drain_ordered(frame.channel_id)]
|
||||
if frame.sequence < next_sequence:
|
||||
return []
|
||||
if frame.sequence > next_sequence:
|
||||
self._ordered_buffers.setdefault(frame.channel_id, {})[frame.sequence] = frame
|
||||
return []
|
||||
|
||||
self._ordered_next_sequence_by_channel[frame.channel_id] = next_sequence + 1
|
||||
return [frame, *self._drain_ordered(frame.channel_id)]
|
||||
|
||||
def _drain_ordered(self, channel_id: str) -> list[CultNetRudpDeliveredFrame]:
|
||||
delivered: list[CultNetRudpDeliveredFrame] = []
|
||||
buffer = self._ordered_buffers.get(channel_id)
|
||||
if buffer is None:
|
||||
return delivered
|
||||
|
||||
while True:
|
||||
next_sequence = self._ordered_next_sequence_by_channel[channel_id]
|
||||
frame = buffer.pop(next_sequence, None)
|
||||
if frame is None:
|
||||
break
|
||||
delivered.append(frame)
|
||||
self._ordered_next_sequence_by_channel[channel_id] = next_sequence + 1
|
||||
return delivered
|
||||
|
||||
def _require_connection(self, packet: CultNetRudpPacket) -> None:
|
||||
if packet.connection_id != self.connection_id:
|
||||
raise ValueError(
|
||||
f"RUDP packet belongs to connection {packet.connection_id}, expected {self.connection_id}"
|
||||
)
|
||||
|
||||
|
||||
def encode_rudp_packet(packet: CultNetRudpPacket) -> bytes:
|
||||
channel_id = packet.channel_id.encode("utf-8")
|
||||
if len(channel_id) > MAX_RUDP_CHANNEL_ID_BYTES:
|
||||
|
|
|
|||
|
|
@ -69,6 +69,9 @@ from cultnet_py import (
|
|||
CultNetSimulationObservationHub,
|
||||
CultNetRudpPacket,
|
||||
CultNetRudpPacketType,
|
||||
CultNetRudpSendOptions,
|
||||
CultNetRudpSession,
|
||||
CultNetRudpSessionOptions,
|
||||
TcpFramedTransportConnection,
|
||||
CultNetWitnessArtifactBundle,
|
||||
apply_raw_document_record,
|
||||
|
|
@ -505,7 +508,8 @@ class CultCacheTests(unittest.TestCase):
|
|||
self.assertEqual(validated.expires_at_utc, expires)
|
||||
self.assertEqual(validated.session_version, 7)
|
||||
|
||||
tampered = token[:-1] + ("A" if token[-1] != "A" else "B")
|
||||
tampered_payload = CultNetSecret.to_base64url(CultNetSecret.from_base64url(payload_b64) + b"tamper")
|
||||
tampered = f"{tampered_payload}.{signature_b64}"
|
||||
self.assertIsNone(CultNetSecret.try_validate_session_token(tampered, security))
|
||||
self.assertIsNone(CultNetSecret.try_validate_session_token(token, security, now_utc=expires))
|
||||
|
||||
|
|
@ -646,6 +650,120 @@ class CultCacheTests(unittest.TestCase):
|
|||
],
|
||||
)
|
||||
|
||||
def test_cultnet_rudp_session_handshake_acks_reliable_connect_and_accept_packets(self) -> None:
|
||||
client = CultNetRudpSession(
|
||||
CultNetRudpSessionOptions(connection_id=0x0A0B0C0D, initial_sequence=1, resend_delay_ms=50)
|
||||
)
|
||||
server = CultNetRudpSession(
|
||||
CultNetRudpSessionOptions(connection_id=0x0A0B0C0D, initial_sequence=100, resend_delay_ms=50)
|
||||
)
|
||||
|
||||
connect = client.create_connect(0, b"join")
|
||||
self.assertEqual(connect.packet_type, CultNetRudpPacketType.CONNECT)
|
||||
self.assertEqual(connect.sequence, 1)
|
||||
self.assertEqual(client.pending_reliable_sequences, (1,))
|
||||
|
||||
accept = server.accept_connect(connect, 10, b"ok")
|
||||
self.assertEqual(accept.packet_type, CultNetRudpPacketType.ACCEPT)
|
||||
self.assertEqual(accept.ack, 1)
|
||||
self.assertTrue(server.connected)
|
||||
self.assertEqual(server.pending_reliable_sequences, (100,))
|
||||
|
||||
client.receive(accept, 20)
|
||||
self.assertTrue(client.connected)
|
||||
self.assertEqual(client.pending_reliable_sequences, ())
|
||||
|
||||
ack = client.create_ack()
|
||||
self.assertEqual(ack.ack, 100)
|
||||
server.receive(ack, 30)
|
||||
self.assertEqual(server.pending_reliable_sequences, ())
|
||||
|
||||
def test_cultnet_rudp_session_computes_ack_masks_and_clears_pending_reliable_packets(self) -> None:
|
||||
sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=7, initial_sequence=10, resend_delay_ms=100))
|
||||
receiver = CultNetRudpSession(
|
||||
CultNetRudpSessionOptions(connection_id=7, initial_sequence=200, resend_delay_ms=100)
|
||||
)
|
||||
sender.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 7, 1, 0, 0, "control")
|
||||
)
|
||||
receiver.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 7, 2, 0, 0, "control")
|
||||
)
|
||||
|
||||
first = sender.send(
|
||||
"schema",
|
||||
b"first",
|
||||
CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=0),
|
||||
)
|
||||
second = sender.send(
|
||||
"schema",
|
||||
b"second",
|
||||
CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=0),
|
||||
)
|
||||
third = sender.send(
|
||||
"schema",
|
||||
b"third",
|
||||
CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=0),
|
||||
)
|
||||
self.assertEqual(sender.pending_reliable_sequences, (10, 11, 12))
|
||||
|
||||
receiver.receive(first)
|
||||
receiver.receive(third)
|
||||
ack_with_gap = receiver.create_ack()
|
||||
self.assertEqual(ack_with_gap.ack, 12)
|
||||
self.assertEqual(ack_with_gap.ack_mask, 0b10 | (1 << 9))
|
||||
sender.receive(ack_with_gap)
|
||||
self.assertEqual(sender.pending_reliable_sequences, (11,))
|
||||
|
||||
receiver.receive(second)
|
||||
full_ack = receiver.create_ack()
|
||||
self.assertEqual(full_ack.ack, 12)
|
||||
self.assertEqual(full_ack.ack_mask, 0b11 | (1 << 9))
|
||||
sender.receive(full_ack)
|
||||
self.assertEqual(sender.pending_reliable_sequences, ())
|
||||
|
||||
def test_cultnet_rudp_session_schedules_reliable_resends_until_acked(self) -> None:
|
||||
session = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=99, initial_sequence=1, resend_delay_ms=100))
|
||||
session.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 99, 50, 0, 0, "control")
|
||||
)
|
||||
sent = session.send(
|
||||
"schema",
|
||||
b"payload",
|
||||
CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=10),
|
||||
)
|
||||
|
||||
self.assertEqual(session.due_resends(90), ())
|
||||
self.assertEqual(tuple(packet.sequence for packet in session.due_resends(110)), (sent.sequence,))
|
||||
self.assertEqual(session.due_resends(150), ())
|
||||
|
||||
session.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACK, 99, 51, sent.sequence, 0, "control")
|
||||
)
|
||||
self.assertEqual(session.due_resends(250), ())
|
||||
|
||||
def test_cultnet_rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads(self) -> None:
|
||||
sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=1))
|
||||
receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=100))
|
||||
sender.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 90, 0, 0, "control")
|
||||
)
|
||||
receiver.receive(
|
||||
CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 91, 0, 0, "control")
|
||||
)
|
||||
|
||||
first = sender.send("schema", b"first", CultNetRudpSendOptions(reliable=True, ordered=True))
|
||||
second = sender.send("schema", b"second", CultNetRudpSendOptions(reliable=True, ordered=True))
|
||||
third = sender.send("schema", b"third", CultNetRudpSendOptions(reliable=True, ordered=True))
|
||||
|
||||
self.assertEqual([frame.payload.decode("utf-8") for frame in receiver.receive(first).delivered], ["first"])
|
||||
self.assertEqual(receiver.receive(third).delivered, ())
|
||||
self.assertEqual(receiver.receive(first).delivered, ())
|
||||
self.assertEqual(
|
||||
[frame.payload.decode("utf-8") for frame in receiver.receive(second).delivered],
|
||||
["second", "third"],
|
||||
)
|
||||
|
||||
def test_cultnet_database_subscription_helpers_match_schema_v0_shape(self) -> None:
|
||||
subscribe = database_subscribe(
|
||||
message_id="sub-message",
|
||||
|
|
|
|||
Loading…
Reference in a new issue