GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Ack Cut 1c tests (C#, Python): Connect generation, handshake-seeded watermark, restarts

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 05:52:37 +02:00
commit 536e650d82
3 changed files with 626 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

74
packages/cultnet-py/tests/test_cultnet.py
View file

@ -711,7 +711,7 @@ class CultNetTests(unittest.TestCase):
server = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=103, initial_sequence=100))
connect = client.create_connect(0, b"join")
accept = server.accept_connect(connect, 0)
reply = server.answer_repeated_connect(replace(connect, ack=accept.sequence), 1)
reply = server.accept_connect(replace(connect, ack=accept.sequence), 1)
self.assertEqual(reply.packet_type, CultNetRudpPacketType.ACK)
self.assertEqual(server.outstanding_reliable_packet_count, 0)
@ -1150,6 +1150,78 @@ class CultNetTests(unittest.TestCase):
peer_socket.close()
server.close()
def test_cultnet_rudp_server_mode_admits_a_restarted_client_and_keeps_its_session_on_a_retransmit(self) -> None:
server, _, peer_socket, _, to_server, drain_peer = self._server_mode_pair(0x10203058)
connection_id = 0x10203058
reliable = CultNetRudpSendOptions(reliable=True, ordered=True)
try:
first = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id, initial_sequence=50))
connect = first.create_connect(0, b"same")
to_server(connect)
server.receive_once()
first.receive([p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0], 0)
sent = [first.send("schema", f"c{index}".encode(), reliable) for index in range(3)]
delivered = []
for packet in sent:
to_server(packet)
delivered.append(server.receive_once().payload)
self.assertEqual(delivered, [b"c0", b"c1", b"c2"])
# A retransmitted Connect is a repeat: what was delivered stays delivered once.
to_server(connect)
server.receive_once()
to_server(sent[0])
self.assertIsNone(server.receive_once(), "the retransmitted Connect reset the session")
# The same client restarted on the same address, connection id and payload.
drain_peer()
second = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id, initial_sequence=7))
to_server(second.create_connect(0, b"same"))
server.receive_once()
second.receive([p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0], 0)
to_server(second.send("schema", b"after restart", reliable))
self.assertEqual(server.receive_once().payload, b"after restart")
finally:
peer_socket.close()
server.close()
def test_cultnet_rudp_server_mode_admits_a_connect_from_a_new_endpoint(self) -> None:
server, server_socket, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203059)
other_socket = bind_udp_socket()
try:
to_server(peer.create_connect(0, b"a"))
server.receive_once()
drain_peer()
other = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=0x10203059, initial_sequence=9))
other_socket.sendto(encode_rudp_packet(other.create_connect(0, b"b")), server_socket.getsockname())
server.receive_once()
wire, _ = other_socket.recvfrom(65535)
other.receive(decode_rudp_packet(wire), 0)
self.assertTrue(other.connected, "the new endpoint was not admitted")
other_socket.sendto(
encode_rudp_packet(other.send("schema", b"from b", CultNetRudpSendOptions(reliable=True, ordered=True))),
server_socket.getsockname(),
)
self.assertEqual(server.receive_once().payload, b"from b")
finally:
other_socket.close()
peer_socket.close()
server.close()
def test_cultnet_rudp_flush_after_a_timeout_names_the_timeout(self) -> None:
server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x1020305A)
try:
to_server(peer.create_connect(0, b"join"))
server.receive_once()
server.send("schema", b"unanswered")
self.assertTrue(server.check_timeout(0, now_ms=10**12))
with self.assertRaisesRegex(ConnectionError, "session timed out"):
server.flush_reliable(0.1)
finally:
peer_socket.close()
server.close()
def test_cultnet_rudp_session_rejects_limits_that_cannot_admit_a_connect(self) -> None:
with self.assertRaises(ValueError):
CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFF))

178
packages/cultnet-py/tests/test_rudp_connect_generation.py Normal file
View file

