Replicate live typed documents over retained sessions
This commit is contained in:
Meta Vens
parent
4d8d43621d
commit
b5529435c3
2 changed files with 247 additions and 0 deletions
190
src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs
Normal file
190
src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs
Normal file
|
|
@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// Replicates selected remote database records into a local typed CultCache over one retained session.
|
||||
/// </summary>
|
||||
public sealed class CultNetDatabaseSubscriptionClient : IDisposable
|
||||
{
|
||||
private readonly ICultNetSchemaClient _client;
|
||||
private readonly CultCache _cache;
|
||||
private readonly CultNetDocumentRegistry _documents;
|
||||
private readonly ConcurrentDictionary<string, TaskCompletionSource<IReadOnlyList<object>>> _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<CultNetSnapshotResponseRawMessage>(HandleSnapshot);
|
||||
_client.OnCultNet<CultNetDatabaseChangeRawMessage>(HandleChange);
|
||||
}
|
||||
|
||||
/// <summary>Raised after a live change has been applied to the local typed cache.</summary>
|
||||
public event Action<CultNetReplicatedDocumentChange>? Changed;
|
||||
|
||||
/// <summary>Gets the local reactive cache receiving replicated documents.</summary>
|
||||
public CultCache Cache => _cache;
|
||||
|
||||
/// <summary>
|
||||
/// Starts a remote subscription and waits until its initial matching snapshot is in the local cache.
|
||||
/// </summary>
|
||||
public Task<IReadOnlyList<object>> SubscribeAsync(
|
||||
string subscriptionId,
|
||||
IEnumerable<string>? recordKeys = null,
|
||||
IEnumerable<string>? 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<IReadOnlyList<object>>(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;
|
||||
}
|
||||
|
||||
/// <summary>Stops one live remote subscription.</summary>
|
||||
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<Task> operation)
|
||||
{
|
||||
lock (_queueLock)
|
||||
{
|
||||
_queue = _queue.ContinueWith(
|
||||
_ => operation(),
|
||||
TaskScheduler.Default).Unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Describes one typed document change after local replication.</summary>
|
||||
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; }
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue