From b5529435c3bb9de0a37e5512ec6eb94794fe73e2 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 12 Jul 2026 12:19:49 +0100 Subject: [PATCH] Replicate live typed documents over retained sessions --- .../CultNetDatabaseSubscriptionClient.cs | 190 ++++++++++++++++++ .../NetworkingTests.cs | 57 ++++++ 2 files changed, 247 insertions(+) create mode 100644 src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs new file mode 100644 index 00000000..151722d6 --- /dev/null +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -0,0 +1,190 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; +using System.Threading.Tasks; +using GameCult.Caching; + +namespace GameCult.Networking +{ + /// + /// Replicates selected remote database records into a local typed CultCache over one retained session. + /// + public sealed class CultNetDatabaseSubscriptionClient : IDisposable + { + private readonly ICultNetSchemaClient _client; + private readonly CultCache _cache; + private readonly CultNetDocumentRegistry _documents; + private readonly ConcurrentDictionary>> _initialSnapshots = new(); + private readonly object _queueLock = new(); + private Task _queue = Task.CompletedTask; + private bool _disposed; + + public CultNetDatabaseSubscriptionClient( + ICultNetSchemaClient client, + CultCache cache, + CultNetDocumentRegistry documents) + { + _client = client ?? throw new ArgumentNullException(nameof(client)); + _cache = cache ?? throw new ArgumentNullException(nameof(cache)); + _documents = documents ?? throw new ArgumentNullException(nameof(documents)); + _client.OnCultNet(HandleSnapshot); + _client.OnCultNet(HandleChange); + } + + /// Raised after a live change has been applied to the local typed cache. + public event Action? Changed; + + /// Gets the local reactive cache receiving replicated documents. + public CultCache Cache => _cache; + + /// + /// Starts a remote subscription and waits until its initial matching snapshot is in the local cache. + /// + public Task> SubscribeAsync( + string subscriptionId, + IEnumerable? recordKeys = null, + IEnumerable? schemaIds = null) + { + if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient)); + if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing."); + if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); + + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); + if (!_initialSnapshots.TryAdd(messageId, completion)) + throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'."); + + _client.SendCultNet(new CultNetDatabaseSubscribeMessage + { + MessageId = messageId, + SubscriptionId = subscriptionId, + RecordKeys = recordKeys?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), + SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), + IncludeSnapshot = true + }); + return completion.Task; + } + + /// Stops one live remote subscription. + public void Unsubscribe(string subscriptionId) + { + if (_disposed) return; + if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); + _client.SendCultNet(new CultNetDatabaseUnsubscribeMessage + { + MessageId = Guid.NewGuid().ToString("N"), + SubscriptionId = subscriptionId + }); + } + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + foreach (var completion in _initialSnapshots.Values) + completion.TrySetException(new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient))); + _initialSnapshots.Clear(); + _client.Dispose(); + } + + private void HandleSnapshot(CultNetSnapshotResponseRawMessage message) + { + Enqueue(async () => + { + if (!_initialSnapshots.TryRemove(message.MessageId, out var completion)) return; + try + { + var applied = await _documents.ApplyRawSnapshotResponseAsync(_cache, message).ConfigureAwait(false); + completion.TrySetResult(applied); + } + catch (Exception error) + { + completion.TrySetException(error); + } + }); + } + + private void HandleChange(CultNetDatabaseChangeRawMessage message) + { + Enqueue(async () => + { + if (string.Equals(message.ChangeKind, "removed", StringComparison.Ordinal)) + { + await DeleteAsync(message).ConfigureAwait(false); + Changed?.Invoke(new CultNetReplicatedDocumentChange( + message.SubscriptionId, + "removed", + message.RecordKey ?? string.Empty, + message.SchemaId ?? string.Empty, + null)); + return; + } + + if (message.Document == null) return; + var document = await _documents.ApplyRawDocumentPutMessageAsync( + _cache, + new CultNetDocumentPutRawMessage + { + MessageId = message.MessageId, + Document = message.Document + }).ConfigureAwait(false); + Changed?.Invoke(new CultNetReplicatedDocumentChange( + message.SubscriptionId, + message.ChangeKind, + message.Document.RecordKey, + message.Document.SchemaId, + document)); + }); + } + + private async Task DeleteAsync(CultNetDatabaseChangeRawMessage message) + { + if (string.IsNullOrWhiteSpace(message.RecordKey) || string.IsNullOrWhiteSpace(message.SchemaId)) return; + var descriptor = _documents.ResolveDescriptorForSchemaId(message.SchemaId!); + var handle = Activator.CreateInstance( + typeof(CultRecordHandle<>).MakeGenericType(descriptor.DocumentType), + new object[] { new CultRecordKey(message.RecordKey!) }); + var task = (Task)typeof(CultCache) + .GetMethod(nameof(CultCache.DeleteAsync), BindingFlags.Public | BindingFlags.Instance)! + .MakeGenericMethod(descriptor.DocumentType) + .Invoke(_cache, new[] { handle })!; + await task.ConfigureAwait(false); + } + + private void Enqueue(Func operation) + { + lock (_queueLock) + { + _queue = _queue.ContinueWith( + _ => operation(), + TaskScheduler.Default).Unwrap(); + } + } + } + + /// Describes one typed document change after local replication. + public sealed class CultNetReplicatedDocumentChange + { + public CultNetReplicatedDocumentChange( + string subscriptionId, + string changeKind, + string recordKey, + string schemaId, + object? document) + { + SubscriptionId = subscriptionId; + ChangeKind = changeKind; + RecordKey = recordKey; + SchemaId = schemaId; + Document = document; + } + + public string SubscriptionId { get; } + public string ChangeKind { get; } + public string RecordKey { get; } + public string SchemaId { get; } + public object? Document { get; } + } +} diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 82f777f5..42707754 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1362,6 +1362,63 @@ namespace GameCult.Networking.Tests Assert.That(update.Document!.RecordKey, Is.EqualTo(recordKey)); } + [Test] + public async Task DatabaseSubscriptionClient_ReplicatesInitialAndLiveTypedDocumentsOverRudp() + { + var sourceCache = new CultCache(); + var sourceDatabase = new CultNetDatabase(sourceCache); + const string recordKey = "tests:subscription-client:note"; + await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "initial" + }); + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "database-subscription-client-server", + Socket = BindUdpSocket() + }); + using var subscriptions = new CultNetDatabaseSubscriptionServer(server, sourceDatabase); + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + }) { IsBackground = true }; + serverThread.Start(); + + var targetCache = new CultCache(); + var targetDocuments = new CultNetDocumentRegistry(targetCache.Registry) + .Register(CultNetDocumentBinding.ForDocument(targetCache.Registry)); + var transport = CultNetSchemaClients.CreateRudp("database-subscription-client"); + using var client = new CultNetDatabaseSubscriptionClient(transport, targetCache, targetDocuments); + transport.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => transport.Connected, TimeSpan.FromSeconds(2)); + + var initial = await AwaitWithTimeout( + client.SubscribeAsync("notes", recordKeys: new[] { recordKey }), + TimeSpan.FromSeconds(2)); + Assert.That(initial.OfType().Single().Text, Is.EqualTo("initial")); + Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.TypeOf()); + + var changed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + client.Changed += change => changed.TrySetResult(change); + await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "live" + }); + var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2)); + + cancellation.Cancel(); + Assert.That(update.SubscriptionId, Is.EqualTo("notes")); + Assert.That(update.Document, Is.TypeOf()); + Assert.That(((NetworkSchemaNote)targetCache.Get(new CultRecordKey(recordKey))!).Text, Is.EqualTo("live")); + } + [Test] public async Task RudpCultNetSchemaServer_DeliversLargeSnapshotResponse() {