From 2453e44f11e7bd6389a5aa17c3ce18c2e448a230 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 16:07:56 +0100 Subject: [PATCH] Add C# transport-aware server listeners --- docs/cultnet-transport-parity.md | 3 +- docs/runtime-parity-scope.md | 11 +-- src/GameCult.Mesh/CultMeshPeerExchange.cs | 8 +- src/GameCult.Mesh/CultMeshVerseDiscovery.cs | 8 +- .../CultNetDatabaseServer.cs | 48 ++--------- .../CultNetSimulationObservationServer.cs | 8 +- src/GameCult.Networking/Server.cs | 83 +++++++++++++++++++ 7 files changed, 103 insertions(+), 66 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 63a3e744..138afe7f 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -367,7 +367,8 @@ Current progress: - The remaining parity work is to finish production service-body adoption behind the shared transport ports, then wire any remaining daemon/socket reconnect loops through the shared controller. C# built-in LiteNetLib service - wrappers now route through `CultNetServerPeer`, TypeScript exposes a + wrappers now register transport-aware `CultNetServerPeer` listeners instead + of requiring each bridge to bounce through `NetPeer`, TypeScript exposes a transport-first TCP peer helper, C# schema-v0 service clients consume `ICultNetSchemaClient`, and the core schema-message RUDP matrix is live in the TypeScript interop gate; broader service adoption remains the unclaimed diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 96901dbc..e7b57afe 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -59,11 +59,12 @@ CultMesh. - Broaden production service adoption so remaining daemon paths use the shared transport port consistently. The C# built-in LiteNetLib service wrappers now - route through transport-aware peer context, 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. + register transport-aware `CultNetServerPeer` listeners instead of requiring + `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. - 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.Mesh/CultMeshPeerExchange.cs b/src/GameCult.Mesh/CultMeshPeerExchange.cs index f53808d2..7a26d892 100644 --- a/src/GameCult.Mesh/CultMeshPeerExchange.cs +++ b/src/GameCult.Mesh/CultMeshPeerExchange.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; using GameCult.Networking; -using LiteNetLib; using R3; namespace GameCult.Mesh @@ -220,7 +219,7 @@ namespace GameCult.Mesh { private readonly Server _server; private readonly CultMeshPeerCatalog _catalog; - private readonly Func _requestHandler; + private readonly Func _requestHandler; private bool _disposed; /// @@ -279,11 +278,6 @@ namespace GameCult.Mesh _server.RemoveCultNetMessageListener(_requestHandler); } - private Task HandleRequestAsync(CultMeshPeerExchangeRequestMessage request, NetPeer peer) - { - return HandleRequestAsync(request, _server.GetPeerContext(peer)); - } - private Task HandleRequestAsync(CultMeshPeerExchangeRequestMessage request, CultNetServerPeer peer) { peer.SendCultNet(CreateResponse(request)); diff --git a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs index d89d3f3d..28d90e35 100644 --- a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs +++ b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; using GameCult.Networking; -using LiteNetLib; namespace GameCult.Mesh { @@ -74,7 +73,7 @@ namespace GameCult.Mesh { private readonly Server _server; private readonly CultMeshVerseCatalog _catalog; - private readonly Func _requestHandler; + private readonly Func _requestHandler; private bool _disposed; /// @@ -125,11 +124,6 @@ namespace GameCult.Mesh _server.RemoveCultNetMessageListener(_requestHandler); } - private Task HandleRequestAsync(CultMeshVerseCatalogRequestMessage request, NetPeer peer) - { - return HandleRequestAsync(request, _server.GetPeerContext(peer)); - } - private Task HandleRequestAsync(CultMeshVerseCatalogRequestMessage request, CultNetServerPeer peer) { peer.SendCultNet(CreateResponse(request)); diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 550be8c4..df8dbbbf 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -17,13 +17,13 @@ namespace GameCult.Networking private readonly Server _server; private readonly CultNetDatabase _database; private readonly CultNetDatabaseServerOptions _options; - private readonly Func _snapshotHandler; - private readonly Func _putHandler; - private readonly Func _deleteHandler; - private readonly Func _shardCatalogHandler; - private readonly Func _shardLogHandler; - private readonly Func _subscribeHandler; - private readonly Func _unsubscribeHandler; + private readonly Func _snapshotHandler; + private readonly Func _putHandler; + private readonly Func _deleteHandler; + private readonly Func _shardCatalogHandler; + private readonly Func _shardLogHandler; + private readonly Func _subscribeHandler; + private readonly Func _unsubscribeHandler; private readonly ConcurrentDictionary _subscriptions = new(StringComparer.Ordinal); private bool _disposed; @@ -207,33 +207,18 @@ namespace GameCult.Networking _subscriptions.Clear(); } - private Task HandleSnapshotRequestAsync(CultNetSnapshotRequestMessage request, NetPeer peer) - { - return HandleSnapshotRequestAsync(request, _server.GetPeerContext(peer)); - } - private Task HandleSnapshotRequestAsync(CultNetSnapshotRequestMessage request, CultNetServerPeer peer) { peer.SendCultNet(CreateSnapshotResponse(request)); return Task.CompletedTask; } - private Task HandleShardCatalogRequestAsync(CultNetShardCatalogRequestMessage request, NetPeer peer) - { - return HandleShardCatalogRequestAsync(request, _server.GetPeerContext(peer)); - } - private Task HandleShardCatalogRequestAsync(CultNetShardCatalogRequestMessage request, CultNetServerPeer peer) { peer.SendCultNet(CreateShardCatalogResponse(request)); return Task.CompletedTask; } - private Task HandleShardLogRequestAsync(CultNetShardLogRequestMessage request, NetPeer peer) - { - return HandleShardLogRequestAsync(request, _server.GetPeerContext(peer)); - } - private Task HandleShardLogRequestAsync(CultNetShardLogRequestMessage request, CultNetServerPeer peer) { try @@ -248,11 +233,6 @@ namespace GameCult.Networking return Task.CompletedTask; } - private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, NetPeer peer) - { - await HandlePutAsync(message, _server.GetPeerContext(peer)).ConfigureAwait(false); - } - private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, CultNetServerPeer peer) { try @@ -274,11 +254,6 @@ namespace GameCult.Networking } } - private async Task HandleDeleteAsync(CultNetDocumentDeleteMessage message, NetPeer peer) - { - await HandleDeleteAsync(message, _server.GetPeerContext(peer)).ConfigureAwait(false); - } - private async Task HandleDeleteAsync(CultNetDocumentDeleteMessage message, CultNetServerPeer peer) { try @@ -298,11 +273,6 @@ namespace GameCult.Networking } } - private Task HandleSubscribeAsync(CultNetDatabaseSubscribeMessage message, NetPeer peer) - { - return HandleSubscribeAsync(message, _server.GetPeerContext(peer)); - } - private Task HandleSubscribeAsync(CultNetDatabaseSubscribeMessage message, CultNetServerPeer peer) { var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId) @@ -337,13 +307,13 @@ namespace GameCult.Networking return Task.CompletedTask; } - private Task HandleUnsubscribeAsync(CultNetDatabaseUnsubscribeMessage message, NetPeer peer) + private Task HandleUnsubscribeAsync(CultNetDatabaseUnsubscribeMessage message, CultNetServerPeer peer) { var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId) ? message.MessageId : message.SubscriptionId; if (!string.IsNullOrWhiteSpace(subscriptionId) && - _subscriptions.TryRemove(SubscriptionKey(peer, subscriptionId), out var subscription)) + _subscriptions.TryRemove(SubscriptionKey(peer.Peer, subscriptionId), out var subscription)) { subscription.Dispose(); } diff --git a/src/GameCult.Networking/CultNetSimulationObservationServer.cs b/src/GameCult.Networking/CultNetSimulationObservationServer.cs index 45084ff8..dea8d637 100644 --- a/src/GameCult.Networking/CultNetSimulationObservationServer.cs +++ b/src/GameCult.Networking/CultNetSimulationObservationServer.cs @@ -1,7 +1,6 @@ using System; using System.Linq; using System.Threading.Tasks; -using LiteNetLib; namespace GameCult.Networking { @@ -12,7 +11,7 @@ namespace GameCult.Networking { private readonly Server _server; private readonly CultNetSimulationObservationHub _hub; - private readonly Func _observationHandler; + private readonly Func _observationHandler; private bool _disposed; /// @@ -59,11 +58,6 @@ namespace GameCult.Networking _server.RemoveCultNetMessageListener(_observationHandler); } - private Task HandleObservationAsync(CultNetSimulationObservationMessage message, NetPeer peer) - { - return HandleObservationAsync(message, _server.GetPeerContext(peer)); - } - private Task HandleObservationAsync(CultNetSimulationObservationMessage message, CultNetServerPeer peer) { try diff --git a/src/GameCult.Networking/Server.cs b/src/GameCult.Networking/Server.cs index 4eb643d8..a6234301 100644 --- a/src/GameCult.Networking/Server.cs +++ b/src/GameCult.Networking/Server.cs @@ -30,6 +30,7 @@ namespace GameCult.Networking private readonly ConcurrentDictionary _messageDelegates = new(); private readonly ConcurrentDictionary _cultNetMessageDelegates = new(); + private readonly ConcurrentDictionary _cultNetServerPeerMessageDelegates = new(); private readonly ConcurrentDictionary _users = new(); private readonly ConcurrentDictionary> _connectionAttempts = new(); private readonly ConcurrentDictionary _connectionAttemptLocks = new(); @@ -106,6 +107,7 @@ namespace GameCult.Networking { _messageDelegates.Clear(); _cultNetMessageDelegates.Clear(); + _cultNetServerPeerMessageDelegates.Clear(); } /// @@ -160,6 +162,35 @@ namespace GameCult.Networking }); } + /// + /// Adds a transport-aware listener for a modern CultNet schema-v0 message type. + /// + public void AddCultNetMessageListener(Func callback) + where T : ICultNetSchemaMessage + { + var type = typeof(T); + _cultNetServerPeerMessageDelegates.AddOrUpdate(type, + _ => callback, + (t, current) => + { + var combined = Delegate.Combine(current, callback) as Func; + return combined ?? throw new InvalidOperationException($"Failed to combine delegates for {t.Name}"); + }); + } + + /// + /// Adds a transport-aware listener for a modern CultNet schema-v0 message type. + /// + public void AddCultNetMessageListener(Action callback) + where T : ICultNetSchemaMessage + { + AddCultNetMessageListener((message, peer) => + { + callback(message, peer); + return Task.CompletedTask; + }); + } + /// /// Adds a listener for a modern CultNet schema-v0 message type. /// @@ -176,6 +207,24 @@ namespace GameCult.Networking AddCultNetMessageListener(callback); } + /// + /// Adds a transport-aware listener for a modern CultNet schema-v0 message type. + /// + public void OnCultNet(Func callback) + where T : ICultNetSchemaMessage + { + AddCultNetMessageListener(callback); + } + + /// + /// Adds a transport-aware listener for a modern CultNet schema-v0 message type. + /// + public void OnCultNet(Action callback) + where T : ICultNetSchemaMessage + { + AddCultNetMessageListener(callback); + } + /// /// Sends a legacy GameCult.Networking union message through this server's per-peer LiteNetLib adapter. /// @@ -245,6 +294,19 @@ namespace GameCult.Networking _cultNetMessageDelegates[typeof(T)] = newDelegate; } } + + if (_cultNetServerPeerMessageDelegates.TryGetValue(typeof(T), out var currentServerPeerDelegate)) + { + var newDelegate = Delegate.Remove(currentServerPeerDelegate, callback); + if (newDelegate == null) + { + _cultNetServerPeerMessageDelegates.TryRemove(typeof(T), out _); + } + else + { + _cultNetServerPeerMessageDelegates[typeof(T)] = newDelegate; + } + } } /// @@ -434,6 +496,7 @@ namespace GameCult.Networking return; } + var handled = false; if (_cultNetMessageDelegates.TryGetValue(message.GetType(), out var del) && del != null) { foreach (var listener in del.GetInvocationList()) @@ -445,6 +508,26 @@ namespace GameCult.Networking } } + handled = true; + } + + if (_cultNetServerPeerMessageDelegates.TryGetValue(message.GetType(), out var serverPeerDel) && serverPeerDel != null) + { + var serverPeer = GetPeerContext(peer); + foreach (var listener in serverPeerDel.GetInvocationList()) + { + var result = listener.DynamicInvoke(message, serverPeer); + if (result is Task task) + { + await task.ConfigureAwait(false); + } + } + + handled = true; + } + + if (handled) + { user.SessionExpiresAt = DateTimeOffset.UtcNow.AddSeconds(SessionTimeoutSeconds); RefreshSessionIfNeeded(peer, user); }