GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 6

Add node-level reactive document shortcuts

Browse source
This commit is contained in:
Meta Vens 2026-06-27 20:35:59 +01:00
commit 8cc147d367
2 changed files with 377 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

372
src/GameCult.Mesh/CultMesh.cs
View file

@ -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
/// <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 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>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>
@ -337,6 +338,114 @@ namespace GameCult.Mesh
return _host.FlushAsync(soft);
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultRecordKey key,
CultMeshVerseContext context,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
return CultMesh.Document<TDocument>(Database, key, context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultRecordKey key,
CultMeshVerse verse,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document<TDocument>(key, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
string recordKey,
CultMeshVerseContext context,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
return Document<TDocument>(new CultRecordKey(recordKey), context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
string recordKey,
CultMeshVerse verse,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document<TDocument>(recordKey, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
CultRecordKey key,
CultMeshVerseContext context,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
return Document<TDocument>(key, context).ReactiveAsync(options);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
CultRecordKey key,
CultMeshVerse verse,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return ReactiveDocumentAsync<TDocument>(key, verse.Context, options);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
string recordKey,
CultMeshVerseContext context,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
return ReactiveDocumentAsync<TDocument>(new CultRecordKey(recordKey), context, options);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
string recordKey,
CultMeshVerse verse,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return ReactiveDocumentAsync<TDocument>(recordKey, verse.Context, options);
}
/// <inheritdoc />
public void Dispose()
{
@ -1457,17 +1566,15 @@ namespace GameCult.Mesh
return Document<TDocument>(
ResolveDocumentId(documentId, key),
context,
async _ => await ReadRequiredAsync<TDocument>(database, key).ConfigureAwait(false),
_ => database.WatchRecord<TDocument>(key)
.Where(change => change.Document != null)
.Select(change => change.Document!),
async _ => await ReadDatabaseDocumentRequiredAsync<TDocument>(database, key).ConfigureAwait(false),
_ => WatchDatabaseRecordAs<TDocument>(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<TDocument>();
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<string>? schemaIds,
IReadOnlyList<string>? recordKeys)
CultMeshQueryContext context,
IReadOnlyList<string>? schemaIds,
IReadOnlyList<string>? 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<TDocument>(
@ -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<TDocument> ReadDatabaseDocumentRequiredAsync<TDocument>(
CultNetDatabase database,
CultRecordKey key)
where TDocument : class
{
var document = await database.GetAsync<TDocument>(key).ConfigureAwait(false);
if (document != null)
return document;
var untyped = database.Cache.Get(key);
if (untyped != null && IsSameCultDocumentSchema<TDocument>(untyped.GetType()))
return ConvertUntypedDocument<TDocument>(untyped);
throw new KeyNotFoundException(
$"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}.");
}
private static R3.Observable<TDocument> WatchDatabaseRecordAs<TDocument>(
CultNetDatabase database,
CultRecordKey key)
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
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<TDocument>(
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<TDocument>(current);
return true;
}
private static bool IsSameCultDocumentSchema<TDocument>(Type documentType)
where TDocument : class
{
return IsSameCultDocumentSchema(
documentType,
CultDocumentRegistry.Shared.GetRequired<TDocument>());
}
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<TDocument>(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<TDocument>(
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<TDocument>()))
{
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);
}
}
}

67
tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
View file

@ -1397,6 +1397,73 @@ public sealed class CultMeshStreamingTests
cache.Get<MeshNoteDocument>(key)!.Text.Should().Be("pilot-coalesced-order");
}
[Test]
public async Task NodeReactiveDocument_OpensSameSchemaAliasWithOneCall()
{
var schemaId = CultDocumentRegistry.Shared.GetRequired<MeshNoteDocument>().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<MeshNoteDocument>(key));
using var pilotReactive = await node.ReactiveDocumentAsync<MeshNoteAliasDocument>(
key,
CultMesh.Verse("starbridge", "pilot-a"),
new CultMeshReactiveDocumentOptions
{
FlushDelay = TimeSpan.FromMinutes(1)
});
var commanderHandle = node.Document<MeshNoteDocument>(
key.Value,
CultMesh.Verse("starbridge", "commander-rts"));
var observedByCommander = new List<MeshNoteDocument>();
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<MeshNoteDocument>(key)!.Text.Should().Be("node-one-call-prediction");
(await commanderHandle.AsSchemaAlias<MeshNoteAliasDocument>().LatestAsync()).Revision.Should().Be(2);
}
[Test]
public async Task ReactiveDocument_SyncedSnapshotEndpointCapturesCanonicalCorrectionAcrossRuntimeBoundary()
{