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

Separate RUDP reliable and lossy sequence domains

Browse source
This commit is contained in:
Meta Vens 2026-08-26 17:16:57 +02:00
commit 0736ae4824
6 changed files with 372 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

65
packages/cultnet-rs/src/rudp.rs
View file

@ -159,11 +159,13 @@ pub struct CultNetRudpSession {
max_pending_fragment_sets: usize,
initial_sequence: u32,
next_sequence: u32,
next_sequenced_by_channel: BTreeMap<String, u32>,
next_fragment_id: u16,
connected: bool,
last_received_at_ms: Option<u64>,
highest_received_sequence: Option<u32>,
received_sequences: BTreeSet<u32>,
latest_sequenced_by_channel: BTreeMap<String, u32>,
pending_reliable: BTreeMap<u32, PendingReliablePacket>,
queued_reliable: VecDeque<CultNetRudpPacket>,
ordered_next_sequence_by_channel: BTreeMap<String, u32>,
@ -182,11 +184,13 @@ impl CultNetRudpSession {
max_pending_fragment_sets: 64,
initial_sequence: options.initial_sequence,
next_sequence: options.initial_sequence,
next_sequenced_by_channel: BTreeMap::new(),
next_fragment_id: 1,
connected: false,
last_received_at_ms: None,
highest_received_sequence: None,
received_sequences: BTreeSet::new(),
latest_sequenced_by_channel: BTreeMap::new(),
pending_reliable: BTreeMap::new(),
queued_reliable: VecDeque::new(),
ordered_next_sequence_by_channel: BTreeMap::new(),
@ -290,11 +294,13 @@ impl CultNetRudpSession {
pub fn reset_peer_state(&mut self) {
self.session_scope = Uuid::new_v4();
self.next_sequence = self.initial_sequence;
self.next_sequenced_by_channel.clear();
self.next_fragment_id = 1;
self.connected = false;
self.last_received_at_ms = None;
self.highest_received_sequence = None;
self.received_sequences.clear();
self.latest_sequenced_by_channel.clear();
self.pending_reliable.clear();
self.queued_reliable.clear();
self.ordered_next_sequence_by_channel.clear();
@ -377,6 +383,9 @@ impl CultNetRudpSession {
"Cannot send RUDP data before the session is connected"
));
}
if options.ordered && !options.reliable {
return Err(anyhow!("RUDP ordered delivery requires reliability"));
}
self.require_payload_size(payload.len())?;
if let Some(max_fragment_bytes) = max_fragment_bytes {
@ -459,7 +468,6 @@ impl CultNetRudpSession {
}
if packet.packet_type == CultNetRudpPacketType::Ping {
self.remember_received(packet.sequence);
return Ok(CultNetRudpReceiveResult {
delivered: Vec::new(),
ready_to_send,
@ -481,9 +489,6 @@ impl CultNetRudpSession {
if packet.packet_type == CultNetRudpPacketType::Ack
|| packet.packet_type == CultNetRudpPacketType::Pong
{
if packet.packet_type == CultNetRudpPacketType::Pong {
self.remember_received(packet.sequence);
}
return Ok(CultNetRudpReceiveResult {
delivered: Vec::new(),
ready_to_send,
@ -500,7 +505,6 @@ impl CultNetRudpSession {
}
if packet.packet_type == CultNetRudpPacketType::Disconnect {
self.remember_received(packet.sequence);
self.connected = false;
return Ok(CultNetRudpReceiveResult {
delivered: Vec::new(),
@ -525,12 +529,15 @@ impl CultNetRudpSession {
});
}
let duplicate = self.received_sequences.contains(&packet.sequence)
|| self.highest_received_sequence.is_some_and(|highest| {
packet.sequence < highest
&& highest - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW as u32
});
self.remember_received(packet.sequence);
let duplicate = packet.reliable
&& (self.received_sequences.contains(&packet.sequence)
|| self.highest_received_sequence.is_some_and(|highest| {
packet.sequence < highest
&& highest - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW as u32
}));
if packet.reliable {
self.remember_received(packet.sequence);
}
if duplicate {
return Ok(CultNetRudpReceiveResult {
delivered: Vec::new(),
@ -556,6 +563,19 @@ impl CultNetRudpSession {
};
let delivered = if ordered {
self.deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized)
} else if packet.sequenced {
let latest = self
.latest_sequenced_by_channel
.get(&frame.channel_id)
.copied();
let newest_sequence = next_sequence.saturating_sub(1);
if latest.is_some_and(|latest| newest_sequence <= latest) {
Vec::new()
} else {
self.latest_sequenced_by_channel
.insert(frame.channel_id.clone(), newest_sequence);
vec![frame]
}
} else {
vec![frame]
};
@ -700,11 +720,24 @@ impl CultNetRudpSession {
fragment_index: u16,
fragment_count: u16,
) -> CultNetRudpPacket {
let sequence = self.next_sequence;
self.next_sequence = self
.next_sequence
.checked_add(1)
.expect("sequence overflow");
let sequence = if reliable {
let sequence = self.next_sequence;
self.next_sequence = self
.next_sequence
.checked_add(1)
.expect("reliable sequence overflow");
sequence
} else if sequenced {
let next = self
.next_sequenced_by_channel
.entry(channel_id.to_string())
.or_insert(1);
let sequence = *next;
*next = next.checked_add(1).expect("sequenced channel overflow");
sequence
} else {
0
};
let (ack, ack_mask) = self.ack_state();
CultNetRudpPacket {
packet_type,

127
packages/cultnet-rs/tests/cultnet.rs
View file

@ -757,6 +757,133 @@ fn rudp_session_pings_and_detects_receive_timeout() -> Result<()> {
Ok(())
}
#[test]
fn rudp_lossy_packets_cannot_create_reliable_ordered_gaps() -> Result<()> {
let mut client = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: 198,
initial_sequence: 1,
..CultNetRudpSessionOptions::default()
});
let mut server = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: 198,
initial_sequence: 100,
..CultNetRudpSessionOptions::default()
});
let connect = client.create_connect(0, Vec::new())?;
let accept = server.accept_connect(&connect, 1, Vec::new())?;
client.receive(&accept, 2)?;
let ping = client.create_ping(b"pulse".to_vec());
let pong = server
.receive(&ping, 3)?
.reply
.expect("ping should produce pong");
let realtime = server.send(
"realtime",
b"discarded realtime".to_vec(),
CultNetRudpSendOptions::default(),
)?;
let latest = server.send(
"latest",
b"discarded latest state".to_vec(),
CultNetRudpSendOptions {
sequenced: true,
..CultNetRudpSendOptions::default()
},
)?;
let schema = server.send(
"schema",
b"committed response".to_vec(),
CultNetRudpSendOptions {
reliable: true,
ordered: true,
..CultNetRudpSendOptions::default()
},
)?;
assert_eq!(ping.sequence, 0);
assert_eq!(pong.sequence, 0);
assert_eq!(realtime.sequence, 0);
assert_eq!(latest.sequence, 1);
assert_eq!(schema.sequence, accept.sequence + 1);
assert_eq!(
client.receive(&schema, 4)?.delivered[0].payload,
b"committed response"
);
Ok(())
}
#[test]
fn rudp_unreliable_sequenced_delivery_is_scoped_to_its_channel() -> Result<()> {
let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: 197,
initial_sequence: 50,
..CultNetRudpSessionOptions::default()
});
let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: 197,
initial_sequence: 100,
..CultNetRudpSessionOptions::default()
});
let connect = sender.create_connect(0, Vec::new())?;
let accept = receiver.accept_connect(&connect, 1, Vec::new())?;
sender.receive(&accept, 2)?;
let options = CultNetRudpSendOptions {
sequenced: true,
..CultNetRudpSendOptions::default()
};
let older = sender.send("latest", b"older".to_vec(), options.clone())?;
let newer = sender.send("latest", b"newer".to_vec(), options)?;
assert_eq!(older.sequence, 1);
assert_eq!(newer.sequence, 2);
assert_eq!(receiver.receive(&newer, 3)?.delivered[0].payload, b"newer");
assert!(receiver.receive(&older, 4)?.delivered.is_empty());
Ok(())
}
#[test]
fn rudp_rejects_unreliable_ordered_delivery() -> Result<()> {
let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: 196,
..CultNetRudpSessionOptions::default()
});
session.receive(
&CultNetRudpPacket {
packet_type: CultNetRudpPacketType::Accept,
connection_id: 196,
sequence: 50,
ack: 0,
ack_mask: 0,
channel_id: "control".to_string(),
reliable: false,
ordered: false,
sequenced: false,
fragment_id: 0,
fragment_index: 0,
fragment_count: 0,
payload: Vec::new(),
},
0,
)?;
let error = session
.send(
"schema",
b"cannot order what will not retransmit".to_vec(),
CultNetRudpSendOptions {
ordered: true,
..CultNetRudpSendOptions::default()
},
)
.unwrap_err();
assert!(
error
.to_string()
.contains("ordered delivery requires reliability")
);
Ok(())
}
#[test]
fn rudp_session_bounds_pending_reliable_packets_before_enqueue() -> Result<()> {
let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions {

55
packages/cultnet-ts/src/rudp.ts
View file

@ -147,12 +147,14 @@ export class CultNetRudpSession {
readonly connectionId: number;
readonly resendDelayMs: number;
#nextSequence: number;
readonly #nextSequencedByChannel = new Map<string, number>();
#nextFragmentId = 1;
#connected = false;
readonly #maxPendingReliablePackets: number | undefined;
#lastReceivedAtMs: number | undefined;
#highestReceivedSequence: number | undefined;
readonly #receivedSequences = new Set<number>();
readonly #latestSequencedByChannel = new Map<string, number>();
readonly #pendingReliable = new Map<number, PendingReliablePacket>();
readonly #queuedReliable: CultNetRudpPacket[] = [];
readonly #orderedNextSequenceByChannel = new Map<string, number>();
@ -194,6 +196,8 @@ export class CultNetRudpSession {
this.#lastReceivedAtMs = undefined;
this.#highestReceivedSequence = undefined;
this.#receivedSequences.clear();
this.#nextSequencedByChannel.clear();
this.#latestSequencedByChannel.clear();
this.#pendingReliable.clear();
this.#queuedReliable.splice(0);
this.#orderedNextSequenceByChannel.clear();
@ -271,6 +275,9 @@ export class CultNetRudpSession {
if (!this.#connected) {
throw new Error("Cannot send RUDP data before the session is connected.");
}
if (options.ordered && !options.reliable) {
throw new Error("RUDP ordered delivery requires reliability.");
}
const maxFragmentBytes = options.maxFragmentBytes;
if (maxFragmentBytes !== undefined && maxFragmentBytes <= 0) {
@ -336,7 +343,6 @@ export class CultNetRudpSession {
}
if (packet.packetType === "ping") {
this.#rememberReceived(packet.sequence);
return {
delivered: [],
readyToSend,
@ -349,9 +355,8 @@ export class CultNetRudpSession {
}
if (packet.packetType === "ack" || packet.packetType === "pong") {
if (packet.packetType === "pong") this.#rememberReceived(packet.sequence);
return {
delivered: this.#drainUnblockedOrdered(),
delivered: [],
readyToSend,
pong: packet.packetType === "pong",
pongPayload: packet.packetType === "pong" ? packet.payload ?? new Uint8Array() : undefined,
@ -359,7 +364,6 @@ export class CultNetRudpSession {
}
if (packet.packetType === "disconnect") {
this.#rememberReceived(packet.sequence);
this.#connected = false;
return {
delivered: [],
@ -373,11 +377,12 @@ export class CultNetRudpSession {
return { delivered: [], readyToSend };
}
const isDuplicate = this.#receivedSequences.has(packet.sequence)
|| (this.#highestReceivedSequence !== undefined
&& packet.sequence < this.#highestReceivedSequence
&& this.#highestReceivedSequence - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW);
this.#rememberReceived(packet.sequence);
const isDuplicate = (packet.reliable ?? false)
&& (this.#receivedSequences.has(packet.sequence)
|| (this.#highestReceivedSequence !== undefined
&& packet.sequence < this.#highestReceivedSequence
&& this.#highestReceivedSequence - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW));
if (packet.reliable ?? false) this.#rememberReceived(packet.sequence);
if (isDuplicate) {
return { delivered: [], readyToSend };
}
@ -387,6 +392,16 @@ export class CultNetRudpSession {
return { delivered: [], readyToSend };
}
if (!reassembled.ordered && (packet.sequenced ?? false)) {
const newestSequence = reassembled.nextSequence - 1;
const latestSequence = this.#latestSequencedByChannel.get(reassembled.frame.channelId);
if (latestSequence !== undefined && newestSequence <= latestSequence) {
return { delivered: [], readyToSend };
}
this.#latestSequencedByChannel.set(reassembled.frame.channelId, newestSequence);
return { delivered: [reassembled.frame], readyToSend };
}
if (!reassembled.ordered) {
return { delivered: [reassembled.frame], readyToSend };
}
@ -492,8 +507,17 @@ export class CultNetRudpSession {
fragmentIndex?: number;
fragmentCount?: number;
}): CultNetRudpPacket {
const sequence = this.#nextSequence;
this.#nextSequence = toUint32(this.#nextSequence + 1, "sequence");
let sequence = 0;
if (packet.reliable ?? false) {
sequence = this.#nextSequence;
this.#nextSequence = toUint32(this.#nextSequence + 1, "reliable sequence");
} else if (packet.sequenced ?? false) {
sequence = this.#nextSequencedByChannel.get(packet.channelId) ?? 1;
this.#nextSequencedByChannel.set(
packet.channelId,
toUint32(sequence + 1, "sequenced channel sequence"),
);
}
const { ack, ackMask } = this.#ackState();
return {
packetType: packet.packetType,
@ -712,15 +736,6 @@ export class CultNetRudpSession {
return delivered;
}
#drainUnblockedOrdered(): CultNetRudpDeliveredFrame[] {
const delivered: CultNetRudpDeliveredFrame[] = [];
for (const channelId of this.#orderedBuffers.keys()) {
this.#skipReceivedNonChannelSequences(channelId);
delivered.push(...this.#drainOrdered(channelId));
}
return delivered;
}
#skipReceivedNonChannelSequences(channelId: string): void {
let next = this.#orderedNextSequenceByChannel.get(channelId);
while (next !== undefined && this.#receivedSequences.has(next) && !this.#orderedBuffers.get(channelId)?.has(next)) {

47
packages/cultnet-ts/test/cultnet.test.ts
View file

@ -553,6 +553,53 @@ test("rudp session pings and detects receive timeout", () => {
assert.equal(client.connected, false);
});
test("rudp lossy packets cannot create reliable ordered gaps", () => {
const client = new CultNetRudpSession({ connectionId: 198, initialSequence: 1 });
const server = new CultNetRudpSession({ connectionId: 198, initialSequence: 100 });
const connect = client.createConnect(0);
const accept = server.acceptConnect(connect, 1);
client.receive(accept, 2);
const ping = client.createPing(Buffer.from("pulse"));
const pong = server.receive(ping, 3).reply!;
const realtime = server.send("realtime", Buffer.from("discarded realtime"));
const latest = server.send("latest", Buffer.from("discarded latest state"), { sequenced: true });
const schema = server.send("schema", Buffer.from("committed response"), { reliable: true, ordered: true });
assert.equal(ping.sequence, 0);
assert.equal(pong.sequence, 0);
assert.equal(realtime.sequence, 0);
assert.equal(latest.sequence, 1);
assert.equal(schema.sequence, accept.sequence + 1);
assert.deepEqual(client.receive(schema, 4).delivered.map(frame => Buffer.from(frame.payload).toString("utf8")), [
"committed response",
]);
});
test("rudp unreliable sequenced delivery is scoped to its channel", () => {
const sender = new CultNetRudpSession({ connectionId: 197, initialSequence: 50 });
const receiver = new CultNetRudpSession({ connectionId: 197, initialSequence: 100 });
const connect = sender.createConnect(0);
const accept = receiver.acceptConnect(connect, 1);
sender.receive(accept, 2);
const older = sender.send("latest", Buffer.from("older"), { sequenced: true });
const newer = sender.send("latest", Buffer.from("newer"), { sequenced: true });
assert.equal(older.sequence, 1);
assert.equal(newer.sequence, 2);
assert.deepEqual(receiver.receive(newer, 3).delivered.map(frame => Buffer.from(frame.payload).toString("utf8")), ["newer"]);
assert.deepEqual(receiver.receive(older, 4).delivered, []);
});
test("rudp rejects unreliable ordered delivery", () => {
const session = new CultNetRudpSession({ connectionId: 196 });
session.receive({ packetType: "accept", connectionId: 196, sequence: 50, ack: 0, ackMask: 0, channelId: "control" });
assert.throws(
() => session.send("schema", Buffer.from("cannot order what will not retransmit"), { ordered: true }),
/ordered delivery requires reliability/,
);
});
test("rudp session bounds pending reliable packets before enqueue", () => {
const session = new CultNetRudpSession({ connectionId: 102, initialSequence: 1, maxPendingReliablePackets: 2 });
session.receive({ packetType: "accept", connectionId: 102, sequence: 50, ack: 0, ackMask: 0, channelId: "control" });

68
src/GameCult.Networking/CultNetTransport.cs
View file

@ -894,12 +894,14 @@ namespace GameCult.Networking
}
private uint _nextSequence;
private readonly Dictionary<string, uint> _nextSequencedByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
private ushort _nextFragmentId = 1;
private readonly int? _maxPendingReliablePackets;
private volatile bool _connected;
private long? _lastReceivedAtMs;
private uint? _highestReceivedSequence;
private readonly HashSet<uint> _receivedSequences = new HashSet<uint>();
private readonly Dictionary<string, uint> _latestSequencedByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
private readonly object _pendingReliableGate = new object();
private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
private readonly Queue<CultNetRudpPacket> _queuedReliable = new Queue<CultNetRudpPacket>();
@ -1022,6 +1024,10 @@ namespace GameCult.Networking
}
options ??= new CultNetRudpSendOptions();
if (options.Ordered && !options.Reliable)
{
throw new InvalidOperationException("RUDP ordered delivery requires reliability.");
}
payload ??= Array.Empty<byte>();
if (maxFragmentBytes.HasValue && maxFragmentBytes.Value <= 0)
{
@ -1099,7 +1105,6 @@ namespace GameCult.Networking
if (packet.PacketType == CultNetRudpPacketType.Ping)
{
RememberReceived(packet.Sequence);
return new CultNetRudpReceiveResult
{
ReadyToSend = readyToSend,
@ -1115,11 +1120,9 @@ namespace GameCult.Networking
if (packet.PacketType == CultNetRudpPacketType.Ack || packet.PacketType == CultNetRudpPacketType.Pong)
{
if (packet.PacketType == CultNetRudpPacketType.Pong)
RememberReceived(packet.Sequence);
return new CultNetRudpReceiveResult
{
Delivered = DrainUnblockedOrdered(),
Delivered = Array.Empty<CultNetRudpDeliveredFrame>(),
ReadyToSend = readyToSend,
Pong = packet.PacketType == CultNetRudpPacketType.Pong,
PongPayload = packet.PacketType == CultNetRudpPacketType.Pong
@ -1130,7 +1133,6 @@ namespace GameCult.Networking
if (packet.PacketType == CultNetRudpPacketType.Disconnect)
{
RememberReceived(packet.Sequence);
_connected = false;
return new CultNetRudpReceiveResult
{
@ -1145,11 +1147,15 @@ namespace GameCult.Networking
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend };
}
var duplicate = _receivedSequences.Contains(packet.Sequence)
|| (_highestReceivedSequence.HasValue
&& packet.Sequence < _highestReceivedSequence.Value
&& _highestReceivedSequence.Value - packet.Sequence >= ReceivedSequenceWindow);
RememberReceived(packet.Sequence);
var duplicate = packet.Reliable
&& (_receivedSequences.Contains(packet.Sequence)
|| (_highestReceivedSequence.HasValue
&& packet.Sequence < _highestReceivedSequence.Value
&& _highestReceivedSequence.Value - packet.Sequence >= ReceivedSequenceWindow));
if (packet.Reliable)
{
RememberReceived(packet.Sequence);
}
if (duplicate)
{
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend };
@ -1161,6 +1167,17 @@ namespace GameCult.Networking
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend };
}
if (!reassembled.Ordered && packet.Sequenced)
{
var newestSequence = reassembled.NextSequence - 1;
if (_latestSequencedByChannel.TryGetValue(reassembled.Frame.ChannelId, out var latestSequence)
&& newestSequence <= latestSequence)
{
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend };
}
_latestSequencedByChannel[reassembled.Frame.ChannelId] = newestSequence;
}
return new CultNetRudpReceiveResult
{
ReadyToSend = readyToSend,
@ -1282,8 +1299,24 @@ namespace GameCult.Networking
ushort fragmentIndex,
ushort fragmentCount)
{
var sequence = _nextSequence;
_nextSequence = checked(_nextSequence + 1);
uint sequence;
if (reliable)
{
sequence = _nextSequence;
_nextSequence = checked(_nextSequence + 1);
}
else if (sequenced)
{
if (!_nextSequencedByChannel.TryGetValue(channelId, out sequence))
{
sequence = 1;
}
_nextSequencedByChannel[channelId] = checked(sequence + 1);
}
else
{
sequence = 0;
}
var (ack, ackMask) = AckState();
return new CultNetRudpPacket
{
@ -1563,17 +1596,6 @@ namespace GameCult.Networking
return delivered;
}
private IReadOnlyList<CultNetRudpDeliveredFrame> DrainUnblockedOrdered()
{
var delivered = new List<CultNetRudpDeliveredFrame>();
foreach (var channelId in _orderedBuffers.Keys.ToArray())
{
SkipReceivedNonChannelSequences(channelId);
delivered.AddRange(DrainOrdered(channelId));
}
return delivered;
}
private void SkipReceivedNonChannelSequences(string channelId)
{
if (!_orderedNextSequenceByChannel.TryGetValue(channelId, out var next))

118
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -889,6 +889,75 @@ namespace GameCult.Networking.Tests
Assert.That(client.Connected, Is.False);
}
[Test]
public void RudpSession_LossyPacketsCannotCreateReliableOrderedGaps()
{
var client = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 198, InitialSequence = 1 });
var server = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 198, InitialSequence = 100 });
var connect = client.CreateConnect(0);
var accept = server.AcceptConnect(connect, 1);
client.Receive(accept, 2);
var ping = client.CreatePing(Encoding.UTF8.GetBytes("pulse"));
var pong = server.Receive(ping, 3).Reply!;
var realtime = server.Send("realtime", Encoding.UTF8.GetBytes("discarded realtime"));
var latest = server.Send(
"latest",
Encoding.UTF8.GetBytes("discarded latest state"),
new CultNetRudpSendOptions { Sequenced = true });
var schema = server.Send(
"schema",
Encoding.UTF8.GetBytes("committed response"),
new CultNetRudpSendOptions { Reliable = true, Ordered = true });
Assert.That(ping.Sequence, Is.Zero);
Assert.That(pong.Sequence, Is.Zero);
Assert.That(realtime.Sequence, Is.Zero);
Assert.That(latest.Sequence, Is.EqualTo(1));
Assert.That(schema.Sequence, Is.EqualTo(accept.Sequence + 1));
Assert.That(
client.Receive(schema, 4).Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray(),
Is.EqualTo(new[] { "committed response" }));
}
[Test]
public void RudpSession_UnreliableSequencedDeliveryIsScopedToItsChannel()
{
var sender = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 197, InitialSequence = 50 });
var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 197, InitialSequence = 100 });
var connect = sender.CreateConnect(0);
var accept = receiver.AcceptConnect(connect, 1);
sender.Receive(accept, 2);
var older = sender.Send("latest", Encoding.UTF8.GetBytes("older"), new CultNetRudpSendOptions { Sequenced = true });
var newer = sender.Send("latest", Encoding.UTF8.GetBytes("newer"), new CultNetRudpSendOptions { Sequenced = true });
Assert.That(older.Sequence, Is.EqualTo(1));
Assert.That(newer.Sequence, Is.EqualTo(2));
Assert.That(
receiver.Receive(newer, 3).Delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)).ToArray(),
Is.EqualTo(new[] { "newer" }));
Assert.That(receiver.Receive(older, 4).Delivered, Is.Empty);
}
[Test]
public void RudpSession_RejectsUnreliableOrderedDelivery()
{
var session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = 196 });
session.Receive(new CultNetRudpPacket
{
PacketType = CultNetRudpPacketType.Accept,
ConnectionId = 196,
Sequence = 50,
ChannelId = "control"
});
var error = Assert.Throws<InvalidOperationException>(() =>
session.Send(
"schema",
Encoding.UTF8.GetBytes("cannot order what will not retransmit"),
new CultNetRudpSendOptions { Ordered = true }));
Assert.That(error!.Message, Does.Contain("ordered delivery requires reliability"));
}
[Test]
public void RudpSession_BoundsPendingReliablePacketsBeforeEnqueue()
{
@ -955,55 +1024,6 @@ namespace GameCult.Networking.Tests
Is.EqualTo(new[] { "second", "third" }));
}
[Test]
public void RudpSession_SequencedControlPacketReleasesBufferedOrderedFrame()
{
var receiver = new CultNetRudpSession(new CultNetRudpSessionOptions
{
ConnectionId = 199,
InitialSequence = 1
});
receiver.Receive(new CultNetRudpPacket
{
PacketType = CultNetRudpPacketType.Accept,
ConnectionId = 199,
Sequence = 100,
ChannelId = "control"
});
var first = receiver.Receive(new CultNetRudpPacket
{
PacketType = CultNetRudpPacketType.Data,
ConnectionId = 199,
Sequence = 101,
ChannelId = "schema",
Reliable = true,
Ordered = true,
Payload = Encoding.UTF8.GetBytes("first")
});
var buffered = receiver.Receive(new CultNetRudpPacket
{
PacketType = CultNetRudpPacketType.Data,
ConnectionId = 199,
Sequence = 103,
ChannelId = "schema",
Reliable = true,
Ordered = true,
Payload = Encoding.UTF8.GetBytes("second")
});
var released = receiver.Receive(new CultNetRudpPacket
{
PacketType = CultNetRudpPacketType.Pong,
ConnectionId = 199,
Sequence = 102,
ChannelId = "control"
});
Assert.That(first.Delivered, Has.Count.EqualTo(1));
Assert.That(buffered.Delivered, Is.Empty);
Assert.That(released.Delivered, Has.Count.EqualTo(1));
Assert.That(Encoding.UTF8.GetString(released.Delivered[0].Payload), Is.EqualTo("second"));
}
[Test]
public void RudpSession_SkipsControlPacketsWhileOrderingSchemaPayloads()
{