diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 3d31483b..515701fb 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -947,6 +947,13 @@ ownership contract. **Outputs:** both modes produce the same typed initial values and change event. Only durable replication mutates the consumer's CultCache. +`CultNetDatabaseSubscriptionClient.SubscribeLiveValueAsync(...)` is the +single-record ergonomic path. Its `CultNetLiveValue` owns the exact +record/schema filter, current value, typed change/removal notifications, and +unsubscribe lifetime. Consumers no longer need a process-wide change handler, +an object type switch, or a renderer-local replica cache to model one reactive +variable. + **Derived state:** renderer-local presentation and reactive variables are derived from live notifications. They are not database records merely because CultNet carried them. diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 4c7770ef..1491e45c 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -55,6 +55,44 @@ namespace GameCult.Networking public Task? BackgroundFailure => (_client as RudpCultNetSchemaClient)?.BackgroundFailure; + /// + /// Subscribes to one exact typed record as an ephemeral reactive value. The returned value + /// owns filtering and unsubscribe; no received payload is written into the local cache. + /// + public async Task> SubscribeLiveValueAsync( + string subscriptionId, + string recordKey, + CancellationToken cancellationToken = default) + where TDocument : class + { + if (string.IsNullOrWhiteSpace(recordKey)) + throw new ArgumentException("Record key is required.", nameof(recordKey)); + var schemaId = _documents.GetByDocumentType(typeof(TDocument))?.SchemaId ?? + _cache.Registry.GetRequired().SchemaId; + var value = new CultNetLiveValue(this, subscriptionId, recordKey, schemaId); + try + { + var initial = await SubscribeAsync( + subscriptionId, + recordKeys: new[] { recordKey }, + schemaIds: new[] { schemaId }, + deliveryMode: CultNetDatabaseSubscriptionDeliveryMode.Live, + cancellationToken: cancellationToken) + .ConfigureAwait(false); + var documents = initial.OfType().ToArray(); + if (documents.Length != 1) + throw new InvalidOperationException( + $"Live value subscription '{subscriptionId}' expected one '{schemaId}' record '{recordKey}' but received {documents.Length}."); + value.Initialize(documents[0]); + return value; + } + catch + { + value.Dispose(); + throw; + } + } + /// /// Starts a remote subscription and waits until its initial matching snapshot is in the local cache. /// @@ -250,6 +288,110 @@ namespace GameCult.Networking } } + /// One exact typed remote record kept as an ephemeral current value. + public sealed class CultNetLiveValue : IDisposable where TDocument : class + { + private readonly CultNetDatabaseSubscriptionClient _owner; + private readonly string _subscriptionId; + private readonly string _recordKey; + private readonly string _schemaId; + private readonly object _gate = new(); + private TDocument? _current; + private bool _hasValue; + private bool _disposed; + + internal CultNetLiveValue( + CultNetDatabaseSubscriptionClient owner, + string subscriptionId, + string recordKey, + string schemaId) + { + _owner = owner ?? throw new ArgumentNullException(nameof(owner)); + _subscriptionId = string.IsNullOrWhiteSpace(subscriptionId) + ? throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)) + : subscriptionId; + _recordKey = recordKey; + _schemaId = schemaId; + _owner.Changed += OnChanged; + } + + /// Raised when the provider publishes a new typed value. + public event Action? Changed; + + /// Raised when the provider removes the selected record. + public event Action? Removed; + + /// Gets whether the selected record currently exists. + public bool HasValue + { + get { lock (_gate) return _hasValue; } + } + + /// Gets the current value, or fails when the record has not been published. + public TDocument Current + { + get + { + lock (_gate) + return _hasValue + ? _current! + : throw new InvalidOperationException( + $"Live value '{_recordKey}' is not currently published."); + } + } + + internal void Initialize(TDocument document) + { + lock (_gate) + { + if (_disposed || _hasValue) return; + _current = document; + _hasValue = true; + } + } + + public void Dispose() + { + lock (_gate) + { + if (_disposed) return; + _disposed = true; + } + _owner.Changed -= OnChanged; + _owner.Unsubscribe(_subscriptionId); + } + + private void OnChanged(CultNetReplicatedDocumentChange change) + { + if (!string.Equals(change.SubscriptionId, _subscriptionId, StringComparison.Ordinal) || + !string.Equals(change.RecordKey, _recordKey, StringComparison.Ordinal) || + !string.Equals(change.SchemaId, _schemaId, StringComparison.Ordinal)) + return; + + if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal)) + { + lock (_gate) + { + if (_disposed) return; + _current = null; + _hasValue = false; + } + Removed?.Invoke(); + return; + } + + if (change.Document is not TDocument document) + return; + lock (_gate) + { + if (_disposed) return; + _current = document; + _hasValue = true; + } + Changed?.Invoke(document); + } + } + /// Describes one typed document change after local replication. public sealed class CultNetReplicatedDocumentChange { diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index ed03fc57..e44f0018 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1693,6 +1693,64 @@ namespace GameCult.Networking.Tests Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null); } + [Test] + public async Task DatabaseSubscriptionClient_LiveValueOwnsExactFilteringCurrentValueAndUnsubscribe() + { + var sourceCache = new CultCache(); + var sourceDatabase = new CultNetDatabase(sourceCache); + const string recordKey = "tests:subscription-client:reactive-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-live-value-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-live-value-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)); + + using var value = await AwaitWithTimeout( + client.SubscribeLiveValueAsync("reactive-note", recordKey), + TimeSpan.FromSeconds(2)); + Assert.That(value.HasValue, Is.True); + Assert.That(value.Current.Text, Is.EqualTo("initial")); + Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null); + + var changed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + value.Changed += document => changed.TrySetResult(document); + await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote + { + Schema = "tests.networking_note.v1", + Text = "updated" + }); + var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2)); + + cancellation.Cancel(); + Assert.That(update.Text, Is.EqualTo("updated")); + Assert.That(value.Current.Text, Is.EqualTo("updated")); + Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null); + } + [Test] public async Task DatabaseSubscription_FiltersLiveChangesByWireSchemaBinding() {