@ -0,0 +1,178 @@
"""A Connect's sequence says whether it repeats the one a session accepted, and
the handshake alone seeds the watermark that orders delivery."""
from __future__ import annotations
import unittest
from dataclasses import replace
from cultnet_py import random_initial_sequence
from cultnet_py.transport import (
CultNetRudpPacketType,
CultNetRudpSendOptions,
CultNetRudpSession,
CultNetRudpSessionOptions,
CultNetRudpSocketMode,
CultNetRudpSocketTransportOptions,
)
CONNECTION_ID = 0x10203090
def session(initial_sequence: int) -> CultNetRudpSession:
return CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=initial_sequence))
def send(from_session: CultNetRudpSession, text: str, *, ordered: bool = True, channel: str = "schema"):
return from_session.send(channel, text.encode(), CultNetRudpSendOptions(reliable=True, ordered=ordered))
def names(result) -> list[str]:
return [frame.payload.decode() for frame in result.delivered]
def handshake(client: CultNetRudpSession, server: CultNetRudpSession):
connect = client.create_connect(0)
accept = server.accept_connect(connect, 0)
client.receive(accept, 0)
assert client.connected and server.connected
return connect, accept
class CultNetRudpConnectGenerationTests(unittest.TestCase):
def test_a_reconnect_on_the_same_session_is_not_stranded_by_a_write_the_ended_session_owed(self) -> None:
client, server = session(1), session(500)
handshake(client, server)
send(server, "lost")
server.create_disconnect(b"gone")
client.receive(server.accept_connect(client.create_connect(1), 1), 1)
self.assertEqual(names(client.receive(send(server, "after"), 2)), ["after"])
def test_a_server_that_accepts_a_new_client_forgets_the_old_clients_sequences(self) -> None:
old_client, server = session(50), session(500)
handshake(old_client, server)
for name in ("c1", "c2", "c3"):
self.assertEqual(names(server.receive(send(old_client, name), 1)), [name])
server.create_disconnect(b"restart")
new_client = session(2)
new_client.receive(server.accept_connect(new_client.create_connect(2), 2), 2)
self.assertEqual(names(server.receive(send(new_client, "fresh"), 3)), ["fresh"])
def test_a_retransmitted_connect_after_data_has_flowed_does_not_reset_the_session(self) -> None:
client, server = session(1), session(500)
connect, _ = handshake(client, server)
frame = send(client, "once")
self.assertEqual(names(server.receive(frame, 1)), ["once"])
self.assertTrue(server.connect_repeats(connect))
self.assertEqual(server.accept_connect(connect, 2).packet_type, CultNetRudpPacketType.ACK)
self.assertEqual(server.outstanding_reliable_packet_count, 0, "a repeat queued something")
self.assertEqual(
names(server.receive(frame, 3)), [], "a retransmitted Connect made the server forget what it had delivered"
)
def test_a_connect_with_another_sequence_is_not_a_repeat_whatever_it_carries(self) -> None:
server = session(500)
connect = session(1).create_connect(0, b"same payload")
server.accept_connect(connect, 0)
self.assertTrue(server.connect_repeats(connect))
self.assertFalse(server.connect_repeats(session(9).create_connect(0, b"same payload")))
self.assertFalse(server.connect_repeats(replace(connect, packet_type=CultNetRudpPacketType.PING)))
server.create_disconnect()
self.assertFalse(server.connect_repeats(connect), "an ended session repeats nothing")
def test_an_accept_that_does_not_name_the_pending_connect_is_ignored(self) -> None:
client, server = session(10), session(500)
connect = client.create_connect(0)
accept = server.accept_connect(connect, 0)
client.receive(replace(accept, ack=connect.sequence + 50, ack_mask=0), 1)
self.assertFalse(client.connected, "a stale Accept connected the session")
self.assertEqual(client.pending_reliable_sequences, (connect.sequence,))
client.receive(replace(accept, ack=connect.sequence + 3, ack_mask=1 << 2), 2)
self.assertTrue(client.connected, "the mask names the Connect")
def test_an_accept_after_a_local_disconnect_does_not_reconnect_the_session(self) -> None:
client, server = session(1), session(500)
_, accept = handshake(client, server)
client.create_disconnect(b"bye")
self.assertFalse(client.connected)
client.receive(accept, 5)
self.assertFalse(client.connected, "a late Accept revived an ended session")
def test_a_duplicate_accept_does_not_reseed_the_watermark(self) -> None:
client, server = session(1), session(500)
_, accept = handshake(client, server)
s1, s2 = send(server, "s1"), send(server, "s2")
client.receive(replace(accept, sequence=s2.sequence), 1)
self.assertEqual(names(client.receive(s2, 2)), [], "s1 is still missing")
self.assertEqual(names(client.receive(s1, 3)), ["s1", "s2"])
def test_reliable_data_before_the_accept_is_neither_delivered_nor_acknowledged(self) -> None:
client, server = session(1), session(500)
accept = server.accept_connect(client.create_connect(0), 0)
a, b = send(server, "A"), send(server, "B")
for early in (b, a):
self.assertEqual(names(client.receive(early, 1)), [])
ack = client.create_ack_for_received(early.sequence)
self.assertEqual((ack.ack, ack.ack_mask), (0, 0), "an unhandled packet was acknowledged")
client.receive(accept, 2)
self.assertTrue(client.connected)
self.assertEqual(server.outstanding_reliable_packet_count, 3, "nothing was acknowledged")
self.assertEqual(names(client.receive(a, 3)), ["A"])
self.assertEqual(names(client.receive(b, 3)), ["B"])
def test_ordered_frames_after_the_accept_are_delivered_in_order_whatever_arrives_first(self) -> None:
client, server = session(1), session(500)
handshake(client, server)
a, b = send(server, "A"), send(server, "B")
self.assertEqual(names(client.receive(b, 1)), [])
self.assertEqual(names(client.receive(a, 2)), ["A", "B"])
def test_an_unordered_reliable_frame_is_delivered_on_receipt_never_held_behind_a_gap(self) -> None:
client, server = session(1), session(500)
handshake(client, server)
send(server, "lost")
unordered = send(server, "now", ordered=False, channel="media")
ordered = send(server, "later")
self.assertEqual(names(client.receive(unordered, 1)), ["now"])
self.assertEqual(names(client.receive(ordered, 2)), [], "the ordered frame waits for the gap")
def test_a_peer_disconnect_does_not_forget_what_the_peer_sent(self) -> None:
client, server = session(1), session(500)
handshake(client, server)
s1, s2 = send(client, "s1"), send(client, "s2")
self.assertEqual(names(server.receive(s2, 1)), [], "s1 is missing, so s2 is held")
server.receive(client.create_disconnect(b"bye"), 2)
self.assertEqual(names(server.receive(s1, 3)), ["s1", "s2"], "the Disconnect forgot the held frame")
self.assertEqual(names(server.receive(s2, 4)), [], "the Disconnect forgot what was received")
def test_default_initial_sequences_are_drawn_at_random_from_one_to_two_to_the_thirty_first(self) -> None:
def assert_random(label: str, draw) -> None:
draws = {draw() for _ in range(64)}
self.assertGreater(len(draws), 32, f"{label} does not draw at random")
self.assertTrue(all(1 <= value < 2**31 for value in draws), f"{label} left [1, 2^31)")
assert_random("random_initial_sequence", random_initial_sequence)
assert_random("session options", lambda: CultNetRudpSessionOptions(connection_id=1).initial_sequence)
assert_random(
"transport options",
lambda: CultNetRudpSocketTransportOptions(
runtime_id="r", socket=None, mode=CultNetRudpSocketMode.CLIENT, connection_id=1
).initial_sequence,
)
assert_random(
"a default session's first Connect",
lambda: CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1)).create_connect(0).sequence,
)
if __name__ == "__main__":
unittest.main()

