diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs
index 45694e6a..3875e251 100644
--- a/src/GameCult.Mesh/CultMesh.cs
+++ b/src/GameCult.Mesh/CultMesh.cs
@@ -4,6 +4,7 @@ using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
+using System.Reflection;
using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
@@ -133,35 +134,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 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;
+
+ /// 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;
}
///
@@ -337,6 +338,114 @@ namespace GameCult.Mesh
return _host.FlushAsync(soft);
}
+ ///
+ /// Creates a typed document handle over one distributed record hosted by this node.
+ ///
+ public CultMeshDocumentHandle Document(
+ CultRecordKey key,
+ CultMeshVerseContext context,
+ string? documentId = null,
+ IEnumerable? sources = null,
+ CultMeshRouteHint? routeHint = null)
+ where TDocument : class
+ {
+ return CultMesh.Document(Database, key, context, documentId, sources, routeHint);
+ }
+
+ ///
+ /// Creates a typed document handle over one distributed record hosted by this node.
+ ///
+ public CultMeshDocumentHandle Document(
+ CultRecordKey key,
+ CultMeshVerse verse,
+ string? documentId = null,
+ IEnumerable? sources = null,
+ CultMeshRouteHint? routeHint = null)
+ where TDocument : class
+ {
+ if (verse == null) throw new ArgumentNullException(nameof(verse));
+ return Document(key, verse.Context, documentId, sources, routeHint);
+ }
+
+ ///
+ /// Creates a typed document handle over one distributed record hosted by this node.
+ ///
+ public CultMeshDocumentHandle Document(
+ string recordKey,
+ CultMeshVerseContext context,
+ string? documentId = null,
+ IEnumerable? sources = null,
+ CultMeshRouteHint? routeHint = null)
+ where TDocument : class
+ {
+ return Document(new CultRecordKey(recordKey), context, documentId, sources, routeHint);
+ }
+
+ ///
+ /// Creates a typed document handle over one distributed record hosted by this node.
+ ///
+ public CultMeshDocumentHandle Document(
+ string recordKey,
+ CultMeshVerse verse,
+ string? documentId = null,
+ IEnumerable? sources = null,
+ CultMeshRouteHint? routeHint = null)
+ where TDocument : class
+ {
+ if (verse == null) throw new ArgumentNullException(nameof(verse));
+ return Document(recordKey, verse.Context, documentId, sources, routeHint);
+ }
+
+ ///
+ /// Creates a managed reactive document mirror over one distributed record hosted by this node.
+ ///
+ public Task> ReactiveDocumentAsync(
+ CultRecordKey key,
+ CultMeshVerseContext context,
+ CultMeshReactiveDocumentOptions? options = null)
+ where TDocument : class
+ {
+ return Document(key, context).ReactiveAsync(options);
+ }
+
+ ///
+ /// Creates a managed reactive document mirror over one distributed record hosted by this node.
+ ///
+ public Task> ReactiveDocumentAsync(
+ CultRecordKey key,
+ CultMeshVerse verse,
+ CultMeshReactiveDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (verse == null) throw new ArgumentNullException(nameof(verse));
+ return ReactiveDocumentAsync(key, verse.Context, options);
+ }
+
+ ///
+ /// Creates a managed reactive document mirror over one distributed record hosted by this node.
+ ///
+ public Task> ReactiveDocumentAsync(
+ string recordKey,
+ CultMeshVerseContext context,
+ CultMeshReactiveDocumentOptions? options = null)
+ where TDocument : class
+ {
+ return ReactiveDocumentAsync(new CultRecordKey(recordKey), context, options);
+ }
+
+ ///
+ /// Creates a managed reactive document mirror over one distributed record hosted by this node.
+ ///
+ public Task> ReactiveDocumentAsync(
+ string recordKey,
+ CultMeshVerse verse,
+ CultMeshReactiveDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (verse == null) throw new ArgumentNullException(nameof(verse));
+ return ReactiveDocumentAsync(recordKey, verse.Context, options);
+ }
+
///
public void Dispose()
{
@@ -1457,17 +1566,15 @@ namespace GameCult.Mesh
return Document(
ResolveDocumentId(documentId, key),
context,
- async _ => await ReadRequiredAsync(database, key).ConfigureAwait(false),
- _ => database.WatchRecord(key)
- .Where(change => change.Document != null)
- .Select(change => change.Document!),
+ async _ => await ReadDatabaseDocumentRequiredAsync(database, key).ConfigureAwait(false),
+ _ => WatchDatabaseRecordAs(database, key),
async value =>
{
- await database.PutAsync(key, value).ConfigureAwait(false);
+ await PutDatabaseDocumentAsync(database, key, value, predicted: false).ConfigureAwait(false);
},
async value =>
{
- await database.PutPredictedAsync(key, value).ConfigureAwait(false);
+ await PutDatabaseDocumentAsync(database, key, value, predicted: true).ConfigureAwait(false);
},
sourceList,
route);
@@ -2159,44 +2266,44 @@ namespace GameCult.Mesh
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired();
- return await FetchSnapshotAsync(
- createClient,
- endpoint,
- ToSnapshotRequestOptions(
- options,
- context,
- options.SchemaIds ?? new[] { descriptor.SchemaId },
- new[] { recordKey }))
+ return await FetchSnapshotAsync(
+ createClient,
+ endpoint,
+ ToSnapshotRequestOptions(
+ options,
+ context,
+ options.SchemaIds ?? new[] { descriptor.SchemaId },
+ new[] { recordKey }))
.ConfigureAwait(false);
}
- private static CultMeshSnapshotRequestOptions ToSnapshotRequestOptions(
+ private static CultMeshSnapshotRequestOptions ToSnapshotRequestOptions(
CultMeshPeerSnapshotDocumentOptions options,
- CultMeshQueryContext context,
- IReadOnlyList? schemaIds,
- IReadOnlyList? recordKeys)
+ CultMeshQueryContext context,
+ IReadOnlyList? schemaIds,
+ IReadOnlyList? recordKeys)
{
- return new CultMeshSnapshotRequestOptions
+ return new CultMeshSnapshotRequestOptions
{
- 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
- };
+ 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(
@@ -2643,5 +2750,146 @@ namespace GameCult.Mesh
?? throw new KeyNotFoundException(
$"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}.");
}
+
+ private static async Task ReadDatabaseDocumentRequiredAsync(
+ CultNetDatabase database,
+ CultRecordKey key)
+ where TDocument : class
+ {
+ var document = await database.GetAsync(key).ConfigureAwait(false);
+ if (document != null)
+ return document;
+
+ var untyped = database.Cache.Get(key);
+ if (untyped != null && IsSameCultDocumentSchema(untyped.GetType()))
+ return ConvertUntypedDocument(untyped);
+
+ throw new KeyNotFoundException(
+ $"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}.");
+ }
+
+ private static R3.Observable WatchDatabaseRecordAs(
+ CultNetDatabase database,
+ CultRecordKey key)
+ where TDocument : class
+ {
+ var descriptor = CultDocumentRegistry.Shared.GetRequired();
+ return database.WatchAllChanges()
+ .Where(change => TryConvertDatabaseChange(change, key, descriptor, out TDocument? _))
+ .Select(change =>
+ {
+ TryConvertDatabaseChange(change, key, descriptor, out TDocument? document);
+ return document!;
+ });
+ }
+
+ private static bool TryConvertDatabaseChange(
+ object change,
+ CultRecordKey key,
+ CultDocumentDescriptor descriptor,
+ out TDocument? document)
+ where TDocument : class
+ {
+ document = null;
+ if (change == null) return false;
+
+ var changeType = change.GetType();
+ if (changeType.GetProperty("Key")?.GetValue(change) is not CultRecordKey changedKey ||
+ !changedKey.Equals(key))
+ {
+ return false;
+ }
+
+ var current = changeType.GetProperty("Document")?.GetValue(change);
+ if (current == null)
+ return false;
+
+ var schemaId = changeType.GetProperty("SchemaId")?.GetValue(change) as string;
+ if (!IsSameCultDocumentSchema(current.GetType(), descriptor, schemaId))
+ return false;
+
+ document = current as TDocument ?? ConvertUntypedDocument(current);
+ return true;
+ }
+
+ private static bool IsSameCultDocumentSchema(Type documentType)
+ where TDocument : class
+ {
+ return IsSameCultDocumentSchema(
+ documentType,
+ CultDocumentRegistry.Shared.GetRequired());
+ }
+
+ private static bool IsSameCultDocumentSchema(
+ Type documentType,
+ CultDocumentDescriptor descriptor,
+ string? schemaId = null)
+ {
+ if (!string.IsNullOrWhiteSpace(schemaId) &&
+ string.Equals(schemaId, descriptor.SchemaId, StringComparison.Ordinal))
+ {
+ return true;
+ }
+
+ CultDocumentDescriptor storedDescriptor;
+ try
+ {
+ storedDescriptor = CultDocumentRegistry.Shared.GetRequired(documentType);
+ }
+ catch (Exception)
+ {
+ return false;
+ }
+
+ return string.Equals(storedDescriptor.SchemaName, descriptor.SchemaName, StringComparison.Ordinal) &&
+ string.Equals(storedDescriptor.SchemaVersion, descriptor.SchemaVersion, StringComparison.Ordinal);
+ }
+
+ private static TDocument ConvertUntypedDocument(object document)
+ where TDocument : class
+ {
+ var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType());
+ return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TDocument), payload);
+ }
+
+ private static object ConvertUntypedDocument(object document, Type targetType)
+ {
+ var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType());
+ return CultDocumentMessagePackSerialization.DeserializeUntyped(targetType, payload);
+ }
+
+ private static async Task PutDatabaseDocumentAsync(
+ CultNetDatabase database,
+ CultRecordKey key,
+ TDocument value,
+ bool predicted)
+ where TDocument : class
+ {
+ var stored = database.Cache.Get(key);
+ object outgoing = value;
+ var outgoingType = typeof(TDocument);
+ if (stored != null &&
+ !outgoingType.IsInstanceOfType(stored) &&
+ IsSameCultDocumentSchema(stored.GetType(), CultDocumentRegistry.Shared.GetRequired()))
+ {
+ outgoing = ConvertUntypedDocument(value, stored.GetType());
+ outgoingType = stored.GetType();
+ }
+
+ var method = typeof(CultNetDatabase)
+ .GetMethods(BindingFlags.Instance | BindingFlags.Public)
+ .Where(candidate => candidate.Name == (predicted
+ ? nameof(CultNetDatabase.PutPredictedAsync)
+ : nameof(CultNetDatabase.PutAsync)))
+ .Single(candidate =>
+ candidate.IsGenericMethodDefinition &&
+ candidate.GetParameters() is { Length: 2 } parameters &&
+ parameters[0].ParameterType == typeof(CultRecordKey));
+
+ var task = (Task)method
+ .MakeGenericMethod(outgoingType)
+ .Invoke(database, new object[] { key, outgoing })!;
+ await task.ConfigureAwait(false);
+ }
}
}
diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
index 2cfa18b8..a13cd9b7 100644
--- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
+++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
@@ -1397,6 +1397,73 @@ public sealed class CultMeshStreamingTests
cache.Get(key)!.Text.Should().Be("pilot-coalesced-order");
}
+ [Test]
+ public async Task NodeReactiveDocument_OpensSameSchemaAliasWithOneCall()
+ {
+ var schemaId = CultDocumentRegistry.Shared.GetRequired().SchemaId;
+ using var temp = new TemporaryDirectory();
+ using var node = await CultMesh.CreateNodeAsync(
+ Path.Combine(temp.Path, "node-reactive-alias.ccmp"),
+ new CultMeshNodeOptions
+ {
+ StartServer = false,
+ DatabaseOptions = new CultNetDatabaseOptions
+ {
+ RuntimeId = "pilot-a",
+ Shards = new[]
+ {
+ new CultNetShardDescriptor(
+ "pilot-inputs",
+ "server",
+ epoch: 1,
+ isPrimary: false,
+ schemaIds: new[] { schemaId },
+ keyPrefix: "input:")
+ },
+ ClientAuthorityScopes = new[]
+ {
+ new CultNetClientAuthorityScope(
+ "pilot-a",
+ schemaIds: new[] { schemaId },
+ keyPrefix: "input:pilot-a")
+ }
+ }
+ });
+ var key = new CultRecordKey("input:pilot-a:node-reactive");
+ await node.Cache.UpsertAsync(new MeshNoteDocument
+ {
+ Schema = "tests.mesh_note.v1",
+ Text = "node-initial",
+ Revision = 1
+ }, new CultRecordHandle(key));
+
+ using var pilotReactive = await node.ReactiveDocumentAsync(
+ key,
+ CultMesh.Verse("starbridge", "pilot-a"),
+ new CultMeshReactiveDocumentOptions
+ {
+ FlushDelay = TimeSpan.FromMinutes(1)
+ });
+ var commanderHandle = node.Document(
+ key.Value,
+ CultMesh.Verse("starbridge", "commander-rts"));
+ var observedByCommander = new List();
+ using var commanderSubscription = commanderHandle.Watch(observedByCommander.Add);
+
+ pilotReactive.Update(document =>
+ {
+ document.Text = "node-one-call-prediction";
+ document.Revision = 2;
+ });
+ await pilotReactive.FlushAsync();
+
+ await WaitForAsync(() => observedByCommander.Any(document => document.Text == "node-one-call-prediction"));
+ pilotReactive.Document.DocumentId.Should().Be(key.Value);
+ pilotReactive.Document.Context.RuntimeId.Should().Be("pilot-a");
+ node.Cache.Get(key)!.Text.Should().Be("node-one-call-prediction");
+ (await commanderHandle.AsSchemaAlias().LatestAsync()).Revision.Should().Be(2);
+ }
+
[Test]
public async Task ReactiveDocument_SyncedSnapshotEndpointCapturesCanonicalCorrectionAcrossRuntimeBoundary()
{