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

Project subscription records per peer

Browse source
This commit is contained in:
Meta Vens 2026-08-19 13:22:36 +02:00
commit eaf1189cc1
2 changed files with 27 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

20
src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
View file

@ -13,6 +13,8 @@ namespace GameCult.Networking
{
/// <summary>
/// Publishes live database changes through any schema-v0 server transport.
/// Optional record projection runs after authorization and owns both initial
/// snapshot and live-update delivery for a peer.
/// </summary>
public sealed class CultNetDatabaseSubscriptionServer : IDisposable
{
@ -22,6 +24,7 @@ namespace GameCult.Networking
private readonly Func<CultNetDatabaseUnsubscribeMessage, ICultNetSchemaServerPeer, Task> _unsubscribe;
private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, bool>? _authorizeRequest;
private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? _authorizeRecord;
private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, CultNetRawDocumentRecord, CultNetRawDocumentRecord?>? _projectRecord;
private readonly ICultNetSchemaServerPeerLifecycle? _peerLifecycle;
private readonly ConcurrentDictionary<SubscriptionKey, IDisposable> _subscriptions =
new ConcurrentDictionary<SubscriptionKey, IDisposable>(SubscriptionKeyComparer.Instance);
@ -34,12 +37,14 @@ namespace GameCult.Networking
ICultNetSchemaServer server,
CultNetDatabase database,
Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, bool>? authorizeRequest = null,
Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? authorizeRecord = null)
Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? authorizeRecord = null,
Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, CultNetRawDocumentRecord, CultNetRawDocumentRecord?>? projectRecord = null)
{
_server = server ?? throw new ArgumentNullException(nameof(server));
_database = database ?? throw new ArgumentNullException(nameof(database));
_authorizeRequest = authorizeRequest;
_authorizeRecord = authorizeRecord;
_projectRecord = projectRecord;
_subscribe = HandleSubscribeAsync;
_unsubscribe = HandleUnsubscribeAsync;
_server.OnCultNet(_subscribe);
@ -134,6 +139,14 @@ namespace GameCult.Networking
request, peer, document.RecordKey, document.SchemaId))
.ToArray();
}
if (_projectRecord != null)
{
snapshot.Documents = snapshot.Documents
.Select(document => _projectRecord(request, peer, document))
.Where(document => document != null)
.Cast<CultNetRawDocumentRecord>()
.ToArray();
}
peer.SendCultNet(snapshot);
}
else
@ -205,6 +218,11 @@ namespace GameCult.Networking
return _database.WatchAllChanges().Subscribe(change =>
{
var message = CreateAuthorizedChange(change, request, subscriptionId, peer);
if (message?.Document != null && _projectRecord != null)
{
message.Document = _projectRecord(request, peer, message.Document);
if (message.Document == null) return;
}
if (message != null) peer.SendCultNet(message);
});
}

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

@ -1476,7 +1476,12 @@ namespace GameCult.Networking.Tests
server,
database,
authorizeRequest: (request, _) => request.ConsumerRuntimeId == "allowed-runtime",
authorizeRecord: (_, _, recordKey, _) => recordKey.StartsWith("tests:public:", StringComparison.Ordinal));
authorizeRecord: (_, _, recordKey, _) => recordKey.StartsWith("tests:public:", StringComparison.Ordinal),
projectRecord: (request, _, record) =>
{
record.SourceRuntimeId = "projection:" + request.ConsumerRuntimeId;
return record;
});
using var cancellation = new CancellationTokenSource();
var serverThread = new Thread(() =>
{
@ -1519,6 +1524,7 @@ namespace GameCult.Networking.Tests
var snapshot = await AwaitWithTimeout(subscribed.Task, TimeSpan.FromSeconds(2));
Assert.That(snapshot.Documents.Select(document => document.RecordKey),
Is.EqualTo(new[] { "tests:public:initial" }));
Assert.That(snapshot.Documents.Single().SourceRuntimeId, Is.EqualTo("projection:allowed-runtime"));
await database.PutAsync(new CultRecordKey("tests:private:live"), new NetworkSchemaNote
{
@ -1535,6 +1541,7 @@ namespace GameCult.Networking.Tests
cancellation.Cancel();
Assert.That(update.Document, Is.Not.Null);
Assert.That(update.Document!.RecordKey, Is.EqualTo("tests:public:live"));
Assert.That(update.Document.SourceRuntimeId, Is.EqualTo("projection:allowed-runtime"));
}
[Test]