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

Add C# transport-aware server listeners

Browse source
This commit is contained in:
Meta Vens 2026-06-16 16:07:56 +01:00
commit 2453e44f11
7 changed files with 103 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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

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

@ -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

8
src/GameCult.Mesh/CultMeshPeerExchange.cs
View file

@ -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<CultMeshPeerExchangeRequestMessage, NetPeer, Task> _requestHandler;
private readonly Func<CultMeshPeerExchangeRequestMessage, CultNetServerPeer, Task> _requestHandler;
private bool _disposed;
/// <summary>
@ -279,11 +278,6 @@ namespace GameCult.Mesh
_server.RemoveCultNetMessageListener<CultMeshPeerExchangeRequestMessage>(_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));

8
src/GameCult.Mesh/CultMeshVerseDiscovery.cs
View file

@ -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<CultMeshVerseCatalogRequestMessage, NetPeer, Task> _requestHandler;
private readonly Func<CultMeshVerseCatalogRequestMessage, CultNetServerPeer, Task> _requestHandler;
private bool _disposed;
/// <summary>
@ -125,11 +124,6 @@ namespace GameCult.Mesh
_server.RemoveCultNetMessageListener<CultMeshVerseCatalogRequestMessage>(_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));

48
src/GameCult.Networking/CultNetDatabaseServer.cs
View file

@ -17,13 +17,13 @@ namespace GameCult.Networking
private readonly Server _server;
private readonly CultNetDatabase _database;
private readonly CultNetDatabaseServerOptions _options;
private readonly Func<CultNetSnapshotRequestMessage, NetPeer, Task> _snapshotHandler;
private readonly Func<CultNetDocumentPutRawMessage, NetPeer, Task> _putHandler;
private readonly Func<CultNetDocumentDeleteMessage, NetPeer, Task> _deleteHandler;
private readonly Func<CultNetShardCatalogRequestMessage, NetPeer, Task> _shardCatalogHandler;
private readonly Func<CultNetShardLogRequestMessage, NetPeer, Task> _shardLogHandler;
private readonly Func<CultNetDatabaseSubscribeMessage, NetPeer, Task> _subscribeHandler;
private readonly Func<CultNetDatabaseUnsubscribeMessage, NetPeer, Task> _unsubscribeHandler;
private readonly Func<CultNetSnapshotRequestMessage, CultNetServerPeer, Task> _snapshotHandler;
private readonly Func<CultNetDocumentPutRawMessage, CultNetServerPeer, Task> _putHandler;
private readonly Func<CultNetDocumentDeleteMessage, CultNetServerPeer, Task> _deleteHandler;
private readonly Func<CultNetShardCatalogRequestMessage, CultNetServerPeer, Task> _shardCatalogHandler;
private readonly Func<CultNetShardLogRequestMessage, CultNetServerPeer, Task> _shardLogHandler;
private readonly Func<CultNetDatabaseSubscribeMessage, CultNetServerPeer, Task> _subscribeHandler;
private readonly Func<CultNetDatabaseUnsubscribeMessage, CultNetServerPeer, Task> _unsubscribeHandler;
private readonly ConcurrentDictionary<string, IDisposable> _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();
}

8
src/GameCult.Networking/CultNetSimulationObservationServer.cs
View file

@ -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<CultNetSimulationObservationMessage, NetPeer, Task> _observationHandler;
private readonly Func<CultNetSimulationObservationMessage, CultNetServerPeer, Task> _observationHandler;
private bool _disposed;
/// <summary>
@ -59,11 +58,6 @@ namespace GameCult.Networking
_server.RemoveCultNetMessageListener<CultNetSimulationObservationMessage>(_observationHandler);
}
private Task HandleObservationAsync(CultNetSimulationObservationMessage message, NetPeer peer)
{
return HandleObservationAsync(message, _server.GetPeerContext(peer));
}
private Task HandleObservationAsync(CultNetSimulationObservationMessage message, CultNetServerPeer peer)
{
try

83
src/GameCult.Networking/Server.cs
View file

@ -30,6 +30,7 @@ namespace GameCult.Networking
private readonly ConcurrentDictionary<Type, Delegate> _messageDelegates = new();
private readonly ConcurrentDictionary<Type, Delegate> _cultNetMessageDelegates = new();
private readonly ConcurrentDictionary<Type, Delegate> _cultNetServerPeerMessageDelegates = new();
private readonly ConcurrentDictionary<long, User> _users = new();
private readonly ConcurrentDictionary<string, Queue<DateTimeOffset>> _connectionAttempts = new();
private readonly ConcurrentDictionary<string, object> _connectionAttemptLocks = new();
@ -106,6 +107,7 @@ namespace GameCult.Networking
{
_messageDelegates.Clear();
_cultNetMessageDelegates.Clear();
_cultNetServerPeerMessageDelegates.Clear();
}
/// <summary>
@ -160,6 +162,35 @@ namespace GameCult.Networking
});
}
/// <summary>
/// Adds a transport-aware listener for a modern CultNet schema-v0 message type.
/// </summary>
public void AddCultNetMessageListener<T>(Func<T, CultNetServerPeer, Task> callback)
where T : ICultNetSchemaMessage
{
var type = typeof(T);
_cultNetServerPeerMessageDelegates.AddOrUpdate(type,
_ => callback,
(t, current) =>
{
var combined = Delegate.Combine(current, callback) as Func<T, CultNetServerPeer, Task>;
return combined ?? throw new InvalidOperationException($"Failed to combine delegates for {t.Name}");
});
}
/// <summary>
/// Adds a transport-aware listener for a modern CultNet schema-v0 message type.
/// </summary>
public void AddCultNetMessageListener<T>(Action<T, CultNetServerPeer> callback)
where T : ICultNetSchemaMessage
{
AddCultNetMessageListener<T>((message, peer) =>
{
callback(message, peer);
return Task.CompletedTask;
});
}
/// <summary>
/// Adds a listener for a modern CultNet schema-v0 message type.
/// </summary>
@ -176,6 +207,24 @@ namespace GameCult.Networking
AddCultNetMessageListener(callback);
}
/// <summary>
/// Adds a transport-aware listener for a modern CultNet schema-v0 message type.
/// </summary>
public void OnCultNet<T>(Func<T, CultNetServerPeer, Task> callback)
where T : ICultNetSchemaMessage
{
AddCultNetMessageListener(callback);
}
/// <summary>
/// Adds a transport-aware listener for a modern CultNet schema-v0 message type.
/// </summary>
public void OnCultNet<T>(Action<T, CultNetServerPeer> callback)
where T : ICultNetSchemaMessage
{
AddCultNetMessageListener(callback);
}
/// <summary>
/// Sends a legacy GameCult.Networking union message through this server's per-peer LiteNetLib adapter.
/// </summary>
@ -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;
}
}
}
/// <summary>
@ -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);
}