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

Add C# RUDP schema client adoption path

Browse source
This commit is contained in:
Meta Vens 2026-06-17 13:27:56 +01:00
commit 114fb86b0f
6 changed files with 370 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

6
docs/cultnet-transport-parity.md
View file

@ -162,7 +162,11 @@ Current progress:
service wrappers now enter their built-in handlers through
`CultNetServerPeer`, a transport-aware server peer context. Responses send
through that context instead of reaching around the per-peer adapter with
direct `NetPeer` extension sends.
direct `NetPeer` extension sends. The C# schema-client port now also has an
RUDP adapter; shard log fetch, shard snapshot fetch, and shard write
forwarding select it for `rudp://` primary endpoints and keep LiteNetLib for
`cultnet://` endpoints, so those service clients no longer need custom
injection to leave the LiteNetLib lane.
- TypeScript has the first narrow transport connection port:
`TcpFramedTransportConnection` owns length-prefixed frame delivery, exposes
frame/close/error events, `send(channel, payload)`, `close`, and transfer

6
docs/runtime-parity-scope.md
View file

@ -73,8 +73,10 @@ server body, is deliberately not made.
`NetPeer` callbacks, and TypeScript TCP peer creation has a transport-first
helper. C# Verse discovery, peer exchange, shard-log fetch, shard-snapshot
fetch, and shard write forwarding clients now consume a schema-client
transport port with the LiteNetLib client as the default adapter; broader
daemon adoption remains the unclaimed layer.
transport port; shard log fetch, shard snapshot fetch, and shard write
forwarding now select the C# RUDP schema client for `rudp://` endpoints while
keeping LiteNetLib for `cultnet://` endpoints. Broader daemon adoption remains
the unclaimed layer.
- Extend RUDP adoption through daemon/socket reconnect loops. RUDP profiles now
advertise the portable reconnect policy, runtime-local tests prove the shared
controller across the targeted runtimes, and the C# LiteNetLib client plus

202
src/GameCult.Networking/CultNetSchemaClient.cs
View file

