From 49142a78e5520484b30c5be0fcc81ae9dca1a98f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 11:54:54 +0100 Subject: [PATCH] Add transport-aware server peer context --- docs/cultnet-transport-parity.md | 5 +- src/GameCult.Mesh/CultMeshPeerExchange.cs | 7 +- src/GameCult.Mesh/CultMeshVerseDiscovery.cs | 7 +- .../CultNetDatabaseServer.cs | 56 +++++++++++---- .../CultNetSimulationObservationServer.cs | 9 ++- src/GameCult.Networking/Server.cs | 69 ++++++++++++++++--- 6 files changed, 124 insertions(+), 29 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index d831fd1a..e81fa375 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -156,8 +156,9 @@ Current progress: well. Inbound payload classification and transfer stats also belong to that channel-aware adapter before dispatching schema-v0 or legacy messages. The C# database, simulation-observation, Verse-discovery, and peer-exchange - service wrappers now send responses through `Server.SendCultNet(...)`, so - server-owned service bodies no longer reach around the per-peer adapter with + 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. - TypeScript has the first narrow transport connection port: `TcpFramedTransportConnection` owns length-prefixed frame delivery, exposes diff --git a/src/GameCult.Mesh/CultMeshPeerExchange.cs b/src/GameCult.Mesh/CultMeshPeerExchange.cs index d0c1d3a1..8c8a81ed 100644 --- a/src/GameCult.Mesh/CultMeshPeerExchange.cs +++ b/src/GameCult.Mesh/CultMeshPeerExchange.cs @@ -281,7 +281,12 @@ namespace GameCult.Mesh private Task HandleRequestAsync(CultMeshPeerExchangeRequestMessage request, NetPeer peer) { - _server.SendCultNet(peer, CreateResponse(request)); + return HandleRequestAsync(request, _server.GetPeerContext(peer)); + } + + private Task HandleRequestAsync(CultMeshPeerExchangeRequestMessage request, CultNetServerPeer peer) + { + peer.SendCultNet(CreateResponse(request)); return Task.CompletedTask; } } diff --git a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs index 7a845492..2cd7e5d0 100644 --- a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs +++ b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs @@ -127,7 +127,12 @@ namespace GameCult.Mesh private Task HandleRequestAsync(CultMeshVerseCatalogRequestMessage request, NetPeer peer) { - _server.SendCultNet(peer, CreateResponse(request)); + return HandleRequestAsync(request, _server.GetPeerContext(peer)); + } + + private Task HandleRequestAsync(CultMeshVerseCatalogRequestMessage request, CultNetServerPeer peer) + { + peer.SendCultNet(CreateResponse(request)); return Task.CompletedTask; } } diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index a9844f62..550be8c4 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -209,31 +209,51 @@ namespace GameCult.Networking private Task HandleSnapshotRequestAsync(CultNetSnapshotRequestMessage request, NetPeer peer) { - _server.SendCultNet(peer, CreateSnapshotResponse(request)); + 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) { - _server.SendCultNet(peer, CreateShardCatalogResponse(request)); + 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 { - _server.SendCultNet(peer, CreateShardLogResponse(request)); + peer.SendCultNet(CreateShardLogResponse(request)); } catch (Exception ex) { - _server.SendCultNet(peer, new CultNetErrorMessage { Error = ex.Message }); + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); } 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 { @@ -244,17 +264,22 @@ namespace GameCult.Networking if (!await TryForwardPutAsync(ex, message).ConfigureAwait(false)) { _server.Logger.LogError($"CultNet raw put authority failed: {ex.Message}"); - _server.SendCultNet(peer, CreateRoutingError(ex)); + peer.SendCultNet(CreateRoutingError(ex)); } } catch (Exception ex) { _server.Logger.LogError($"CultNet raw put failed: {ex.Message}"); - _server.SendCultNet(peer, new CultNetErrorMessage { Error = ex.Message }); + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); } } 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 { @@ -264,27 +289,32 @@ namespace GameCult.Networking { if (!await TryForwardDeleteAsync(ex, message).ConfigureAwait(false)) { - _server.SendCultNet(peer, CreateRoutingError(ex)); + peer.SendCultNet(CreateRoutingError(ex)); } } catch (Exception ex) { - _server.SendCultNet(peer, new CultNetErrorMessage { Error = ex.Message }); + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); } } 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) ? message.MessageId : message.SubscriptionId; if (string.IsNullOrWhiteSpace(subscriptionId)) { - _server.SendCultNet(peer, new CultNetErrorMessage { Error = "Database subscription requires a subscriptionId or messageId." }); + peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription requires a subscriptionId or messageId." }); return Task.CompletedTask; } - var key = SubscriptionKey(peer, subscriptionId); + var key = SubscriptionKey(peer.Peer, subscriptionId); _subscriptions.AddOrUpdate( key, _ => CreateSubscription(message, subscriptionId, peer), @@ -296,7 +326,7 @@ namespace GameCult.Networking if (message.IncludeSnapshot) { - _server.SendCultNet(peer, CreateSnapshotResponse(new CultNetSnapshotRequestMessage + peer.SendCultNet(CreateSnapshotResponse(new CultNetSnapshotRequestMessage { MessageId = message.MessageId, SchemaIds = message.SchemaIds, @@ -324,14 +354,14 @@ namespace GameCult.Networking private IDisposable CreateSubscription( CultNetDatabaseSubscribeMessage request, string subscriptionId, - NetPeer peer) + CultNetServerPeer peer) { return _database.WatchAllChanges().Subscribe(change => { var outbound = CreateChangeMessage(change, subscriptionId, request); if (outbound != null) { - _server.SendCultNet(peer, outbound); + peer.SendCultNet(outbound); } }); } diff --git a/src/GameCult.Networking/CultNetSimulationObservationServer.cs b/src/GameCult.Networking/CultNetSimulationObservationServer.cs index 307493c1..45084ff8 100644 --- a/src/GameCult.Networking/CultNetSimulationObservationServer.cs +++ b/src/GameCult.Networking/CultNetSimulationObservationServer.cs @@ -60,17 +60,22 @@ namespace GameCult.Networking } private Task HandleObservationAsync(CultNetSimulationObservationMessage message, NetPeer peer) + { + return HandleObservationAsync(message, _server.GetPeerContext(peer)); + } + + private Task HandleObservationAsync(CultNetSimulationObservationMessage message, CultNetServerPeer peer) { try { foreach (var candidate in CreateCandidateMessages(message)) { - _server.SendCultNet(peer, candidate); + peer.SendCultNet(candidate); } } catch (Exception ex) { - _server.SendCultNet(peer, new CultNetErrorMessage { Error = ex.Message }); + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); } return Task.CompletedTask; diff --git a/src/GameCult.Networking/Server.cs b/src/GameCult.Networking/Server.cs index 3315750e..4eb643d8 100644 --- a/src/GameCult.Networking/Server.cs +++ b/src/GameCult.Networking/Server.cs @@ -193,6 +193,19 @@ namespace GameCult.Networking SendSchema(peer, message); } + /// + /// Gets the transport-aware server peer context for a connected LiteNetLib peer. + /// + public CultNetServerPeer GetPeerContext(NetPeer peer) + { + if (peer == null) throw new ArgumentNullException(nameof(peer)); + var user = _users.GetOrAdd( + peer.Id, + _ => new User { Peer = peer, Transport = new LiteNetLibTransportConnection(peer) }); + user.Transport ??= new LiteNetLibTransportConnection(peer); + return new CultNetServerPeer(peer, user.Transport); + } + /// /// Removes a previously registered listener for a specific message type. /// @@ -662,11 +675,7 @@ namespace GameCult.Networking return; } - var user = _users.GetOrAdd( - peer.Id, - _ => new User { Peer = peer, Transport = new LiteNetLibTransportConnection(peer) }); - user.Transport ??= new LiteNetLibTransportConnection(peer); - user.Transport.SendLegacy(message); + GetPeerContext(peer).Send(message); } private void SendSchema(NetPeer? peer, T message) @@ -677,11 +686,7 @@ namespace GameCult.Networking return; } - var user = _users.GetOrAdd( - peer.Id, - _ => new User { Peer = peer, Transport = new LiteNetLibTransportConnection(peer) }); - user.Transport ??= new LiteNetLibTransportConnection(peer); - user.Transport.SendSchema(message); + GetPeerContext(peer).SendCultNet(message); } private static bool CheckRateLimit( @@ -708,6 +713,50 @@ namespace GameCult.Networking } } + /// + /// Transport-aware server peer context for built-in CultNet service bodies. + /// + public sealed class CultNetServerPeer + { + internal CultNetServerPeer(NetPeer peer, LiteNetLibTransportConnection transport) + { + Peer = peer ?? throw new ArgumentNullException(nameof(peer)); + Transport = transport ?? throw new ArgumentNullException(nameof(transport)); + } + + /// + /// Gets the underlying LiteNetLib peer identity. + /// + public NetPeer Peer { get; } + + /// + /// Gets the channel-aware LiteNetLib transport adapter for this peer. + /// + public LiteNetLibTransportConnection Transport { get; } + + /// + /// Gets a snapshot of transport counters for this peer. + /// + public CultNetTransportStats Stats => Transport.Stats; + + /// + /// Sends a legacy GameCult.Networking union message through the peer transport adapter. + /// + public void Send(Message message) + { + Transport.SendLegacy(message); + } + + /// + /// Sends a CultNet schema-v0 message through the peer transport adapter. + /// + public void SendCultNet(T message) + where T : ICultNetSchemaMessage + { + Transport.SendSchema(message); + } + } + /// /// Represents an active authenticated session. ///