diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs
index 6d392918..ecc79c49 100644
--- a/src/GameCult.Mesh/CultMesh.cs
+++ b/src/GameCult.Mesh/CultMesh.cs
@@ -133,11 +133,35 @@ namespace GameCult.Mesh
/// Gets or sets the message id prefix used for snapshot requests.
public string? MessageIdPrefix { get; set; }
+ /// Gets or sets schema ids to request. Empty means the document type schema is requested.
+ public IReadOnlyList? SchemaIds { get; set; }
+
+ /// Gets or sets the target shard id, when the endpoint is shard-aware.
+ public string? ShardId { get; set; }
+
+ /// Gets or sets the target shard epoch, when the endpoint is shard-aware.
+ public long? ShardEpoch { get; set; }
+
/// Gets or sets client security options for endpoint-created clients.
public ClientSecurityOptions? Security { get; set; }
/// Gets or sets a callback used to configure endpoint-created LiteNetLib clients.
public Action? ConfigureClient { get; set; }
+
+ /// Gets or sets the runtime id used by endpoint-created RUDP clients.
+ public string? RudpRuntimeId { get; set; }
+
+ /// Gets or sets the connection id used by endpoint-created RUDP clients.
+ public uint RudpConnectionId { get; set; } = 0x43554c54;
+
+ /// Gets or sets the connect payload used by endpoint-created RUDP clients.
+ public string RudpConnectPayload { get; set; } = "cultnet-schema-rudp";
+
+ /// Gets or sets the maximum fragment size used by endpoint-created RUDP clients.
+ public int RudpMaxFragmentBytes { get; set; } = 1024;
+
+ /// Gets or sets the resend delay used by endpoint-created RUDP clients.
+ public long RudpResendDelayMs { get; set; } = 25;
}
///
@@ -2086,73 +2110,44 @@ namespace GameCult.Mesh
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired();
- var messageId = CreateSnapshotMessageId(options, context);
- var completion = new TaskCompletionSource(
- TaskCreationOptions.RunContinuationsAsynchronously);
- using var client = createClient();
- client.OnCultNet(response =>
- {
- if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
- {
- completion.TrySetResult(response);
- }
- });
- client.OnCultNet(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? schemaIds,
+ IReadOnlyList? 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 WaitForPeerSnapshotResponseAsync(
- Task 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(
diff --git a/src/GameCult.Mesh/CultMeshSnapshots.cs b/src/GameCult.Mesh/CultMeshSnapshots.cs
index 986b901f..bd595799 100644
--- a/src/GameCult.Mesh/CultMeshSnapshots.cs
+++ b/src/GameCult.Mesh/CultMeshSnapshots.cs
@@ -59,8 +59,186 @@ namespace GameCult.Mesh
public long RudpResendDelayMs { get; set; } = 25;
}
+ ///
+ /// Options for a typed remote snapshot surface bound to one CultNet endpoint.
+ ///
+ public sealed class CultMeshSnapshotEndpointOptions
+ {
+ /// Gets or sets the Verse context used by handles from this endpoint.
+ public CultMeshVerseContext? Context { get; set; }
+
+ /// Gets or sets the document registry used for raw snapshot payload decoding.
+ public CultNetDocumentRegistry? DocumentRegistry { get; set; }
+
+ /// Gets or sets request options applied to each snapshot fetch.
+ public CultMeshSnapshotRequestOptions? Request { get; set; }
+
+ /// Gets or sets the source id advertised by diagnostics. Defaults to the endpoint.
+ public string? SourceId { get; set; }
+
+ /// Gets or sets the route hint for resulting handles.
+ public CultMeshRouteHint? RouteHint { get; set; }
+
+ /// Gets or sets the polling interval for watch fallback.
+ public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
+ }
+
+ ///
+ /// Typed snapshot surface for one remote CultNet endpoint.
+ ///
+ 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;
+ }
+
+ /// Gets the endpoint this typed surface reads from.
+ public string Endpoint { get; }
+
+ /// Gets the Verse context used by document handles from this endpoint.
+ public CultMeshVerseContext Context { get; }
+
+ /// Gets the document registry used for raw snapshot payload decoding.
+ public CultNetDocumentRegistry DocumentRegistry { get; }
+
+ /// Gets request options applied to each snapshot fetch.
+ public CultMeshSnapshotRequestOptions Request { get; }
+
+ /// Gets the source id advertised by diagnostics.
+ public string SourceId { get; }
+
+ /// Gets the route hint for resulting handles.
+ public CultMeshRouteHint RouteHint { get; }
+
+ /// Gets the polling interval for watch fallback.
+ public TimeSpan PollInterval { get; }
+
+ /// Fetches one raw snapshot with this endpoint's configured request policy.
+ public Task FetchSnapshotAsync(
+ IReadOnlyList? schemaIds = null,
+ IReadOnlyList? recordKeys = null)
+ {
+ return CultMesh.FetchSnapshotAsync(Endpoint, CreateRequest(schemaIds, recordKeys));
+ }
+
+ /// Fetches and decodes documents assignable to the requested type or matching its schema.
+ public Task> FetchDocumentsAsync(
+ IReadOnlyList? recordKeys = null,
+ IReadOnlyList? schemaIds = null)
+ where TDocument : class
+ {
+ var descriptor = CultDocumentRegistry.Shared.GetRequired();
+ return CultMesh.FetchSnapshotDocumentsAsync(
+ Endpoint,
+ CreateRequest(schemaIds ?? new[] { descriptor.SchemaId }, recordKeys),
+ DocumentRegistry);
+ }
+
+ /// Fetches one typed document by record key.
+ public async Task FetchDocumentAsync(string recordKey)
+ where TDocument : class
+ {
+ if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
+ var documents = await FetchDocumentsAsync(new[] { recordKey }).ConfigureAwait(false);
+ return documents.FirstOrDefault()
+ ?? throw new InvalidOperationException(
+ $"CultNet snapshot endpoint '{Endpoint}' did not return {typeof(TDocument).FullName} record '{recordKey}'.");
+ }
+
+ /// Creates a typed document handle over one remote endpoint record.
+ public CultMeshDocumentHandle Document(
+ 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();
+ var sources = new[]
+ {
+ CultMesh.ProjectionSource($"{SourceId}:{recordKey}", descriptor.SchemaId, "CultNet snapshot endpoint")
+ };
+ var watch = CultMesh.PollingQueryWatcher(
+ async (_parameters, _context) => await FetchDocumentAsync(recordKey).ConfigureAwait(false),
+ new CultMeshPollingWatchOptions(PollInterval));
+
+ return CultMesh.Document(
+ string.IsNullOrWhiteSpace(documentId) ? recordKey : documentId!,
+ Context,
+ _ => FetchDocumentAsync(recordKey),
+ queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext),
+ sources,
+ RouteHint);
+ }
+
+ /// Creates a schema-aware catalog over endpoint-backed document handles.
+ public CultMeshDocumentCatalog Documents(params ICultMeshDocumentHandle[] documents)
+ {
+ return CultMesh.Documents(documents);
+ }
+
+ private CultMeshSnapshotRequestOptions CreateRequest(
+ IReadOnlyList? schemaIds,
+ IReadOnlyList? 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
{
+ ///
+ /// Binds one CultNet endpoint as a typed snapshot surface.
+ ///
+ public static CultMeshSnapshotEndpoint SnapshotEndpoint(
+ string endpoint,
+ CultMeshSnapshotEndpointOptions? options = null)
+ {
+ return new CultMeshSnapshotEndpoint(endpoint, options);
+ }
+
///
/// Fetches one scoped raw CultNet snapshot from an endpoint.
///
@@ -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();
var documents = new List(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(record.Payload));
}
return documents;
diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
index c1a9240d..c93f5ee7 100644
--- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
+++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
@@ -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().SchemaId);
requests[0].RecordKeys.Should().ContainSingle().Which.Should().Be(key.Value);
(await catalog.LatestAsync()).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(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(key));
+ var requests = new List();
+ 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(key.Value);
+ var aliasHandle = surface.Document(key.Value);
+ var aliasLatest = await aliasHandle.LatestAsync();
+ var catalog = surface.Documents(surface.Document(key.Value));
+ var catalogAlias = await catalog.LatestAsync();
+
+ 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().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()
{