@ -1,4 +1,11 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
namespace GameCult.Networking
{
@ -81,6 +88,157 @@ namespace GameCult.Networking
}
}
/// <summary>
/// RUDP-backed implementation of the CultNet schema client port.
/// </summary>
public sealed class RudpCultNetSchemaClient : ICultNetSchemaClient
{
private readonly string _runtimeId;
private readonly uint _connectionId;
private readonly string _connectPayload;
private readonly int _maxFragmentBytes;
private readonly long _resendDelayMs;
private readonly ConcurrentDictionary<Type, List<Delegate>> _handlers = new();
private readonly object _handlerLock = new();
private CultNetRudpSocketTransportConnection? _transport;
private Thread? _pumpThread;
private volatile bool _disposed;
/// <summary>
/// Creates a RUDP schema client adapter.
/// </summary>
public RudpCultNetSchemaClient(
string runtimeId = "csharp-rudp-schema-client",
uint connectionId = 0x43554c54,
string connectPayload = "cultnet-schema-rudp",
int maxFragmentBytes = 1024,
long resendDelayMs = 25)
{
if (string.IsNullOrWhiteSpace(runtimeId)) throw new ArgumentException("Runtime id is required.", nameof(runtimeId));
_runtimeId = runtimeId;
_connectionId = connectionId;
_connectPayload = connectPayload;
_maxFragmentBytes = maxFragmentBytes;
_resendDelayMs = resendDelayMs;
}
/// <inheritdoc />
public bool Connected => _transport?.Connected == true;
/// <inheritdoc />
public void Connect(string host, int port)
{
if (_disposed) throw new ObjectDisposedException(nameof(RudpCultNetSchemaClient));
if (string.IsNullOrWhiteSpace(host)) throw new ArgumentException("Host is required.", nameof(host));
if (port <= 0 || port > 65535) throw new ArgumentOutOfRangeException(nameof(port));
var remoteAddress = IPAddress.TryParse(host, out var parsed)
? parsed
: Dns.GetHostAddresses(host)[0];
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
socket.Bind(new IPEndPoint(IPAddress.Any, 0));
socket.ReceiveTimeout = 20;
_transport = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = _runtimeId,
Mode = CultNetRudpSocketMode.Client,
Socket = socket,
RemoteEndPoint = new IPEndPoint(remoteAddress, port),
ConnectionId = _connectionId,
TransportId = "schema-rudp",
MaxFragmentBytes = _maxFragmentBytes,
ResendDelayMs = _resendDelayMs
});
_transport.Connect(Encoding.UTF8.GetBytes(_connectPayload));
_pumpThread = new Thread(Pump)
{
IsBackground = true,
Name = "CultNet RUDP schema client pump"
};
_pumpThread.Start();
}
/// <inheritdoc />
public void SendCultNet<T>(T message)
where T : ICultNetSchemaMessage
{
if (message == null) throw new ArgumentNullException(nameof(message));
var transport = _transport ?? throw new InvalidOperationException("RUDP schema client is not connected.");
transport.SendSchemaMessage(message);
}
/// <inheritdoc />
public void OnCultNet<T>(Action<T> callback)
where T : ICultNetSchemaMessage
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
lock (_handlerLock)
{
var handlers = _handlers.GetOrAdd(typeof(T), _ => new List<Delegate>());
handlers.Add(callback);
}
}
/// <inheritdoc />
public void Dispose()
{
_disposed = true;
_transport?.Dispose();
}
private void Pump()
{
while (!_disposed)
{
var transport = _transport;
if (transport == null)
{
return;
}
ICultNetSchemaMessage? message;
try
{
message = transport.ReceiveSchemaMessageOnce();
transport.PollResends();
}
catch (ObjectDisposedException) when (_disposed)
{
return;
}
catch (SocketException) when (_disposed)
{
return;
}
if (message != null)
{
Dispatch(message);
}
Thread.Sleep(5);
}
}
private void Dispatch(ICultNetSchemaMessage message)
{
List<Delegate>? handlers;
lock (_handlerLock)
{
if (!_handlers.TryGetValue(message.GetType(), out var registered))
{
return;
}
handlers = registered.ToList();
}
foreach (var handler in handlers)
{
handler.DynamicInvoke(message);
}
}
}
/// <summary>
/// Factory helpers for schema client adapters.
/// </summary>
@ -95,5 +253,49 @@ namespace GameCult.Networking
{
return new LiteNetLibCultNetSchemaClient(security, configureClient);
}
/// <summary>
/// Creates the C# RUDP schema client adapter.
/// </summary>
public static ICultNetSchemaClient CreateRudp(
string runtimeId = "csharp-rudp-schema-client",
uint connectionId = 0x43554c54,
string connectPayload = "cultnet-schema-rudp",
int maxFragmentBytes = 1024,
long resendDelayMs = 25)
{
return new RudpCultNetSchemaClient(runtimeId, connectionId, connectPayload, maxFragmentBytes, resendDelayMs);
}
/// <summary>
/// Creates the default schema client adapter for an advertised endpoint URI.
/// </summary>
public static ICultNetSchemaClient CreateForEndpoint(
string endpoint,
ClientSecurityOptions? security = null,
Action<Client>? configureClient = null)
{
if (string.IsNullOrWhiteSpace(endpoint))
{
throw new ArgumentException("Endpoint must be non-empty.", nameof(endpoint));
}
if (!Uri.TryCreate(endpoint, UriKind.Absolute, out var uri))
{
throw new FormatException($"CultNet endpoint '{endpoint}' must be an absolute URI.");
}
if (string.Equals(uri.Scheme, "rudp", StringComparison.OrdinalIgnoreCase))
{
return CreateRudp();
}
if (string.Equals(uri.Scheme, "cultnet", StringComparison.OrdinalIgnoreCase))
{
return CreateLiteNetLib(security, configureClient);
}
throw new FormatException($"CultNet endpoint '{endpoint}' must use cultnet://host:port or rudp://host:port.");
}
}
}

18
src/GameCult.Networking/CultNetShardReplication.cs
View file

