diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
index cd224816..c1ca3c9d 100644
--- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
+++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
@@ -13,6 +13,8 @@ namespace GameCult.Networking
{
///
/// 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.
///
public sealed class CultNetDatabaseSubscriptionServer : IDisposable
{
@@ -22,6 +24,7 @@ namespace GameCult.Networking
private readonly Func _unsubscribe;
private readonly Func? _authorizeRequest;
private readonly Func? _authorizeRecord;
+ private readonly Func? _projectRecord;
private readonly ICultNetSchemaServerPeerLifecycle? _peerLifecycle;
private readonly ConcurrentDictionary _subscriptions =
new ConcurrentDictionary(SubscriptionKeyComparer.Instance);
@@ -34,12 +37,14 @@ namespace GameCult.Networking
ICultNetSchemaServer server,
CultNetDatabase database,
Func? authorizeRequest = null,
- Func? authorizeRecord = null)
+ Func? authorizeRecord = null,
+ Func? 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()
+ .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);
});
}
diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs
index ae3265ca..deac0d02 100644
--- a/tests/GameCult.Networking.Tests/NetworkingTests.cs
+++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs
@@ -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]