Add typed snapshot endpoint surface
This commit is contained in:
Meta Vens
parent
a6172aa586
commit
8b48aa512f
3 changed files with 300 additions and 64 deletions
|
|
@ -133,11 +133,35 @@ namespace GameCult.Mesh
|
|||
/// <summary>Gets or sets the message id prefix used for snapshot requests.</summary>
|
||||
public string? MessageIdPrefix { get; set; }
|
||||
|
||||
/// <summary>Gets or sets schema ids to request. Empty means the document type schema is requested.</summary>
|
||||
public IReadOnlyList<string>? SchemaIds { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the target shard id, when the endpoint is shard-aware.</summary>
|
||||
public string? ShardId { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the target shard epoch, when the endpoint is shard-aware.</summary>
|
||||
public long? ShardEpoch { get; set; }
|
||||
|
||||
/// <summary>Gets or sets client security options for endpoint-created clients.</summary>
|
||||
public ClientSecurityOptions? Security { get; set; }
|
||||
|
||||
/// <summary>Gets or sets a callback used to configure endpoint-created LiteNetLib clients.</summary>
|
||||
public Action<Client>? ConfigureClient { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the runtime id used by endpoint-created RUDP clients.</summary>
|
||||
public string? RudpRuntimeId { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the connection id used by endpoint-created RUDP clients.</summary>
|
||||
public uint RudpConnectionId { get; set; } = 0x43554c54;
|
||||
|
||||
/// <summary>Gets or sets the connect payload used by endpoint-created RUDP clients.</summary>
|
||||
public string RudpConnectPayload { get; set; } = "cultnet-schema-rudp";
|
||||
|
||||
/// <summary>Gets or sets the maximum fragment size used by endpoint-created RUDP clients.</summary>
|
||||
public int RudpMaxFragmentBytes { get; set; } = 1024;
|
||||
|
||||
/// <summary>Gets or sets the resend delay used by endpoint-created RUDP clients.</summary>
|
||||
public long RudpResendDelayMs { get; set; } = 25;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -2086,73 +2110,44 @@ namespace GameCult.Mesh
|
|||
where TDocument : class
|
||||
{
|
||||
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
|
||||
var messageId = CreateSnapshotMessageId(options, context);
|
||||
var completion = new TaskCompletionSource<CultNetSnapshotResponseRawMessage>(
|
||||
TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using var client = createClient();
|
||||
client.OnCultNet<CultNetSnapshotResponseRawMessage>(response =>
|
||||
{
|
||||
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
|
||||
{
|
||||
completion.TrySetResult(response);
|
||||
}
|
||||
});
|
||||
client.OnCultNet<CultNetErrorMessage>(error =>
|
||||
completion.TrySetException(new InvalidOperationException(error.Error)));
|
||||
|
||||
var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(endpoint);
|
||||
client.Connect(host, port);
|
||||
await WaitForSchemaClientConnectionAsync(client, endpoint, options.ConnectTimeout).ConfigureAwait(false);
|
||||
client.SendCultNet(new CultNetSnapshotRequestMessage
|
||||
{
|
||||
MessageId = messageId,
|
||||
RecordKeys = new[] { recordKey }
|
||||
});
|
||||
|
||||
return await WaitForPeerSnapshotResponseAsync(completion.Task, endpoint, options.ResponseTimeout)
|
||||
return await FetchSnapshotAsync(
|
||||
createClient,
|
||||
endpoint,
|
||||
ToSnapshotRequestOptions(
|
||||
options,
|
||||
context,
|
||||
options.SchemaIds ?? new[] { descriptor.SchemaId },
|
||||
new[] { recordKey }))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private static string CreateSnapshotMessageId(
|
||||
private static CultMeshSnapshotRequestOptions ToSnapshotRequestOptions(
|
||||
CultMeshPeerSnapshotDocumentOptions options,
|
||||
CultMeshQueryContext context)
|
||||
CultMeshQueryContext context,
|
||||
IReadOnlyList<string>? schemaIds,
|
||||
IReadOnlyList<string>? recordKeys)
|
||||
{
|
||||
var prefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix)
|
||||
? $"cultmesh:{context.RuntimeId}:snapshot"
|
||||
: options.MessageIdPrefix!;
|
||||
return $"{prefix}:{Guid.NewGuid():N}";
|
||||
}
|
||||
|
||||
private static async Task WaitForSchemaClientConnectionAsync(
|
||||
ICultNetSchemaClient client,
|
||||
string endpoint,
|
||||
TimeSpan timeout)
|
||||
{
|
||||
var deadline = DateTimeOffset.UtcNow + timeout;
|
||||
while (!client.Connected)
|
||||
return new CultMeshSnapshotRequestOptions
|
||||
{
|
||||
if (DateTimeOffset.UtcNow >= deadline)
|
||||
{
|
||||
throw new TimeoutException($"Timed out connecting to CultNet snapshot endpoint {endpoint}.");
|
||||
}
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task<CultNetSnapshotResponseRawMessage> WaitForPeerSnapshotResponseAsync(
|
||||
Task<CultNetSnapshotResponseRawMessage> responseTask,
|
||||
string endpoint,
|
||||
TimeSpan timeout)
|
||||
{
|
||||
var timeoutTask = Task.Delay(timeout);
|
||||
var completed = await Task.WhenAny(responseTask, timeoutTask).ConfigureAwait(false);
|
||||
if (completed != responseTask)
|
||||
{
|
||||
throw new TimeoutException($"Timed out waiting for CultNet snapshot response from {endpoint}.");
|
||||
}
|
||||
|
||||
return await responseTask.ConfigureAwait(false);
|
||||
SchemaIds = schemaIds,
|
||||
RecordKeys = recordKeys,
|
||||
ShardId = options.ShardId,
|
||||
ShardEpoch = options.ShardEpoch,
|
||||
ResponseTimeout = options.ResponseTimeout,
|
||||
ConnectTimeout = options.ConnectTimeout,
|
||||
MessageIdPrefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix)
|
||||
? $"cultmesh:{context.RuntimeId}:snapshot"
|
||||
: options.MessageIdPrefix,
|
||||
Security = options.Security,
|
||||
ConfigureClient = options.ConfigureClient,
|
||||
RudpRuntimeId = string.IsNullOrWhiteSpace(options.RudpRuntimeId)
|
||||
? context.RuntimeId
|
||||
: options.RudpRuntimeId,
|
||||
RudpConnectionId = options.RudpConnectionId,
|
||||
RudpConnectPayload = options.RudpConnectPayload,
|
||||
RudpMaxFragmentBytes = options.RudpMaxFragmentBytes,
|
||||
RudpResendDelayMs = options.RudpResendDelayMs
|
||||
};
|
||||
}
|
||||
|
||||
private static TDocument ReadDocumentFromSnapshotResponse<TDocument>(
|
||||
|
|
|
|||
|
|
@ -59,8 +59,186 @@ namespace GameCult.Mesh
|
|||
public long RudpResendDelayMs { get; set; } = 25;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Options for a typed remote snapshot surface bound to one CultNet endpoint.
|
||||
/// </summary>
|
||||
public sealed class CultMeshSnapshotEndpointOptions
|
||||
{
|
||||
/// <summary>Gets or sets the Verse context used by handles from this endpoint.</summary>
|
||||
public CultMeshVerseContext? Context { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the document registry used for raw snapshot payload decoding.</summary>
|
||||
public CultNetDocumentRegistry? DocumentRegistry { get; set; }
|
||||
|
||||
/// <summary>Gets or sets request options applied to each snapshot fetch.</summary>
|
||||
public CultMeshSnapshotRequestOptions? Request { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the source id advertised by diagnostics. Defaults to the endpoint.</summary>
|
||||
public string? SourceId { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the route hint for resulting handles.</summary>
|
||||
public CultMeshRouteHint? RouteHint { get; set; }
|
||||
|
||||
/// <summary>Gets or sets the polling interval for watch fallback.</summary>
|
||||
public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Typed snapshot surface for one remote CultNet endpoint.
|
||||
/// </summary>
|
||||
public sealed class CultMeshSnapshotEndpoint
|
||||
{
|
||||
internal CultMeshSnapshotEndpoint(string endpoint, CultMeshSnapshotEndpointOptions? options)
|
||||
{
|
||||
Endpoint = string.IsNullOrWhiteSpace(endpoint)
|
||||
? throw new ArgumentException("Value must be non-empty.", nameof(endpoint))
|
||||
: endpoint;
|
||||
|
||||
var resolvedOptions = options ?? new CultMeshSnapshotEndpointOptions();
|
||||
Context = resolvedOptions.Context ?? CultMesh.Verse("remote", "cultmesh-snapshot-client").Context;
|
||||
DocumentRegistry = resolvedOptions.DocumentRegistry ?? new CultNetDocumentRegistry();
|
||||
Request = CloneSnapshotRequestOptions(resolvedOptions.Request);
|
||||
RouteHint = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Network, Endpoint);
|
||||
SourceId = string.IsNullOrWhiteSpace(resolvedOptions.SourceId) ? Endpoint : resolvedOptions.SourceId!;
|
||||
PollInterval = resolvedOptions.PollInterval;
|
||||
}
|
||||
|
||||
/// <summary>Gets the endpoint this typed surface reads from.</summary>
|
||||
public string Endpoint { get; }
|
||||
|
||||
/// <summary>Gets the Verse context used by document handles from this endpoint.</summary>
|
||||
public CultMeshVerseContext Context { get; }
|
||||
|
||||
/// <summary>Gets the document registry used for raw snapshot payload decoding.</summary>
|
||||
public CultNetDocumentRegistry DocumentRegistry { get; }
|
||||
|
||||
/// <summary>Gets request options applied to each snapshot fetch.</summary>
|
||||
public CultMeshSnapshotRequestOptions Request { get; }
|
||||
|
||||
/// <summary>Gets the source id advertised by diagnostics.</summary>
|
||||
public string SourceId { get; }
|
||||
|
||||
/// <summary>Gets the route hint for resulting handles.</summary>
|
||||
public CultMeshRouteHint RouteHint { get; }
|
||||
|
||||
/// <summary>Gets the polling interval for watch fallback.</summary>
|
||||
public TimeSpan PollInterval { get; }
|
||||
|
||||
/// <summary>Fetches one raw snapshot with this endpoint's configured request policy.</summary>
|
||||
public Task<CultNetSnapshotResponseRawMessage> FetchSnapshotAsync(
|
||||
IReadOnlyList<string>? schemaIds = null,
|
||||
IReadOnlyList<string>? recordKeys = null)
|
||||
{
|
||||
return CultMesh.FetchSnapshotAsync(Endpoint, CreateRequest(schemaIds, recordKeys));
|
||||
}
|
||||
|
||||
/// <summary>Fetches and decodes documents assignable to the requested type or matching its schema.</summary>
|
||||
public Task<IReadOnlyList<TDocument>> FetchDocumentsAsync<TDocument>(
|
||||
IReadOnlyList<string>? recordKeys = null,
|
||||
IReadOnlyList<string>? schemaIds = null)
|
||||
where TDocument : class
|
||||
{
|
||||
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
|
||||
return CultMesh.FetchSnapshotDocumentsAsync<TDocument>(
|
||||
Endpoint,
|
||||
CreateRequest(schemaIds ?? new[] { descriptor.SchemaId }, recordKeys),
|
||||
DocumentRegistry);
|
||||
}
|
||||
|
||||
/// <summary>Fetches one typed document by record key.</summary>
|
||||
public async Task<TDocument> FetchDocumentAsync<TDocument>(string recordKey)
|
||||
where TDocument : class
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
|
||||
var documents = await FetchDocumentsAsync<TDocument>(new[] { recordKey }).ConfigureAwait(false);
|
||||
return documents.FirstOrDefault()
|
||||
?? throw new InvalidOperationException(
|
||||
$"CultNet snapshot endpoint '{Endpoint}' did not return {typeof(TDocument).FullName} record '{recordKey}'.");
|
||||
}
|
||||
|
||||
/// <summary>Creates a typed document handle over one remote endpoint record.</summary>
|
||||
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
|
||||
string recordKey,
|
||||
string? documentId = null)
|
||||
where TDocument : class
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
|
||||
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
|
||||
var sources = new[]
|
||||
{
|
||||
CultMesh.ProjectionSource($"{SourceId}:{recordKey}", descriptor.SchemaId, "CultNet snapshot endpoint")
|
||||
};
|
||||
var watch = CultMesh.PollingQueryWatcher<CultMeshDocumentQueryParameters, TDocument>(
|
||||
async (_parameters, _context) => await FetchDocumentAsync<TDocument>(recordKey).ConfigureAwait(false),
|
||||
new CultMeshPollingWatchOptions<TDocument>(PollInterval));
|
||||
|
||||
return CultMesh.Document<TDocument>(
|
||||
string.IsNullOrWhiteSpace(documentId) ? recordKey : documentId!,
|
||||
Context,
|
||||
_ => FetchDocumentAsync<TDocument>(recordKey),
|
||||
queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext),
|
||||
sources,
|
||||
RouteHint);
|
||||
}
|
||||
|
||||
/// <summary>Creates a schema-aware catalog over endpoint-backed document handles.</summary>
|
||||
public CultMeshDocumentCatalog Documents(params ICultMeshDocumentHandle[] documents)
|
||||
{
|
||||
return CultMesh.Documents(documents);
|
||||
}
|
||||
|
||||
private CultMeshSnapshotRequestOptions CreateRequest(
|
||||
IReadOnlyList<string>? schemaIds,
|
||||
IReadOnlyList<string>? recordKeys)
|
||||
{
|
||||
var request = CloneSnapshotRequestOptions(Request);
|
||||
request.SchemaIds = schemaIds ?? request.SchemaIds;
|
||||
request.RecordKeys = recordKeys ?? request.RecordKeys;
|
||||
if (string.IsNullOrWhiteSpace(request.RudpRuntimeId))
|
||||
request.RudpRuntimeId = Context.RuntimeId;
|
||||
if (string.IsNullOrWhiteSpace(request.MessageIdPrefix))
|
||||
request.MessageIdPrefix = $"cultmesh:{Context.RuntimeId}:snapshot";
|
||||
return request;
|
||||
}
|
||||
|
||||
private static CultMeshSnapshotRequestOptions CloneSnapshotRequestOptions(CultMeshSnapshotRequestOptions? source)
|
||||
{
|
||||
if (source == null)
|
||||
return new CultMeshSnapshotRequestOptions();
|
||||
|
||||
return new CultMeshSnapshotRequestOptions
|
||||
{
|
||||
SchemaIds = source.SchemaIds,
|
||||
RecordKeys = source.RecordKeys,
|
||||
ShardId = source.ShardId,
|
||||
ShardEpoch = source.ShardEpoch,
|
||||
ResponseTimeout = source.ResponseTimeout,
|
||||
ConnectTimeout = source.ConnectTimeout,
|
||||
MessageIdPrefix = source.MessageIdPrefix,
|
||||
Security = source.Security,
|
||||
ConfigureClient = source.ConfigureClient,
|
||||
CreateClient = source.CreateClient,
|
||||
RudpRuntimeId = source.RudpRuntimeId,
|
||||
RudpConnectionId = source.RudpConnectionId,
|
||||
RudpConnectPayload = source.RudpConnectPayload,
|
||||
RudpMaxFragmentBytes = source.RudpMaxFragmentBytes,
|
||||
RudpResendDelayMs = source.RudpResendDelayMs
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
public static partial class CultMesh
|
||||
{
|
||||
/// <summary>
|
||||
/// Binds one CultNet endpoint as a typed snapshot surface.
|
||||
/// </summary>
|
||||
public static CultMeshSnapshotEndpoint SnapshotEndpoint(
|
||||
string endpoint,
|
||||
CultMeshSnapshotEndpointOptions? options = null)
|
||||
{
|
||||
return new CultMeshSnapshotEndpoint(endpoint, options);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fetches one scoped raw CultNet snapshot from an endpoint.
|
||||
/// </summary>
|
||||
|
|
@ -232,6 +410,7 @@ namespace GameCult.Mesh
|
|||
if (snapshot == null) throw new ArgumentNullException(nameof(snapshot));
|
||||
if (registry == null) throw new ArgumentNullException(nameof(registry));
|
||||
|
||||
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
|
||||
var documents = new List<TDocument>(snapshot.Documents.Length);
|
||||
foreach (var record in snapshot.Documents)
|
||||
{
|
||||
|
|
@ -239,7 +418,12 @@ namespace GameCult.Mesh
|
|||
continue;
|
||||
|
||||
var binding = registry.GetBySchemaId(record.SchemaId);
|
||||
if (binding == null || !typeof(TDocument).IsAssignableFrom(binding.DocumentType))
|
||||
var canDeserializeWithBinding =
|
||||
binding != null &&
|
||||
typeof(TDocument).IsAssignableFrom(binding.DocumentType);
|
||||
var canDeserializeAsSchemaAlias =
|
||||
string.Equals(record.SchemaId, descriptor.SchemaId, StringComparison.Ordinal);
|
||||
if (!canDeserializeWithBinding && !canDeserializeAsSchemaAlias)
|
||||
continue;
|
||||
|
||||
if (!string.Equals(record.PayloadEncoding, "messagepack", StringComparison.Ordinal))
|
||||
|
|
@ -248,7 +432,9 @@ namespace GameCult.Mesh
|
|||
$"CultNet raw document payloadEncoding must be \"messagepack\", not \"{record.PayloadEncoding}\".");
|
||||
}
|
||||
|
||||
documents.Add((TDocument)binding.PayloadDeserializer(record.Payload));
|
||||
documents.Add(canDeserializeWithBinding
|
||||
? (TDocument)binding!.PayloadDeserializer(record.Payload)
|
||||
: CultDocumentMessagePackSerialization.Deserialize<TDocument>(record.Payload));
|
||||
}
|
||||
|
||||
return documents;
|
||||
|
|
|
|||
|
|
@ -1004,7 +1004,8 @@ public sealed class CultMeshStreamingTests
|
|||
handle.RouteHint.Kind.Should().Be(CultMeshLocalityKind.Network);
|
||||
handle.Sources.Should().ContainSingle().Which.SourceId.Should().Be(key.Value);
|
||||
requests.Should().ContainSingle();
|
||||
requests[0].SchemaIds.Should().BeNull();
|
||||
requests[0].SchemaIds.Should().ContainSingle().Which
|
||||
.Should().Be(CultDocumentRegistry.Shared.GetRequired<MeshNoteDocument>().SchemaId);
|
||||
requests[0].RecordKeys.Should().ContainSingle().Which.Should().Be(key.Value);
|
||||
(await catalog.LatestAsync<MeshNoteAliasDocument>()).Text.Should().Be("remote-snapshot");
|
||||
|
||||
|
|
@ -1100,6 +1101,60 @@ public sealed class CultMeshStreamingTests
|
|||
request.MessageId.StartsWith("mesh-test-scoped-snapshot:", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task SnapshotEndpoint_ProvidesTypedHandlesAndSchemaAliases()
|
||||
{
|
||||
var sourceCache = new CultCache();
|
||||
var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared);
|
||||
registry.Register(CultNetDocumentBinding.ForDocument<MeshNoteDocument>(sourceCache.Registry));
|
||||
var key = new CultRecordKey("mesh-note:snapshot-endpoint");
|
||||
await sourceCache.UpsertAsync(new MeshNoteDocument
|
||||
{
|
||||
Schema = "tests.mesh_note.v1",
|
||||
Text = "snapshot-endpoint",
|
||||
Revision = 19
|
||||
}, new CultRecordHandle<MeshNoteDocument>(key));
|
||||
var requests = new List<CultNetSnapshotRequestMessage>();
|
||||
var endpoint = "cultnet://snapshot-endpoint.test:3076";
|
||||
var surface = CultMesh.SnapshotEndpoint(
|
||||
endpoint,
|
||||
new CultMeshSnapshotEndpointOptions
|
||||
{
|
||||
Context = CultMesh.Verse("starbridge", "browser-starfire").Context,
|
||||
DocumentRegistry = registry,
|
||||
Request = new CultMeshSnapshotRequestOptions
|
||||
{
|
||||
ShardId = "primary",
|
||||
ShardEpoch = 9,
|
||||
CreateClient = () => new MeshSnapshotSchemaClient(request =>
|
||||
{
|
||||
requests.Add(request);
|
||||
return registry.CreateRawSnapshotResponse(sourceCache, request.MessageId, request);
|
||||
})
|
||||
}
|
||||
});
|
||||
|
||||
var fetchedAlias = await surface.FetchDocumentAsync<MeshNoteAliasDocument>(key.Value);
|
||||
var aliasHandle = surface.Document<MeshNoteAliasDocument>(key.Value);
|
||||
var aliasLatest = await aliasHandle.LatestAsync();
|
||||
var catalog = surface.Documents(surface.Document<MeshNoteDocument>(key.Value));
|
||||
var catalogAlias = await catalog.LatestAsync<MeshNoteAliasDocument>();
|
||||
|
||||
fetchedAlias.Text.Should().Be("snapshot-endpoint");
|
||||
aliasLatest.Revision.Should().Be(19);
|
||||
catalogAlias.Text.Should().Be("snapshot-endpoint");
|
||||
requests.Should().HaveCount(3);
|
||||
requests.Should().OnlyContain(request =>
|
||||
request.SchemaIds!.SequenceEqual(new[]
|
||||
{
|
||||
CultDocumentRegistry.Shared.GetRequired<MeshNoteAliasDocument>().SchemaId
|
||||
}) &&
|
||||
request.RecordKeys!.SequenceEqual(new[] { key.Value }) &&
|
||||
request.ShardId == "primary" &&
|
||||
request.ShardEpoch == 9 &&
|
||||
request.MessageId.StartsWith("cultmesh:browser-starfire:snapshot:", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task CollectionHandle_ReadsAndWatchesDatabaseCollectionsByIndex()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue