diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs
index 08260e8a..ddc55a38 100644
--- a/src/GameCult.Mesh/CultMeshClient.cs
+++ b/src/GameCult.Mesh/CultMeshClient.cs
@@ -115,17 +115,25 @@ namespace GameCult.Mesh
}
/// Opens a live collection of one typed schema by stable provider identity.
+ public Task> CollectionAsync(
+ string endpointId,
+ CancellationToken cancellationToken)
+ where TDocument : class =>
+ CollectionAsync(endpointId, includeInitialSnapshot: true, cancellationToken);
+
+ /// Opens a live collection and optionally skips replaying its initial snapshot.
public async Task> CollectionAsync(
string endpointId,
+ bool includeInitialSnapshot = true,
CancellationToken cancellationToken = default)
where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
var identity = endpointId.Trim();
- var key = identity + "\u001f" + typeof(TDocument).AssemblyQualifiedName;
+ var key = identity + "\u001f" + typeof(TDocument).AssemblyQualifiedName + "\u001f" + includeInitialSnapshot;
var lazy = _collections.GetOrAdd(key, _ => new Lazy>(
- async () => await OpenCollectionAsync(identity).ConfigureAwait(false),
+ async () => await OpenCollectionAsync(identity, includeInitialSnapshot).ConfigureAwait(false),
LazyThreadSafetyMode.ExecutionAndPublication));
try
{
@@ -157,12 +165,44 @@ namespace GameCult.Mesh
$"CultMesh endpoint '{endpointId}' did not publish {typeof(TDocument).FullName} record '{recordKey}'.");
}
+ /// Reads one typed record once with an explicit response deadline.
+ public async Task ReadAsync(
+ string endpointId,
+ string recordKey,
+ TimeSpan responseTimeout,
+ CancellationToken cancellationToken = default)
+ where TDocument : class
+ {
+ var documents = await ReadManyCoreAsync(
+ endpointId,
+ new[] { recordKey },
+ responseTimeout,
+ cancellationToken)
+ .ConfigureAwait(false);
+ return documents.FirstOrDefault()
+ ?? throw new InvalidOperationException(
+ $"CultMesh endpoint '{endpointId}' did not publish {typeof(TDocument).FullName} record '{recordKey}'.");
+ }
+
/// Reads typed records once in one request through the reusable document session.
public async Task> ReadManyAsync(
string endpointId,
IReadOnlyList recordKeys,
CancellationToken cancellationToken = default)
where TDocument : class
+ => await ReadManyCoreAsync(
+ endpointId,
+ recordKeys,
+ TimeSpan.FromSeconds(10),
+ cancellationToken)
+ .ConfigureAwait(false);
+
+ private async Task> ReadManyCoreAsync(
+ string endpointId,
+ IReadOnlyList recordKeys,
+ TimeSpan responseTimeout,
+ CancellationToken cancellationToken)
+ where TDocument : class
{
ThrowIfDisposed();
if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId));
@@ -177,7 +217,7 @@ namespace GameCult.Mesh
new CultMeshSnapshotRequestOptions
{
ConnectTimeout = TimeSpan.FromSeconds(5),
- ResponseTimeout = TimeSpan.FromSeconds(10),
+ ResponseTimeout = responseTimeout,
MessageIdPrefix = "cultmesh-client-read"
},
networkRegistry,
@@ -223,11 +263,11 @@ namespace GameCult.Mesh
}
}
- private async Task