diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs
index 12ccd470..027cc68e 100644
--- a/src/GameCult.Mesh/CultMeshClient.cs
+++ b/src/GameCult.Mesh/CultMeshClient.cs
@@ -42,6 +42,40 @@ namespace GameCult.Mesh
public TimeSpan SubscriptionResponseTimeout { get; set; } = TimeSpan.FromSeconds(2);
}
+ /// Owns one shared live document subscription until disposed.
+ public sealed class CultMeshDocumentLease : IDisposable where TDocument : class
+ {
+ private Action? _release;
+
+ internal CultMeshDocumentLease(CultMeshDocumentHandle handle, Action release)
+ {
+ Handle = handle ?? throw new ArgumentNullException(nameof(handle));
+ _release = release ?? throw new ArgumentNullException(nameof(release));
+ }
+
+ /// Gets the live typed document handle owned by this lease.
+ public CultMeshDocumentHandle Handle { get; }
+
+ public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke();
+ }
+
+ /// Owns one shared live collection subscription until disposed.
+ public sealed class CultMeshCollectionLease : IDisposable where TDocument : class
+ {
+ private Action? _release;
+
+ internal CultMeshCollectionLease(CultMeshCollectionHandle handle, Action release)
+ {
+ Handle = handle ?? throw new ArgumentNullException(nameof(handle));
+ _release = release ?? throw new ArgumentNullException(nameof(release));
+ }
+
+ /// Gets the live typed collection handle owned by this lease.
+ public CultMeshCollectionHandle Handle { get; }
+
+ public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke();
+ }
+
///
/// Owns discovery and reusable sessions for one application lifetime.
/// Applications address stable identities; CultMesh owns physical routes and reconnection.
@@ -51,9 +85,8 @@ namespace GameCult.Mesh
private readonly CultMeshDiscoveryService _discovery;
private readonly CultMeshSessionManager _sessions;
private readonly TimeSpan _subscriptionResponseTimeout;
- private readonly ConcurrentDictionary>> _documents = new(StringComparer.Ordinal);
- private readonly ConcurrentDictionary>> _collections = new(StringComparer.Ordinal);
- private readonly ConcurrentBag _documentResources = new();
+ private readonly ConcurrentDictionary _documents = new(StringComparer.Ordinal);
+ private readonly ConcurrentDictionary _collections = new(StringComparer.Ordinal);
private bool _disposed;
public CultMeshClient(CultMeshClientOptions options)
@@ -147,8 +180,8 @@ namespace GameCult.Mesh
options);
}
- /// Opens one live typed document by stable provider identity and record key.
- public async Task> DocumentAsync(
+ /// Leases one shared live typed document by stable provider identity and record key.
+ public async Task> LeaseDocumentAsync(
string endpointId,
string recordKey,
CancellationToken cancellationToken = default)
@@ -158,31 +191,40 @@ namespace GameCult.Mesh
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Record key is required.", nameof(recordKey));
var key = endpointId.Trim() + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + recordKey.Trim();
- var lazy = _documents.GetOrAdd(key, _ => new Lazy>(
- async () => await OpenDocumentAsync(endpointId.Trim(), recordKey.Trim()).ConfigureAwait(false),
- LazyThreadSafetyMode.ExecutionAndPublication));
- try
+ while (true)
{
- var binding = (RemoteDocumentBinding)await AwaitForCallerAsync(lazy.Value, cancellationToken)
- .ConfigureAwait(false);
- return binding.Handle;
- }
- catch
- {
- _documents.TryRemove(key, out _);
- throw;
+ var resource = _documents.GetOrAdd(key, _ => new SharedResource(
+ async owner => await OpenDocumentAsync(endpointId.Trim(), recordKey.Trim(), owner).ConfigureAwait(false)));
+ if (!resource.TryAcquire())
+ {
+ RemoveExact(_documents, key, resource);
+ continue;
+ }
+ try
+ {
+ var binding = (RemoteDocumentBinding)await AwaitForCallerAsync(resource.Value, cancellationToken)
+ .ConfigureAwait(false);
+ return new CultMeshDocumentLease(
+ binding.Handle,
+ () => Release(_documents, key, resource));
+ }
+ catch
+ {
+ Release(_documents, key, resource);
+ throw;
+ }
}
}
- /// Opens a live collection of one typed schema by stable provider identity.
- public Task> CollectionAsync(
+ /// Leases a shared live collection of one typed schema by stable provider identity.
+ public Task> LeaseCollectionAsync(
string endpointId,
CancellationToken cancellationToken)
where TDocument : class =>
- CollectionAsync(endpointId, includeInitialSnapshot: true, cancellationToken);
+ LeaseCollectionAsync(endpointId, includeInitialSnapshot: true, cancellationToken);
- /// Opens a live collection and optionally skips replaying its initial snapshot.
- public async Task> CollectionAsync(
+ /// Leases a live collection and optionally skips replaying its initial snapshot.
+ public async Task> LeaseCollectionAsync(
string endpointId,
bool includeInitialSnapshot = true,
CancellationToken cancellationToken = default)
@@ -192,22 +234,74 @@ namespace GameCult.Mesh
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
var identity = endpointId.Trim();
var key = identity + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + includeInitialSnapshot;
- var lazy = _collections.GetOrAdd(key, _ => new Lazy>(
- async () => await OpenCollectionAsync(identity, includeInitialSnapshot).ConfigureAwait(false),
- LazyThreadSafetyMode.ExecutionAndPublication));
- try
+ while (true)
{
- var binding = (RemoteCollectionBinding)await AwaitForCallerAsync(lazy.Value, cancellationToken)
- .ConfigureAwait(false);
- return binding.Handle;
- }
- catch
- {
- _collections.TryRemove(key, out _);
- throw;
+ var resource = _collections.GetOrAdd(key, _ => new SharedResource(
+ async owner => await OpenCollectionAsync(identity, includeInitialSnapshot, owner).ConfigureAwait(false)));
+ if (!resource.TryAcquire())
+ {
+ RemoveExact(_collections, key, resource);
+ continue;
+ }
+ try
+ {
+ var binding = (RemoteCollectionBinding)await AwaitForCallerAsync(resource.Value, cancellationToken)
+ .ConfigureAwait(false);
+ return new CultMeshCollectionLease(
+ binding.Handle,
+ () => Release(_collections, key, resource));
+ }
+ catch
+ {
+ Release(_collections, key, resource);
+ throw;
+ }
}
}
+ ///
+ /// Submits one typed document to a provider by stable identity. Submission is uncommitted input;
+ /// only a provider-authored receipt or resulting state can establish acceptance.
+ ///
+ public async Task SubmitDocumentAsync(
+ string endpointId,
+ string recordKey,
+ TDocument document,
+ string sourceRuntimeId,
+ string sourceRole,
+ CancellationToken cancellationToken = default)
+ where TDocument : class
+ {
+ ThrowIfDisposed();
+ if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
+ if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Record key is required.", nameof(recordKey));
+ if (document == null) throw new ArgumentNullException(nameof(document));
+ if (string.IsNullOrWhiteSpace(sourceRuntimeId)) throw new ArgumentException("Source runtime identity is required.", nameof(sourceRuntimeId));
+ if (string.IsNullOrWhiteSpace(sourceRole)) throw new ArgumentException("Source role is required.", nameof(sourceRole));
+
+ var cacheRegistry = CultMesh.CreateCultCacheDocumentRegistry(typeof(TDocument));
+ var networkRegistry = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(TDocument) }, cacheRegistry);
+ var message = networkRegistry.CreateRawDocumentPutMessage(
+ "cultmesh-client-submit-" + Guid.NewGuid().ToString("N"),
+ new CultRecordHandle(new CultRecordKey(recordKey.Trim())),
+ document,
+ new CultNetDocumentMessageOptions
+ {
+ SourceRuntimeId = sourceRuntimeId.Trim(),
+ SourceRole = sourceRole.Trim()
+ });
+ var session = await ConnectAsync(endpointId.Trim(), CultMeshProtocols.Documents, cancellationToken)
+ .ConfigureAwait(false);
+ cancellationToken.ThrowIfCancellationRequested();
+ session.OpenSchemaClient().SendCultNet(message);
+ }
+
+ /// Gets the number of currently leased shared document subscriptions.
+ public int ActiveDocumentResourceCount => _documents.Count;
+
+ /// Gets the number of currently leased shared collection subscriptions.
+ public int ActiveCollectionResourceCount => _collections.Count;
+
/// Reads one typed record once through the reusable document session.
public async Task ReadAsync(
string endpointId,
@@ -293,7 +387,7 @@ namespace GameCult.Mesh
{
if (_disposed) return;
_disposed = true;
- foreach (var resource in _documentResources) resource.Dispose();
+ foreach (var resource in _documents.Values.Concat(_collections.Values).Distinct()) resource.Dispose();
_documents.Clear();
_collections.Clear();
_sessions.Dispose();
@@ -305,12 +399,15 @@ namespace GameCult.Mesh
if (_disposed) throw new ObjectDisposedException(nameof(CultMeshClient));
}
- private async Task