@ -395,7 +395,8 @@ namespace GameCult.Networking
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// Gets or sets the schema-v0 client factory. Defaults to an endpoint-aware adapter:
/// rudp:// endpoints use the C# RUDP schema client and cultnet:// endpoints use LiteNetLib.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
@ -428,7 +429,7 @@ namespace GameCult.Networking
var completion = new TaskCompletionSource<CultNetShardLogResponseMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = CreateClient();
using var client = CreateClient(endpoint);
client.OnCultNet<CultNetShardLogResponseMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
@ -453,10 +454,10 @@ namespace GameCult.Networking
return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false);
}
private ICultNetSchemaClient CreateClient()
private ICultNetSchemaClient CreateClient(string endpoint)
{
return _options.CreateClient?.Invoke()
?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient);
?? CultNetSchemaClients.CreateForEndpoint(endpoint, _options.Security, _options.ConfigureClient);
}
private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint)
@ -521,7 +522,8 @@ namespace GameCult.Networking
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// Gets or sets the schema-v0 client factory. Defaults to an endpoint-aware adapter:
/// rudp:// endpoints use the C# RUDP schema client and cultnet:// endpoints use LiteNetLib.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
@ -551,7 +553,7 @@ namespace GameCult.Networking
var completion = new TaskCompletionSource<CultNetSnapshotResponseRawMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
using var client = CreateClient();
using var client = CreateClient(endpoint);
client.OnCultNet<CultNetSnapshotResponseRawMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
@ -575,10 +577,10 @@ namespace GameCult.Networking
return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false);
}
private ICultNetSchemaClient CreateClient()
private ICultNetSchemaClient CreateClient(string endpoint)
{
return _options.CreateClient?.Invoke()
?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient);
?? CultNetSchemaClients.CreateForEndpoint(endpoint, _options.Security, _options.ConfigureClient);
}
private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint)

19
src/GameCult.Networking/CultNetShardWriteForwarding.cs
View file

@ -57,7 +57,8 @@ namespace GameCult.Networking
public Action<Client>? ConfigureClient { get; set; }
/// <summary>
/// Gets or sets the schema-v0 client factory. Defaults to the C# LiteNetLib adapter.
/// Gets or sets the schema-v0 client factory. Defaults to an endpoint-aware adapter:
/// rudp:// endpoints use the C# RUDP schema client and cultnet:// endpoints use LiteNetLib.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
}
@ -108,10 +109,10 @@ namespace GameCult.Networking
}
if (!Uri.TryCreate(endpoint, UriKind.Absolute, out var uri) ||
!string.Equals(uri.Scheme, "cultnet", StringComparison.OrdinalIgnoreCase) ||
!IsSupportedSchemaEndpointScheme(uri.Scheme) ||
string.IsNullOrWhiteSpace(uri.Host))
{
throw new FormatException($"CultNet endpoint '{endpoint}' must use cultnet://host:port.");
throw new FormatException($"CultNet endpoint '{endpoint}' must use cultnet://host:port or rudp://host:port.");
}
return (uri.Host, uri.IsDefaultPort ? 3075 : uri.Port);
@ -120,16 +121,22 @@ namespace GameCult.Networking
private async Task SendAsync<T>(string endpoint, T message) where T : ICultNetSchemaMessage
{
var (host, port) = ParseEndpoint(endpoint);
using var client = CreateClient();
using var client = CreateClient(endpoint);
client.Connect(host, port);
await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false);
client.SendCultNet(message);
}
private ICultNetSchemaClient CreateClient()
private ICultNetSchemaClient CreateClient(string endpoint)
{
return _options.CreateClient?.Invoke()
?? CultNetSchemaClients.CreateLiteNetLib(_options.Security, _options.ConfigureClient);
?? CultNetSchemaClients.CreateForEndpoint(endpoint, _options.Security, _options.ConfigureClient);
}
private static bool IsSupportedSchemaEndpointScheme(string scheme)
{
return string.Equals(scheme, "cultnet", StringComparison.OrdinalIgnoreCase) ||
string.Equals(scheme, "rudp", StringComparison.OrdinalIgnoreCase);
}
private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint)

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

