diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index cc3b2a6c..579a1878 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -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 diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 57485a4e..895312db 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -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 diff --git a/src/GameCult.Networking/CultNetSchemaClient.cs b/src/GameCult.Networking/CultNetSchemaClient.cs index eb113a2a..7cc87537 100644 --- a/src/GameCult.Networking/CultNetSchemaClient.cs +++ b/src/GameCult.Networking/CultNetSchemaClient.cs @@ -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 } } + /// + /// RUDP-backed implementation of the CultNet schema client port. + /// + 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> _handlers = new(); + private readonly object _handlerLock = new(); + private CultNetRudpSocketTransportConnection? _transport; + private Thread? _pumpThread; + private volatile bool _disposed; + + /// + /// Creates a RUDP schema client adapter. + /// + 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; + } + + /// + public bool Connected => _transport?.Connected == true; + + /// + 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(); + } + + /// + public void SendCultNet(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); + } + + /// + public void OnCultNet(Action callback) + where T : ICultNetSchemaMessage + { + if (callback == null) throw new ArgumentNullException(nameof(callback)); + lock (_handlerLock) + { + var handlers = _handlers.GetOrAdd(typeof(T), _ => new List()); + handlers.Add(callback); + } + } + + /// + 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? handlers; + lock (_handlerLock) + { + if (!_handlers.TryGetValue(message.GetType(), out var registered)) + { + return; + } + handlers = registered.ToList(); + } + + foreach (var handler in handlers) + { + handler.DynamicInvoke(message); + } + } + } + /// /// Factory helpers for schema client adapters. /// @@ -95,5 +253,49 @@ namespace GameCult.Networking { return new LiteNetLibCultNetSchemaClient(security, configureClient); } + + /// + /// Creates the C# RUDP schema client adapter. + /// + 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); + } + + /// + /// Creates the default schema client adapter for an advertised endpoint URI. + /// + public static ICultNetSchemaClient CreateForEndpoint( + string endpoint, + ClientSecurityOptions? security = null, + Action? 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."); + } } } diff --git a/src/GameCult.Networking/CultNetShardReplication.cs b/src/GameCult.Networking/CultNetShardReplication.cs index 1d81ebbe..88eaac7a 100644 --- a/src/GameCult.Networking/CultNetShardReplication.cs +++ b/src/GameCult.Networking/CultNetShardReplication.cs @@ -395,7 +395,8 @@ namespace GameCult.Networking public Action? ConfigureClient { get; set; } /// - /// 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. /// public Func? CreateClient { get; set; } } @@ -428,7 +429,7 @@ namespace GameCult.Networking var completion = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); - using var client = CreateClient(); + using var client = CreateClient(endpoint); client.OnCultNet(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? ConfigureClient { get; set; } /// - /// 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. /// public Func? CreateClient { get; set; } } @@ -551,7 +553,7 @@ namespace GameCult.Networking var completion = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); - using var client = CreateClient(); + using var client = CreateClient(endpoint); client.OnCultNet(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) diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs index 686621be..44cb4582 100644 --- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs +++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs @@ -57,7 +57,8 @@ namespace GameCult.Networking public Action? ConfigureClient { get; set; } /// - /// 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. /// public Func? 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(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) diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 0be77bed..e116a4aa 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -85,6 +85,44 @@ namespace GameCult.Networking.Tests throw new InvalidOperationException("Unreachable."); } + private static async Task WaitUntilAsync(Func 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 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."); + } + + 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()); } + [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()); + Assert.That(liteNetLib, Is.TypeOf()); + } + + [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(); + 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( + TaskCreationOptions.RunContinuationsAsynchronously); + client.OnCultNet(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() {