375
tests/GameCult.Networking.Tests/CultNetRudpConnectGenerationTests.cs Normal file
View file

@ -0,0 +1,375 @@
#nullable enable
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using GameCult.Mesh;
using NUnit.Framework;
namespace GameCult.Networking.Tests
{
/// <summary>
/// A Connect's sequence says whether it repeats the one a session accepted, and the handshake
/// alone seeds the watermark that orders delivery.
/// </summary>
public sealed class CultNetRudpConnectGenerationTests
{
private const uint ConnectionId = 0x10203090;
private static readonly CultNetRudpSendOptions Ordered = new CultNetRudpSendOptions { Reliable = true, Ordered = true };
private static CultNetRudpSession Session(uint initialSequence) =>
new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = initialSequence });
private static CultNetRudpPacket Send(CultNetRudpSession session, string text, bool ordered = true, string channel = "schema") =>
session.Send(channel, Encoding.UTF8.GetBytes(text), new CultNetRudpSendOptions { Reliable = true, Ordered = ordered });
private static string[] Names(CultNetRudpReceiveResult result) =>
result.Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray();
private static (CultNetRudpPacket Connect, CultNetRudpPacket Accept) Handshake(CultNetRudpSession client, CultNetRudpSession server)
{
var connect = client.CreateConnect(0);
var accept = server.AcceptConnect(connect, 0);
client.Receive(accept, 0);
Assert.That(client.Connected && server.Connected, Is.True);
return (connect, accept);
}
private static Socket Bind()
{
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
socket.ReceiveTimeout = 20;
return socket;
}
private static List<CultNetRudpPacket> Drain(Socket socket)
{
var packets = new List<CultNetRudpPacket>();
var buffer = new byte[65535];
EndPoint from = new IPEndPoint(IPAddress.Any, 0);
while (socket.Poll(20_000, SelectMode.SelectRead))
{
var received = socket.ReceiveFrom(buffer, ref from);
packets.Add(CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray()));
}
return packets;
}
[Test]
public void AReconnectOnTheSameSessionIsNotStrandedByAWriteTheEndedSessionOwed()
{
var client = Session(1);
var server = Session(500);
Handshake(client, server);
Send(server, "lost");
server.CreateDisconnect(Encoding.UTF8.GetBytes("gone"));
client.Receive(server.AcceptConnect(client.CreateConnect(1), 1), 1);
Assert.That(Names(client.Receive(Send(server, "after"), 2)), Is.EqualTo(new[] { "after" }));
}
[Test]
public void AServerThatAcceptsANewClientForgetsTheOldClientsSequences()
{
var oldClient = Session(50);
var server = Session(500);
Handshake(oldClient, server);
foreach (var name in new[] { "c1", "c2", "c3" })
Assert.That(Names(server.Receive(Send(oldClient, name), 1)), Is.EqualTo(new[] { name }));
server.CreateDisconnect(Encoding.UTF8.GetBytes("restart"));
var newClient = Session(2);
newClient.Receive(server.AcceptConnect(newClient.CreateConnect(2), 2), 2);
Assert.That(Names(server.Receive(Send(newClient, "fresh"), 3)), Is.EqualTo(new[] { "fresh" }));
}
[Test]
public void ARetransmittedConnectAfterDataHasFlowedDoesNotResetTheSession()
{
var client = Session(1);
var server = Session(500);
var (connect, _) = Handshake(client, server);
var frame = Send(client, "once");
Assert.That(Names(server.Receive(frame, 1)), Is.EqualTo(new[] { "once" }));
Assert.That(server.ConnectRepeats(connect), Is.True);
Assert.That(server.AcceptConnect(connect, 2).PacketType, Is.EqualTo(CultNetRudpPacketType.Ack), "the Accept was already acknowledged");
Assert.That(server.OutstandingReliablePacketCount, Is.EqualTo(0), "a repeat queued something");
Assert.That(server.Receive(frame, 3).Delivered, Is.Empty, "a retransmitted Connect made the server forget what it had delivered");
}
[Test]
public void AConnectWithAnotherSequenceIsNotARepeatWhateverItCarries()
{
var server = Session(500);
var connect = Session(1).CreateConnect(0, Encoding.UTF8.GetBytes("same payload"));
server.AcceptConnect(connect, 0);
Assert.That(server.ConnectRepeats(connect), Is.True);
Assert.That(server.ConnectRepeats(Session(9).CreateConnect(0, Encoding.UTF8.GetBytes("same payload"))), Is.False);
var notAConnect = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(connect));
notAConnect.PacketType = CultNetRudpPacketType.Ping;
Assert.That(server.ConnectRepeats(notAConnect), Is.False);
server.CreateDisconnect();
Assert.That(server.ConnectRepeats(connect), Is.False, "an ended session repeats nothing");
}
[Test]
public void AnAcceptThatDoesNotNameThePendingConnectIsIgnored()
{
var client = Session(10);
var server = Session(500);
var connect = client.CreateConnect(0);
var accept = server.AcceptConnect(connect, 0);
var stale = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(accept));
stale.Ack = connect.Sequence + 50;
stale.AckMask = 0;
client.Receive(stale, 1);
Assert.That(client.Connected, Is.False, "a stale Accept connected the session");
Assert.That(client.PendingReliableSequences, Is.EqualTo(new[] { connect.Sequence }));
var named = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(accept));
named.Ack = connect.Sequence + 3;
named.AckMask = 1u << 2;
client.Receive(named, 2);
Assert.That(client.Connected, Is.True, "the mask names the Connect");
}
[Test]
public void AnAcceptAfterALocalDisconnectDoesNotReconnectTheSession()
{
var client = Session(1);
var server = Session(500);
var (_, accept) = Handshake(client, server);
client.CreateDisconnect(Encoding.UTF8.GetBytes("bye"));
Assert.That(client.Connected, Is.False);
client.Receive(accept, 5);
Assert.That(client.Connected, Is.False, "a late Accept revived an ended session");
}
[Test]
public void ADuplicateAcceptDoesNotReseedTheWatermark()
{
var client = Session(1);
var server = Session(500);
var (_, accept) = Handshake(client, server);
var s1 = Send(server, "s1");
var s2 = Send(server, "s2");
var forged = CultNetRudpPacketCodec.Decode(CultNetRudpPacketCodec.Encode(accept));
forged.Sequence = s2.Sequence;
client.Receive(forged, 1);
Assert.That(client.Receive(s2, 2).Delivered, Is.Empty, "s1 is still missing");
Assert.That(Names(client.Receive(s1, 3)), Is.EqualTo(new[] { "s1", "s2" }));
}
[Test]
public void ReliableDataBeforeTheAcceptIsNeitherDeliveredNorAcknowledged()
{
var client = Session(1);
var server = Session(500);
var accept = server.AcceptConnect(client.CreateConnect(0), 0);
var a = Send(server, "A");
var b = Send(server, "B");
foreach (var early in new[] { b, a })
{
Assert.That(client.Receive(early, 1).Delivered, Is.Empty);
var ack = client.CreateAckForReceived(early.Sequence);
Assert.That((ack.Ack, ack.AckMask), Is.EqualTo((0u, 0u)), "an unhandled packet was acknowledged");
}
client.Receive(accept, 2);
Assert.That(client.Connected, Is.True);
Assert.That(server.OutstandingReliablePacketCount, Is.EqualTo(3), "nothing was acknowledged");
Assert.That(Names(client.Receive(a, 3)), Is.EqualTo(new[] { "A" }));
Assert.That(Names(client.Receive(b, 3)), Is.EqualTo(new[] { "B" }));
}
[Test]
public void OrderedFramesAfterTheAcceptAreDeliveredInOrderWhateverArrivesFirst()
{
var client = Session(1);
var server = Session(500);
Handshake(client, server);
var a = Send(server, "A");
var b = Send(server, "B");
Assert.That(client.Receive(b, 1).Delivered, Is.Empty);
Assert.That(Names(client.Receive(a, 2)), Is.EqualTo(new[] { "A", "B" }));
}
[Test]
public void AnUnorderedReliableFrameIsDeliveredOnReceiptNeverHeldBehindAGap()
{
var client = Session(1);
var server = Session(500);
Handshake(client, server);
Send(server, "lost");
var unordered = Send(server, "now", ordered: false, channel: "media");
var ordered = Send(server, "later");
Assert.That(Names(client.Receive(unordered, 1)), Is.EqualTo(new[] { "now" }));
Assert.That(client.Receive(ordered, 2).Delivered, Is.Empty, "the ordered frame waits for the gap");
}
[Test]
public void APeerDisconnectDoesNotForgetWhatThePeerSent()
{
var client = Session(1);
var server = Session(500);
Handshake(client, server);
var s1 = Send(client, "s1");
var s2 = Send(client, "s2");
Assert.That(server.Receive(s2, 1).Delivered, Is.Empty, "s1 is missing, so s2 is held");
server.Receive(client.CreateDisconnect(Encoding.UTF8.GetBytes("bye")), 2);
Assert.That(Names(server.Receive(s1, 3)), Is.EqualTo(new[] { "s1", "s2" }), "the Disconnect forgot the held frame");
Assert.That(server.Receive(s2, 4).Delivered, Is.Empty, "the Disconnect forgot what was received");
}
[Test]
public void DefaultInitialSequencesAreDrawnAtRandomFromOneToTwoToTheThirtyFirst()
{
static void AssertRandom(string label, Func<uint> draw)
{
var draws = Enumerable.Range(0, 64).Select(_ => draw()).ToArray();
Assert.That(draws.Distinct().Count(), Is.GreaterThan(32), $"{label} does not draw at random");
Assert.That(draws.All(value => value >= 1 && value < (1u << 31)), Is.True, $"{label} left [1, 2^31)");
}
AssertRandom("RandomInitialSequence", CultNetRudpSession.RandomInitialSequence);
AssertRandom("session options", () => new CultNetRudpSessionOptions().InitialSequence);
AssertRandom("socket transport options", () => new CultNetRudpSocketTransportOptions().InitialSequence);
AssertRandom("listener options", () => new CultNetRudpSocketTransportServerOptions().InitialSequence);
AssertRandom("schema server options", () => new RudpCultNetSchemaServerOptions().InitialSequence);
AssertRandom("cultmesh options", () => new CultMeshRudpSocketOptions().InitialSequence);
}
// Transports.
[Test]
public void ServerModeAdmitsARestartedClientKeepsItsSessionOnARetransmitAndAdmitsANewEndpoint()
{
using var serverSocket = Bind();
using var peerSocket = Bind();
using var otherSocket = Bind();
using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = "csharp-rudp-server",
Socket = serverSocket,
Mode = CultNetRudpSocketMode.Server,
ConnectionId = ConnectionId
});
var serverEndPoint = serverSocket.LocalEndPoint!;
void ToServer(Socket from, CultNetRudpPacket packet) => from.SendTo(CultNetRudpPacketCodec.Encode(packet), serverEndPoint);
var first = Session(50);
var connect = first.CreateConnect(0, Encoding.UTF8.GetBytes("same"));
ToServer(peerSocket, connect);
server.ReceiveOnce();
first.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
var sent = new[] { "c0", "c1", "c2" }.Select(name => Send(first, name)).ToArray();
foreach (var packet in sent)
{
ToServer(peerSocket, packet);
Assert.That(Encoding.UTF8.GetString(server.ReceiveOnce()!.Payload), Is.EqualTo(Encoding.UTF8.GetString(packet.Payload)));
}
// A retransmitted Connect is a repeat: what was delivered stays delivered once.
ToServer(peerSocket, connect);
server.ReceiveOnce();
ToServer(peerSocket, sent[0]);
Assert.That(server.ReceiveOnce(), Is.Null, "the retransmitted Connect reset the session");
// The same client restarted on the same address, connection id and payload.
Drain(peerSocket);
var second = Session(7);
ToServer(peerSocket, second.CreateConnect(0, Encoding.UTF8.GetBytes("same")));
server.ReceiveOnce();
second.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
ToServer(peerSocket, Send(second, "after restart"));
Assert.That(Encoding.UTF8.GetString(server.ReceiveOnce()!.Payload), Is.EqualTo("after restart"));
// A Connect from a new endpoint claims the transport.
var third = Session(9);
ToServer(otherSocket, third.CreateConnect(0, Encoding.UTF8.GetBytes("elsewhere")));
server.ReceiveOnce();
third.Receive(Drain(otherSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
ToServer(otherSocket, Send(third, "from elsewhere"));
Assert.That(Encoding.UTF8.GetString(server.ReceiveOnce()!.Payload), Is.EqualTo("from elsewhere"));
}
[Test]
public void TheListenerAdmitsARestartedClientWithTheSameAddressIdAndPayload()
{
using var serverSocket = Bind();
using var peerSocket = Bind();
using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
{
RuntimeId = "csharp-rudp-listener",
Socket = serverSocket,
ConnectionId = ConnectionId
});
var ended = new List<CultNetRudpSocketServerPeer>();
server.PeerDisconnected += ended.Add;
var serverEndPoint = serverSocket.LocalEndPoint!;
void ToServer(CultNetRudpPacket packet) => peerSocket.SendTo(CultNetRudpPacketCodec.Encode(packet), serverEndPoint);
var first = Session(50);
ToServer(first.CreateConnect(0, Encoding.UTF8.GetBytes("same")));
server.ReceiveOnce();
first.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
for (var index = 0; index < 3; index++)
{
ToServer(Send(first, $"c{index}"));
Assert.That(server.ReceiveOnce(), Is.Not.Null);
}
Assert.That(ended, Is.Empty);
var second = Session(7);
ToServer(second.CreateConnect(0, Encoding.UTF8.GetBytes("same")));
server.ReceiveOnce();
Assert.That(ended, Has.Count.EqualTo(1), "the restarted client was taken for a retransmit");
second.Receive(Drain(peerSocket).First(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
ToServer(Send(second, "after restart"));
var frame = server.ReceiveOnce();
Assert.That(frame, Is.Not.Null);
Assert.That(Encoding.UTF8.GetString(frame!.Frame.Payload), Is.EqualTo("after restart"));
}
/// <summary>
/// Windows reports the ICMP port-unreachable for a datagram sent to a closed port as a connection
/// reset on the next receive. Nothing ended, so nothing may be reported. Other platforms do not
/// raise it, so the test is inconclusive there.
/// </summary>
[Test]
public void AnIcmpConnectionResetDoesNotEndTheSession()
{
Assume.That(OperatingSystem.IsWindows(), "only Windows raises an ICMP connection reset on UDP");
using var serverSocket = Bind();
using var peerSocket = Bind();
using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = "csharp-rudp-server",
Socket = serverSocket,
Mode = CultNetRudpSocketMode.Server,
ConnectionId = ConnectionId
});
peerSocket.SendTo(CultNetRudpPacketCodec.Encode(Session(1).CreateConnect(0)), serverSocket.LocalEndPoint!);
server.ReceiveOnce();
Assert.That(server.Connected, Is.True);
var closed = Bind();
var closedEndPoint = closed.LocalEndPoint!;
closed.Dispose();
serverSocket.SendTo(new byte[] { 1 }, closedEndPoint);
System.Threading.Thread.Sleep(50);
server.ReceiveOnce();
Assert.That(server.DisconnectReason, Is.Null, "an ICMP reset claimed an ending that did not happen");
Assert.That(server.Connected, Is.True);
}
}
}