@ -85,6 +85,44 @@ namespace GameCult.Networking.Tests
throw new InvalidOperationException("Unreachable.");
}
private static async Task WaitUntilAsync(Func<bool> predicate, TimeSpan timeout)
{
var deadline = DateTimeOffset.UtcNow + timeout;
while (DateTimeOffset.UtcNow < deadline)
{
if (predicate())
{
return;
}
await Task.Delay(TimeSpan.FromMilliseconds(5));
}
Assert.Fail("Condition was not satisfied before timeout.");
}
private static async Task<T> AwaitWithTimeout<T>(Task<T> task, TimeSpan timeout)
{
var completed = await Task.WhenAny(task, Task.Delay(timeout));
if (completed != task)
{
Assert.Fail("Task did not complete before timeout.");
}
return await task;
}
private static async Task AwaitWithTimeout(Task task, TimeSpan timeout)
{
var completed = await Task.WhenAny(task, Task.Delay(timeout));
if (completed != task)
{
Assert.Fail("Task did not complete before timeout.");
}
await task;
}
[Test]
public void EncryptDecrypt_Roundtrip()
{
@ -1692,6 +1730,15 @@ namespace GameCult.Networking.Tests
Assert.That(parsed.Port, Is.EqualTo(3075));
}
[Test]
public void CultNetSchemaWriteForwarder_Parses_RudpEndpoints()
{
var parsed = CultNetSchemaWriteForwarder.ParseEndpoint("rudp://primary.example.test:5075");
Assert.That(parsed.Host, Is.EqualTo("primary.example.test"));
Assert.That(parsed.Port, Is.EqualTo(5075));
}
[Test]
public void CultNetSchemaWriteForwarder_Rejects_InvalidEndpoint()
{
@ -1700,6 +1747,95 @@ namespace GameCult.Networking.Tests
Throws.TypeOf<FormatException>());
}
[Test]
public void CultNetSchemaClients_CreateForEndpoint_Selects_RudpClient()
{
using var rudp = CultNetSchemaClients.CreateForEndpoint("rudp://127.0.0.1:5075");
using var liteNetLib = CultNetSchemaClients.CreateForEndpoint(
"cultnet://127.0.0.1:3075",
DevelopmentClientSecurity);
Assert.That(rudp, Is.TypeOf<RudpCultNetSchemaClient>());
Assert.That(liteNetLib, Is.TypeOf<LiteNetLibCultNetSchemaClient>());
}
[Test]
public async Task RudpCultNetSchemaClient_Carries_SchemaMessages()
{
using var serverSocket = BindUdpSocket();
using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = "csharp-rudp-schema-server",
Mode = CultNetRudpSocketMode.Server,
Socket = serverSocket,
ConnectionId = 0x43554c54,
InitialSequence = 100,
TransportId = "schema-rudp",
ResendDelayMs = 25,
MaxFragmentBytes = 1024
});
var serverDone = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var serverThread = new Thread(() =>
{
try
{
while (!serverDone.Task.IsCompleted)
{
var request = server.ReceiveSchemaMessageOnce<CultNetSchemaCatalogRequestMessage>();
server.PollResends();
if (request == null)
{
Thread.Sleep(5);
continue;
}
server.SendSchemaMessage(new CultNetSchemaCatalogResponseMessage
{
MessageId = request.MessageId,
Schemas =
[
new CultNetSchemaDescriptor
{
SchemaId = "rudp.schema.test",
Kind = "wire_message",
SchemaVersion = "rudp.schema.test.v0",
WireContracts = ["cultnet.schema.v0"]
}
]
});
serverDone.TrySetResult();
return;
}
}
catch (Exception error)
{
serverDone.TrySetException(error);
}
})
{
IsBackground = true
};
serverThread.Start();
using var client = CultNetSchemaClients.CreateRudp(runtimeId: "csharp-rudp-schema-client");
var responseCompletion = new TaskCompletionSource<CultNetSchemaCatalogResponseMessage>(
TaskCreationOptions.RunContinuationsAsynchronously);
client.OnCultNet<CultNetSchemaCatalogResponseMessage>(response => responseCompletion.TrySetResult(response));
client.Connect("127.0.0.1", ((IPEndPoint)serverSocket.LocalEndPoint!).Port);
await WaitUntilAsync(() => client.Connected, TimeSpan.FromSeconds(2));
client.SendCultNet(new CultNetSchemaCatalogRequestMessage
{
MessageId = "rudp-schema-client-test",
IncludeSchemaJson = false
});
var response = await AwaitWithTimeout(responseCompletion.Task, TimeSpan.FromSeconds(2));
await AwaitWithTimeout(serverDone.Task, TimeSpan.FromSeconds(2));
Assert.That(response.MessageId, Is.EqualTo("rudp-schema-client-test"));
Assert.That(response.Schemas.Single().SchemaId, Is.EqualTo("rudp.schema.test"));
}
[Test]
public void CultMesh_CreateClient_Returns_CultNetClient()
{