diff --git a/Directory.Build.props b/Directory.Build.props index a3840327..8b298d76 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -14,7 +14,7 @@ MIT © Meta Vens and contributors true - 1.0.19 + 1.0.20 true diff --git a/src/GameCult.Mesh/CultMeshBodies.cs b/src/GameCult.Mesh/CultMeshBodies.cs index 2ab935db..4f2b3514 100644 --- a/src/GameCult.Mesh/CultMeshBodies.cs +++ b/src/GameCult.Mesh/CultMeshBodies.cs @@ -185,7 +185,7 @@ namespace GameCult.Mesh CultMeshBodyDescriptor networkFallback, CultMeshBodyValidationRequest request) { - return NegotiateReadOnly(null, preferred, networkFallback, request); + return NegotiateReadOnly(null, new[] { preferred, networkFallback }, request); } internal CultMeshBodyNegotiationResult NegotiateReadOnly( @@ -194,12 +194,33 @@ namespace GameCult.Mesh CultMeshBodyDescriptor networkFallback, CultMeshBodyValidationRequest request) { - if (preferred == null) throw new ArgumentNullException(nameof(preferred)); - if (networkFallback == null) throw new ArgumentNullException(nameof(networkFallback)); - if (!SameLogicalGeneration(preferred, networkFallback)) - throw new InvalidOperationException("CultMesh local and network representations describe different logical body generations."); - CultMeshBodyDescriptorValidator.Validate(preferred, request); - CultMeshBodyDescriptorValidator.Validate(networkFallback, request); + return NegotiateReadOnly(producerId, new[] { preferred, networkFallback }, request); + } + + /// + /// Selects the first representation the consumer can validate and open. Representation order is + /// publisher preference; adapter availability and capability validity decide the actual plane. + /// + public CultMeshBodyNegotiationResult NegotiateReadOnly( + IReadOnlyList representations, + CultMeshBodyValidationRequest request) => + NegotiateReadOnly(null, representations, request); + + internal CultMeshBodyNegotiationResult NegotiateReadOnly( + string? producerId, + IReadOnlyList representations, + CultMeshBodyValidationRequest request) + { + if (representations == null) throw new ArgumentNullException(nameof(representations)); + if (representations.Count == 0 || representations.Any(value => value == null)) + throw new ArgumentException("At least one body representation is required.", nameof(representations)); + var preferred = representations[0]; + foreach (var representation in representations) + { + if (!SameLogicalGeneration(preferred, representation)) + throw new InvalidOperationException("CultMesh representations describe different logical body generations."); + CultMeshBodyDescriptorValidator.Validate(representation, request); + } var authorized = _authorizeNamedProducer == null ? _authorizeProducer(preferred) : !string.IsNullOrWhiteSpace(producerId) && _authorizeNamedProducer(producerId!, preferred); @@ -207,24 +228,29 @@ namespace GameCult.Mesh throw new UnauthorizedAccessException("CultMesh body producer is not authorized for this logical body."); Exception? preferredFailure = null; - try + foreach (var representation in representations) { - if (!_adapters.TryGetValue(preferred.TransportKind, out var local)) - throw new NotSupportedException($"No CultMesh {preferred.TransportKind} body adapter is available."); - if (!local.CanOpen(preferred)) - throw new NotSupportedException($"The CultMesh {preferred.TransportKind} body adapter declined the preferred descriptor."); - var lease = local.OpenReadOnly(preferred, request); - return new CultMeshBodyNegotiationResult(lease, preferred.TransportKind, null); - } - catch (Exception error) when (!(error is UnauthorizedAccessException)) - { - preferredFailure = error; + try + { + if (!_adapters.TryGetValue(representation.TransportKind, out var adapter)) + throw new NotSupportedException($"No CultMesh {representation.TransportKind} body adapter is available."); + if (!adapter.CanOpen(representation)) + throw new NotSupportedException($"The CultMesh {representation.TransportKind} body adapter declined its descriptor."); + var lease = adapter.OpenReadOnly(representation, request); + return new CultMeshBodyNegotiationResult( + lease, + preferred.TransportKind, + preferredFailure); + } + catch (Exception error) when (!(error is UnauthorizedAccessException)) + { + preferredFailure ??= error; + } } - if (!_adapters.TryGetValue(CultMeshBodyTransportKind.Network, out var network) || !network.CanOpen(networkFallback)) - throw new NotSupportedException("No CultMesh network body fallback is available."); - var fallback = network.OpenReadOnly(networkFallback, request); - return new CultMeshBodyNegotiationResult(fallback, preferred.TransportKind, preferredFailure); + throw new NotSupportedException( + "No advertised CultMesh body representation could be opened by this consumer.", + preferredFailure); } private static bool SameLogicalGeneration(CultMeshBodyDescriptor left, CultMeshBodyDescriptor right) => diff --git a/src/GameCult.Mesh/CultMeshBodyDemand.cs b/src/GameCult.Mesh/CultMeshBodyDemand.cs new file mode 100644 index 00000000..049d762a --- /dev/null +++ b/src/GameCult.Mesh/CultMeshBodyDemand.cs @@ -0,0 +1,142 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using GameCult.Networking; + +#nullable enable + +namespace GameCult.Mesh +{ + /// Derived transport work for one logical hot body. + public sealed class CultMeshBodyDemandPlan + { + internal CultMeshBodyDemandPlan( + string bodyId, + IReadOnlyList consumers) + { + BodyId = bodyId; + Consumers = consumers; + RequiresSharedMemory = consumers.Any(value => value.Transport == CultMeshBodyTransportKind.SharedMemory); + RequiresNetwork = consumers.Any(value => value.Transport == CultMeshBodyTransportKind.Network); + } + + public string BodyId { get; } + public bool HasConsumers => Consumers.Count > 0; + public bool RequiresSharedMemory { get; } + public bool RequiresNetwork { get; } + public IReadOnlyList Consumers { get; } + } + + /// One consumer's owner-derived route for one logical body. + public sealed class CultMeshBodyConsumerRoute + { + internal CultMeshBodyConsumerRoute( + string consumerRuntimeId, + string subscriptionId, + CultMeshBodyTransportKind transport) + { + ConsumerRuntimeId = consumerRuntimeId; + SubscriptionId = subscriptionId; + Transport = transport; + } + + public string ConsumerRuntimeId { get; } + public string SubscriptionId { get; } + public CultMeshBodyTransportKind Transport { get; } + } + + /// + /// Owns the live projection from exact state subscriptions to required body transport planes. + /// A same-machine claim is transport evidence only; shared memory is selected only when both + /// the server peer and consumer capability agree. + /// + public sealed class CultMeshBodyDemandTracker : IDisposable + { + private readonly CultNetDatabaseSubscriptionServer? _subscriptions; + private readonly object _gate = new(); + private readonly Dictionary _demands = + new(StringComparer.Ordinal); + private bool _disposed; + + public CultMeshBodyDemandTracker(CultNetDatabaseSubscriptionServer subscriptions) + { + _subscriptions = subscriptions ?? throw new ArgumentNullException(nameof(subscriptions)); + _subscriptions.DemandChanged += OnDemandChanged; + } + + /// Creates a tracker fed explicitly by a subscription/session control plane. + public CultMeshBodyDemandTracker() + { + } + + public CultMeshBodyDemandPlan Plan(string bodyId) + { + if (string.IsNullOrWhiteSpace(bodyId)) throw new ArgumentException("Body identity is required.", nameof(bodyId)); + lock (_gate) + { + ThrowIfDisposed(); + var routes = _demands.Values + .Where(value => value.BodyIds.Contains(bodyId, StringComparer.Ordinal)) + .Select(value => Route(value)) + .Where(value => value != null) + .Cast() + .OrderBy(value => value.ConsumerRuntimeId, StringComparer.Ordinal) + .ThenBy(value => value.SubscriptionId, StringComparer.Ordinal) + .ToArray(); + return new CultMeshBodyDemandPlan(bodyId, routes); + } + } + + public void Dispose() + { + lock (_gate) + { + if (_disposed) return; + _disposed = true; + if (_subscriptions != null) _subscriptions.DemandChanged -= OnDemandChanged; + _demands.Clear(); + } + } + + /// Applies one active or withdrawn typed body-demand observation. + public void Observe(CultNetDatabaseSubscriptionDemand demand) + { + if (demand == null) throw new ArgumentNullException(nameof(demand)); + var key = demand.ConsumerRuntimeId + "\u001f" + demand.SubscriptionId; + lock (_gate) + { + if (_disposed) return; + if (demand.Active) _demands[key] = demand; + else _demands.Remove(key); + } + } + + private void OnDemandChanged(CultNetDatabaseSubscriptionDemand demand) => Observe(demand); + + private static CultMeshBodyConsumerRoute? Route(CultNetDatabaseSubscriptionDemand demand) + { + var supported = demand.SupportedBodyTransports; + if (demand.SameMachine && Supports(supported, CultMeshBodyTransportKind.SharedMemory)) + return new CultMeshBodyConsumerRoute( + demand.ConsumerRuntimeId, + demand.SubscriptionId, + CultMeshBodyTransportKind.SharedMemory); + if (Supports(supported, CultMeshBodyTransportKind.Network)) + return new CultMeshBodyConsumerRoute( + demand.ConsumerRuntimeId, + demand.SubscriptionId, + CultMeshBodyTransportKind.Network); + return null; + } + + private static bool Supports( + IReadOnlyList supported, + CultMeshBodyTransportKind transport) => + supported.Any(value => string.Equals(value, transport.ToString(), StringComparison.OrdinalIgnoreCase)); + + private void ThrowIfDisposed() + { + if (_disposed) throw new ObjectDisposedException(nameof(CultMeshBodyDemandTracker)); + } + } +} diff --git a/src/GameCult.Mesh/CultMeshBodyPublications.cs b/src/GameCult.Mesh/CultMeshBodyPublications.cs index 48d6e7b8..37f2955f 100644 --- a/src/GameCult.Mesh/CultMeshBodyPublications.cs +++ b/src/GameCult.Mesh/CultMeshBodyPublications.cs @@ -13,7 +13,7 @@ namespace GameCult.Mesh { public static class CultMeshBodyPublicationSchemaVersions { - public const string Publication = "gamecult.mesh.body_publication.v1"; + public const string Publication = "gamecult.mesh.body_publication.v2"; public const string NetworkBody = "gamecult.mesh.network_body.v1"; } @@ -241,8 +241,7 @@ namespace GameCult.Mesh [Key(7)] public long Sequence { get; set; } [Key(8)] public CultMeshBodySynchronization Synchronization { get; set; } [Key(9)] public long LivenessExpiresAtUnixMs { get; set; } - [Key(10)] public CultMeshBodyDescriptor PreferredLocal { get; set; } = new(); - [Key(11)] public CultMeshBodyDescriptor NetworkFallback { get; set; } = new(); + [Key(10)] public CultMeshBodyDescriptor[] Representations { get; set; } = Array.Empty(); public static CultRecordKey CreateRecordKey(string bodyId, long producerEpoch, long sequence) { @@ -326,12 +325,13 @@ namespace GameCult.Mesh throw new InvalidOperationException("CultMesh body publication generation values must be non-negative."); if (expectedRecordKey.HasValue && !publication.RecordKey.Equals(expectedRecordKey.Value)) throw new InvalidOperationException("CultMesh body publication record key disagrees with its generation envelope."); - ValidateDescriptor(publication, publication.PreferredLocal, "preferred local"); - ValidateDescriptor(publication, publication.NetworkFallback, "network fallback"); - if (publication.PreferredLocal.TransportKind == CultMeshBodyTransportKind.Network) - throw new InvalidOperationException("CultMesh preferred local body descriptor must use a local transport."); - if (publication.NetworkFallback.TransportKind != CultMeshBodyTransportKind.Network) - throw new InvalidOperationException("CultMesh network fallback body descriptor must use network transport."); + var representations = publication.Representations ?? Array.Empty(); + if (representations.Length == 0) + throw new InvalidOperationException("CultMesh body publication has no transport representations."); + if (representations.Select(value => value?.TransportKind).Distinct().Count() != representations.Length) + throw new InvalidOperationException("CultMesh body publication advertises the same transport representation more than once."); + foreach (var representation in representations) + ValidateDescriptor(publication, representation, representation?.TransportKind.ToString() ?? "missing"); } private static void ValidateDescriptor( @@ -376,8 +376,7 @@ namespace GameCult.Mesh throw new InvalidOperationException("CultMesh body publication is no longer live."); return _transport.NegotiateReadOnly( publication.ProducerId, - publication.PreferredLocal, - publication.NetworkFallback, + publication.Representations, request); } } diff --git a/src/GameCult.Mesh/CultMeshFrameBodies.cs b/src/GameCult.Mesh/CultMeshFrameBodies.cs index da188317..e01a0901 100644 --- a/src/GameCult.Mesh/CultMeshFrameBodies.cs +++ b/src/GameCult.Mesh/CultMeshFrameBodies.cs @@ -1,10 +1,12 @@ using System; +using System.Buffers; using System.Collections.Generic; using System.IO; using System.IO.MemoryMappedFiles; using System.Security.Cryptography; using System.Text; using System.Threading; +using Microsoft.Win32.SafeHandles; #nullable enable @@ -18,6 +20,7 @@ namespace GameCult.Mesh private const int HeaderSize = 128; private const int SequenceOffset = 8; private const int ReaderCountOffset = 32; + private const int WriterCountOffset = 48; private const int ContractHashOffset = 64; private readonly object _gate = new(); private readonly MemoryMappedFile _mapping; @@ -27,6 +30,7 @@ namespace GameCult.Mesh private readonly string _token; private int _cursor; private long _nextSequence; + private CultMeshFrameBodyWriteLease? _activeWrite; private bool _disposed; public CultMeshFrameBodyPublisher( @@ -68,42 +72,132 @@ namespace GameCult.Mesh public int SlotByteLength { get; } public bool TryPublish(ReadOnlySpan body, DateTimeOffset nowUtc, out CultMeshBodyDescriptor descriptor) + { + if (body.Length > SlotByteLength) throw new ArgumentOutOfRangeException(nameof(body)); + if (!TryAcquireWrite(out var write)) + { + descriptor = null!; + return false; + } + using (write) + { + body.CopyTo(write.Span); + descriptor = write.Commit(body.Length, nowUtc); + return true; + } + } + + /// + /// Reserves one producer-owned shared-memory slot for direct writes. The caller must commit or dispose the lease. + /// + public bool TryAcquireWrite(out CultMeshFrameBodyWriteLease lease) { lock (_gate) { ThrowIfDisposed(); - if (body.Length > SlotByteLength) throw new ArgumentOutOfRangeException(nameof(body)); - var publishedBytes = body.ToArray(); - CultMeshBodyDescriptor? published = null; + if (_activeWrite != null) + { + lease = null!; + return false; + } + + var slot = -1; WithMutex(() => { for (var attempt = 0; attempt < CultMeshFrameRegion.SlotCount; attempt++) { - var slot = (_cursor + attempt) % CultMeshFrameRegion.SlotCount; - if (_control.ReadInt32(ReaderCountOffset + slot * sizeof(int)) != 0) continue; - using (var view = _mapping.CreateViewAccessor(SlotOffset(slot, SlotByteLength), SlotByteLength, MemoryMappedFileAccess.Write)) - view.WriteArray(0, publishedBytes, 0, publishedBytes.Length); - _control.Write(SequenceOffset + slot * sizeof(long), _nextSequence); - _cursor = (slot + 1) % CultMeshFrameRegion.SlotCount; - published = Descriptor(slot, publishedBytes, nowUtc); - _nextSequence++; + var candidate = (_cursor + attempt) % CultMeshFrameRegion.SlotCount; + if (_control.ReadInt32(ReaderCountOffset + candidate * sizeof(int)) != 0 || + _control.ReadInt32(WriterCountOffset + candidate * sizeof(int)) != 0) + continue; + _control.Write(WriterCountOffset + candidate * sizeof(int), 1); + slot = candidate; + _cursor = (candidate + 1) % CultMeshFrameRegion.SlotCount; return; } }); - descriptor = published!; - return published != null; + if (slot < 0) + { + lease = null!; + return false; + } + + try + { + var view = _mapping.CreateViewAccessor( + SlotOffset(slot, SlotByteLength), + SlotByteLength, + MemoryMappedFileAccess.ReadWrite); + lease = new CultMeshFrameBodyWriteLease(this, view, slot, _nextSequence, SlotByteLength); + _activeWrite = lease; + return true; + } + catch + { + WithMutex(() => _control.Write(WriterCountOffset + slot * sizeof(int), 0)); + throw; + } } } - private CultMeshBodyDescriptor Descriptor(int slot, ReadOnlySpan body, DateTimeOffset nowUtc) => new() + internal CultMeshBodyDescriptor Commit( + CultMeshFrameBodyWriteLease lease, + int byteLength, + DateTimeOffset nowUtc) { - BodyId = BodyId, SchemaId = SchemaId, LayoutVersion = LayoutVersion, ByteSize = body.Length, - Capacity = Capacity, ProducerEpoch = ProducerEpoch, Sequence = _nextSequence, + lock (_gate) + { + ThrowIfDisposed(); + if (!ReferenceEquals(_activeWrite, lease) || lease.Sequence != _nextSequence) + throw new InvalidOperationException("The CultMesh frame write lease is stale or is not owned by this publisher."); + if (byteLength < 0 || byteLength > SlotByteLength) + throw new ArgumentOutOfRangeException(nameof(byteLength)); + var semanticHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(lease.Span[..byteLength]); + WithMutex(() => + { + var writerOffset = WriterCountOffset + lease.SlotIndex * sizeof(int); + if (_control.ReadInt32(writerOffset) != 1 || + _control.ReadInt32(ReaderCountOffset + lease.SlotIndex * sizeof(int)) != 0) + throw new InvalidOperationException("The CultMesh frame write reservation was lost before commit."); + _control.Write(SequenceOffset + lease.SlotIndex * sizeof(long), lease.Sequence); + _control.Write(writerOffset, 0); + }); + var descriptor = Descriptor(lease.SlotIndex, byteLength, lease.Sequence, semanticHash, nowUtc); + _nextSequence++; + _activeWrite = null; + lease.Complete(); + return descriptor; + } + } + + internal void Abandon(CultMeshFrameBodyWriteLease lease) + { + lock (_gate) + { + if (!ReferenceEquals(_activeWrite, lease)) return; + if (!_disposed) + { + WithMutex(() => + _control.Write(WriterCountOffset + lease.SlotIndex * sizeof(int), 0)); + } + _activeWrite = null; + } + } + + private CultMeshBodyDescriptor Descriptor( + int slot, + int byteLength, + long sequence, + string semanticHash, + DateTimeOffset nowUtc) => new() + { + BodyId = BodyId, SchemaId = SchemaId, LayoutVersion = LayoutVersion, ByteSize = byteLength, + Capacity = Capacity, ProducerEpoch = ProducerEpoch, Sequence = sequence, AccessMode = CultMeshBodyAccessMode.ReadOnly, Synchronization = CultMeshBodySynchronization.TripleBuffer, LeaseExpiresAtUnixMs = nowUtc.Add(_leaseDuration).ToUnixTimeMilliseconds(), TransportKind = CultMeshBodyTransportKind.SharedMemory, CapabilityToken = _token + "." + slot, - SemanticHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(body) + SemanticHash = semanticHash }; internal static bool IsFrameCapability(string capability) => capability.StartsWith(CapabilityPrefix, StringComparison.Ordinal); @@ -178,6 +272,8 @@ namespace GameCult.Mesh { if (_disposed) return; _disposed = true; + _activeWrite?.Invalidate(); + _activeWrite = null; _control.Dispose(); _mapping.Dispose(); _mutex.Dispose(); @@ -214,6 +310,100 @@ namespace GameCult.Mesh private static string Require(string value, string name) => string.IsNullOrWhiteSpace(value) ? throw new ArgumentException("Value is required.", name) : value; } + /// A bounded producer lease over one directly writable shared-memory frame slot. + public sealed unsafe class CultMeshFrameBodyWriteLease : IDisposable + { + private CultMeshFrameBodyPublisher? _owner; + private readonly MemoryMappedViewAccessor _view; + private readonly CultMeshMappedMemoryManager _memory; + private bool _completed; + + internal CultMeshFrameBodyWriteLease( + CultMeshFrameBodyPublisher owner, + MemoryMappedViewAccessor view, + int slotIndex, + long sequence, + int byteLength) + { + _owner = owner; + _view = view; + _memory = new CultMeshMappedMemoryManager(view, byteLength); + SlotIndex = slotIndex; + Sequence = sequence; + } + + public int SlotIndex { get; } + public long Sequence { get; } + public Memory Memory => _memory.Memory; + public Span Span => Memory.Span; + + public CultMeshBodyDescriptor Commit(int byteLength, DateTimeOffset nowUtc) => + (_owner ?? throw new ObjectDisposedException(nameof(CultMeshFrameBodyWriteLease))) + .Commit(this, byteLength, nowUtc); + + internal void Complete() + { + _completed = true; + ReleaseResources(); + } + + internal void Invalidate() + { + _owner = null; + ReleaseResources(); + } + + public void Dispose() + { + var owner = Interlocked.Exchange(ref _owner, null); + if (!_completed) owner?.Abandon(this); + ReleaseResources(); + } + + private void ReleaseResources() + { + ((IDisposable)_memory).Dispose(); + _view.Dispose(); + _owner = null; + } + } + + internal sealed unsafe class CultMeshMappedMemoryManager : MemoryManager + { + private SafeMemoryMappedViewHandle? _handle; + private byte* _pointer; + private readonly int _length; + + public CultMeshMappedMemoryManager(MemoryMappedViewAccessor view, int length) + { + _handle = view.SafeMemoryMappedViewHandle; + _length = length; + byte* pointer = null; + _handle.AcquirePointer(ref pointer); + _pointer = pointer + view.PointerOffset; + } + + public override Span GetSpan() => + _handle == null ? throw new ObjectDisposedException(nameof(CultMeshMappedMemoryManager)) : new Span(_pointer, _length); + + public override MemoryHandle Pin(int elementIndex = 0) + { + if ((uint)elementIndex > (uint)_length) throw new ArgumentOutOfRangeException(nameof(elementIndex)); + if (_handle == null) throw new ObjectDisposedException(nameof(CultMeshMappedMemoryManager)); + return new MemoryHandle(_pointer + elementIndex); + } + + public override void Unpin() { } + + protected override void Dispose(bool disposing) + { + var handle = Interlocked.Exchange(ref _handle, null); + if (handle == null) return; + handle.ReleasePointer(); + _pointer = null; + } + } + public sealed class CultMeshFrameGeneration { internal CultMeshFrameGeneration( diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index 165c9189..dcc58143 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -3,6 +3,7 @@ using System.Collections; using System.Collections.Generic; using System.Globalization; using System.Linq; +using System.Linq.Expressions; using System.Reflection; using System.Threading; using System.Threading.Tasks; @@ -1002,6 +1003,27 @@ namespace GameCult.Mesh return Watch().Subscribe(onNext); } + /// + /// Projects one typed document field into a stable reactive state pointer. The expression is + /// structural metadata: only member access and constant collection indices are accepted. + /// + public CultMeshStatePointer Field(Expression> field) + { + if (field == null) throw new ArgumentNullException(nameof(field)); + var path = CultMeshFieldPath.Parse(field.Body, field.Parameters[0]); + var read = field.Compile(); + return new CultMeshStatePointer( + DocumentId + "#" + path, + async context => read(await Feed + .SnapshotAsync(CultMeshDocumentQueryParameters.Empty, context) + .ConfigureAwait(false)), + context => Feed + .Watch(CultMeshDocumentQueryParameters.Empty, context) + .Select(read), + RouteHint, + Sources); + } + /// Replaces the underlying document when this handle is backed by mutable state. public Task ReplaceAsync(TDocument value) { @@ -1122,6 +1144,62 @@ namespace GameCult.Mesh payload); } + private static class CultMeshFieldPath + { + public static string Parse(Expression expression, ParameterExpression root) + { + var segments = new Stack(); + while (expression != root) + { + if (expression is MemberExpression member) + { + segments.Push(member.Member.Name); + expression = member.Expression + ?? throw Invalid(); + continue; + } + + if (expression is MethodCallExpression call && + call.Method.Name == "get_Item" && + call.Object != null && + call.Arguments.Count == 1) + { + segments.Push("[" + ConstantIndex(call.Arguments[0]) + "]"); + expression = call.Object; + continue; + } + + if (expression is BinaryExpression binary && binary.NodeType == ExpressionType.ArrayIndex) + { + segments.Push("[" + ConstantIndex(binary.Right) + "]"); + expression = binary.Left; + continue; + } + + throw Invalid(); + } + + if (segments.Count == 0) throw Invalid(); + var path = string.Empty; + foreach (var segment in segments) + path += segment.StartsWith("[", StringComparison.Ordinal) || path.Length == 0 + ? segment + : "." + segment; + return path; + } + + private static string ConstantIndex(Expression expression) + { + if (expression is ConstantExpression constant && constant.Value != null) + return Convert.ToString(constant.Value, CultureInfo.InvariantCulture)!; + throw Invalid(); + } + + private static ArgumentException Invalid() => new( + "CultMesh field selectors may contain only member access and constant collection indices.", + "field"); + } + internal static TTarget ConvertDocument(TSource document) where TSource : class where TTarget : class diff --git a/src/GameCult.Mesh/GameCult.Mesh.csproj b/src/GameCult.Mesh/GameCult.Mesh.csproj index c9904d28..898fa54f 100644 --- a/src/GameCult.Mesh/GameCult.Mesh.csproj +++ b/src/GameCult.Mesh/GameCult.Mesh.csproj @@ -4,6 +4,7 @@ GameCult.Mesh game;distributed;realtime;mesh;database;simulation;consensus README.md + true diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md index 25cd2865..c52ea4dd 100644 --- a/src/GameCult.Mesh/docs/reliability-migration-plan.md +++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md @@ -889,6 +889,42 @@ before fetch, corrupt network bytes never produce a lease, local failure is reported, fallback preserves identity, and authorization is evaluated separately from content integrity. +### Phase 5 exact-demand body map + +**Owner:** `CultMeshBodyDemandTracker` owns the live derivation from retained, +exact database subscriptions to the transport representations a producer must +publish. The provider owns body contents. A body adapter owns only access to +one representation. A renderer owns none of those decisions. + +**Inputs:** consumer runtime identity, exact logical body IDs, advertised +consumer transport capabilities, and server-observed peer locality. A +same-machine claim selects shared memory only when the transport peer is +actually loopback and the consumer explicitly supports shared memory. + +**Outputs:** a per-body plan requiring shared memory, network, both, or no +publication. Producers use that plan before allocating, copying, hashing, or +packaging a representation. + +**Derived state:** transport preference and publication work are notification +state derived from active subscriptions. Heavy database snapshots remain +bootstrap/recovery state. They are not a frame delivery mechanism. + +**Forbidden writers:** providers cannot infer demand from renderer identity; +renderers cannot choose a provider-specific plane; CDN publication cannot run +unconditionally beside a local mapped write; body descriptors cannot create +stream demand merely by existing. + +**Shared paths:** subscribe, resubscribe, replacement, explicit unsubscribe, +and server disposal all update the same typed demand projection. Stable peer +identity, not per-message wrapper identity, keys subscription lifetime. + +**Cut line:** local frame publication writes directly into a mapped write lease +and commits its descriptor. The compatibility byte-array publisher is copy-in +only. The current remote frame representation may still use the verified CDN +body path when remote demand exists; Phase 6 must replace that expiry-sensitive +traffic with a stream/datagram body plane rather than enlarging snapshot or CDN +timeouts. + ### Gate - Two local consumers map one verified CDN artifact without duplicate process diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 53da60d5..cfe349c5 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -48,7 +48,10 @@ namespace GameCult.Networking IEnumerable? recordKeys = null, IEnumerable? schemaIds = null, bool includeSnapshot = true, - CancellationToken cancellationToken = default) + CancellationToken cancellationToken = default, + string? consumerRuntimeId = null, + IEnumerable? bodyIds = null, + IEnumerable? supportedBodyTransports = null) { if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient)); if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing."); @@ -82,7 +85,13 @@ namespace GameCult.Networking SubscriptionId = subscriptionId, RecordKeys = recordKeys?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), - IncludeSnapshot = includeSnapshot + IncludeSnapshot = includeSnapshot, + ConsumerRuntimeId = consumerRuntimeId, + BodyIds = bodyIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), + SupportedBodyTransports = supportedBodyTransports? + .Where(value => !string.IsNullOrWhiteSpace(value)) + .Distinct(StringComparer.OrdinalIgnoreCase) + .ToArray() }); } catch diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index f61f7b93..0d8e4ff2 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -4,6 +4,7 @@ using System.Collections.Generic; using System.Linq; using System.Reflection; using System.Runtime.CompilerServices; +using System.Net; using System.Threading.Tasks; using GameCult.Caching; using R3; @@ -21,6 +22,8 @@ namespace GameCult.Networking private readonly Func _unsubscribe; private readonly ConcurrentDictionary _subscriptions = new ConcurrentDictionary(SubscriptionKeyComparer.Instance); + private readonly ConcurrentDictionary _requests = + new ConcurrentDictionary(SubscriptionKeyComparer.Instance); private bool _disposed; /// Attaches live database subscription handlers to a schema server. @@ -34,6 +37,9 @@ namespace GameCult.Networking _server.OnCultNet(_unsubscribe); } + /// Raised when exact document subscriptions add or remove typed hot-body demand. + public event Action? DemandChanged; + /// Detaches handlers and releases all active watches. public void Dispose() { @@ -41,8 +47,14 @@ namespace GameCult.Networking _disposed = true; _server.RemoveCultNetMessageListener(_subscribe); _server.RemoveCultNetMessageListener(_unsubscribe); - foreach (var subscription in _subscriptions.Values) subscription.Dispose(); + foreach (var entry in _subscriptions) + { + entry.Value.Dispose(); + if (_requests.TryRemove(entry.Key, out var request)) + PublishDemand(request, entry.Key, active: false); + } _subscriptions.Clear(); + _requests.Clear(); } private Task HandleSubscribeAsync(CultNetDatabaseSubscribeMessage request, ICultNetSchemaServerPeer peer) @@ -57,6 +69,8 @@ namespace GameCult.Networking } var key = new SubscriptionKey(peer, subscriptionId); + if (_requests.TryGetValue(key, out var previous)) + PublishDemand(previous, key, active: false); _subscriptions.AddOrUpdate( key, _ => Watch(request, subscriptionId, peer), @@ -65,6 +79,8 @@ namespace GameCult.Networking current.Dispose(); return Watch(request, subscriptionId, peer); }); + _requests[key] = request; + PublishDemand(request, key, active: true); if (request.IncludeSnapshot) { peer.SendCultNet(_database.Documents.CreateRawSnapshotResponse( @@ -93,12 +109,41 @@ namespace GameCult.Networking var subscriptionId = string.IsNullOrWhiteSpace(request.SubscriptionId) ? request.MessageId : request.SubscriptionId; - if (!string.IsNullOrWhiteSpace(subscriptionId) && - _subscriptions.TryRemove(new SubscriptionKey(peer, subscriptionId), out var subscription)) - subscription.Dispose(); + if (!string.IsNullOrWhiteSpace(subscriptionId)) + { + var key = new SubscriptionKey(peer, subscriptionId); + if (_subscriptions.TryRemove(key, out var subscription)) subscription.Dispose(); + if (_requests.TryRemove(key, out var subscribed)) + PublishDemand(subscribed, key, active: false); + } return Task.CompletedTask; } + private void PublishDemand( + CultNetDatabaseSubscribeMessage request, + SubscriptionKey key, + bool active) + { + if (request.BodyIds is not { Length: > 0 }) return; + DemandChanged?.Invoke(new CultNetDatabaseSubscriptionDemand( + string.IsNullOrWhiteSpace(request.ConsumerRuntimeId) ? key.Id : request.ConsumerRuntimeId!, + key.Id, + request.BodyIds.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(), + (request.SupportedBodyTransports ?? Array.Empty()) + .Where(value => !string.IsNullOrWhiteSpace(value)) + .Distinct(StringComparer.OrdinalIgnoreCase) + .ToArray(), + IsSameMachine(key.Peer), + active)); + } + + private static bool IsSameMachine(ICultNetSchemaServerPeer peer) + { + if (peer is RudpCultNetSchemaServerPeer rudp && rudp.RemoteEndPoint is IPEndPoint ip) + return IPAddress.IsLoopback(ip.Address); + return peer is CultNetServerPeer liteNet && IPAddress.IsLoopback(liteNet.Peer.Address); + } + private IDisposable Watch( CultNetDatabaseSubscribeMessage request, string subscriptionId, @@ -177,9 +222,55 @@ namespace GameCult.Networking { public static SubscriptionKeyComparer Instance { get; } = new SubscriptionKeyComparer(); public bool Equals(SubscriptionKey? x, SubscriptionKey? y) => - x != null && y != null && ReferenceEquals(x.Peer, y.Peer) && string.Equals(x.Id, y.Id, StringComparison.Ordinal); + x != null && y != null && SamePeer(x.Peer, y.Peer) && string.Equals(x.Id, y.Id, StringComparison.Ordinal); public int GetHashCode(SubscriptionKey value) => - (RuntimeHelpers.GetHashCode(value.Peer) * 397) ^ StringComparer.Ordinal.GetHashCode(value.Id); + (PeerHash(value.Peer) * 397) ^ StringComparer.Ordinal.GetHashCode(value.Id); + + private static bool SamePeer(ICultNetSchemaServerPeer left, ICultNetSchemaServerPeer right) + { + if (ReferenceEquals(left, right)) return true; + if (left is RudpCultNetSchemaServerPeer leftRudp && right is RudpCultNetSchemaServerPeer rightRudp) + return ReferenceEquals(leftRudp.TransportPeer, rightRudp.TransportPeer); + if (left is CultNetServerPeer leftLite && right is CultNetServerPeer rightLite) + return ReferenceEquals(leftLite.Peer, rightLite.Peer); + return false; + } + + private static int PeerHash(ICultNetSchemaServerPeer peer) + { + if (peer is RudpCultNetSchemaServerPeer rudp) + return RuntimeHelpers.GetHashCode(rudp.TransportPeer); + if (peer is CultNetServerPeer lite) + return RuntimeHelpers.GetHashCode(lite.Peer); + return RuntimeHelpers.GetHashCode(peer); + } } } + + /// One active or withdrawn consumer demand for logical hot bodies. + public sealed class CultNetDatabaseSubscriptionDemand + { + public CultNetDatabaseSubscriptionDemand( + string consumerRuntimeId, + string subscriptionId, + IReadOnlyList bodyIds, + IReadOnlyList supportedBodyTransports, + bool sameMachine, + bool active) + { + ConsumerRuntimeId = consumerRuntimeId; + SubscriptionId = subscriptionId; + BodyIds = bodyIds; + SupportedBodyTransports = supportedBodyTransports; + SameMachine = sameMachine; + Active = active; + } + + public string ConsumerRuntimeId { get; } + public string SubscriptionId { get; } + public IReadOnlyList BodyIds { get; } + public IReadOnlyList SupportedBodyTransports { get; } + public bool SameMachine { get; } + public bool Active { get; } + } } diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 77151dd1..2dc38ffa 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -787,6 +787,18 @@ namespace GameCult.Networking /// Gets or sets whether the server should send a matching snapshot before live changes. /// [Key("includeSnapshot")] public bool IncludeSnapshot { get; set; } = true; + /// + /// Gets or sets the stable consumer runtime identity used for subscription diagnostics and body-plane demand. + /// + [Key("consumerRuntimeId")] public string? ConsumerRuntimeId { get; set; } + /// + /// Gets or sets logical hot-body identities needed by this exact state subscription. + /// + [Key("bodyIds")] public string[]? BodyIds { get; set; } + /// + /// Gets or sets body transports the consumer can open, ordered by consumer preference. + /// + [Key("supportedBodyTransports")] public string[]? SupportedBodyTransports { get; set; } } /// diff --git a/tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs b/tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs index ee7a8cf4..f997bdad 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs @@ -49,8 +49,7 @@ public sealed class CultMeshBodyPublicationTests var first = Publication(new byte[] { 1, 2, 3, 4 }, now); var second = Publication(new byte[] { 5, 6, 7, 8 }, now); second.Sequence++; - second.PreferredLocal.Sequence++; - second.NetworkFallback.Sequence++; + foreach (var representation in second.Representations) representation.Sequence++; var cache = new CultCache(); await cache.UpsertAsync(first, new CultRecordHandle(first.RecordKey)); @@ -70,8 +69,7 @@ public sealed class CultMeshBodyPublicationTests var staleLatestKey = CultMeshBodyPublicationDocument.CreateLatestRecordKey(publication.BodyId); var next = Publication(new byte[4], DateTimeOffset.UtcNow); next.Sequence++; - next.PreferredLocal.Sequence++; - next.NetworkFallback.Sequence++; + foreach (var representation in next.Representations) representation.Sequence++; Action staleLatest = () => CultMeshBodyPublicationValidator.Validate( publication, publication.BodyId, publication.ProducerEpoch, publication.Sequence, staleLatestKey); @@ -95,13 +93,30 @@ public sealed class CultMeshBodyPublicationTests lease.ReadSingle(0).Should().Be(12.5f); } + [Test] + public void Resolver_AcceptsLocalOnlyPublicationWithoutInventingNetworkWork() + { + var now = DateTimeOffset.UtcNow; + var bytes = BitConverter.GetBytes(21.25f); + var publication = Publication(bytes, now); + publication.Representations = new[] { publication.Representations[0] }; + var fetched = false; + var resolver = Resolver(bytes, (_, _) => true, () => fetched = true); + + using var lease = resolver.ResolveReadOnly(publication, Request(publication, now)); + + lease.TransportKind.Should().Be(publication.Representations[0].TransportKind); + lease.ReadSingle(0).Should().Be(21.25f); + fetched.Should().BeFalse(); + } + [Test] public void Resolver_FallsBackToEquivalentNetworkRepresentation() { var now = DateTimeOffset.UtcNow; var bytes = new byte[] { 4, 8, 15, 16 }; var publication = Publication(bytes, now); - new CultMeshMappedBodyPublisher(_root).Revoke(publication.PreferredLocal); + new CultMeshMappedBodyPublisher(_root).Revoke(publication.Representations[0]); var result = Resolver(bytes, (_, _) => true).NegotiateReadOnly(publication, Request(publication, now)); using var lease = result.Lease; @@ -120,7 +135,7 @@ public sealed class CultMeshBodyPublicationTests { var now = DateTimeOffset.UtcNow; var publication = Publication(new byte[4], now); - var descriptor = publication.NetworkFallback; + var descriptor = publication.Representations[1]; if (field == "body") descriptor.BodyId += ":wrong"; if (field == "schema") descriptor.SchemaId += ".wrong"; if (field == "capacity") descriptor.Capacity++; @@ -184,8 +199,7 @@ public sealed class CultMeshBodyPublicationTests Sequence = local.Sequence, Synchronization = local.Synchronization, LivenessExpiresAtUnixMs = local.LeaseExpiresAtUnixMs, - PreferredLocal = local, - NetworkFallback = Network(local) + Representations = new[] { local, Network(local) } }; } diff --git a/tests/GameCult.Mesh.Tests/CultMeshBodyTransportTests.cs b/tests/GameCult.Mesh.Tests/CultMeshBodyTransportTests.cs index f3e9afb0..ca3e3045 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshBodyTransportTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshBodyTransportTests.cs @@ -3,6 +3,7 @@ using System.IO; using FluentAssertions; using NUnit.Framework; using System.IO.MemoryMappedFiles; +using GameCult.Networking; namespace GameCult.Mesh.Tests; @@ -263,6 +264,99 @@ public sealed class CultMeshBodyTransportTests lease.TransportKind.Should().Be(CultMeshBodyTransportKind.SharedMemory); } + [Test] + public void FramePublisher_DirectWriteLeaseCommitsMappedBytesWithoutCompatibilityCopy() + { + using var publisher = new CultMeshFrameBodyPublisher( + "aetheria:entities", + "gamecult.eve.entity_soa.body.v2", + layoutVersion: 2, + capacity: 128, + producerEpoch: 7, + slotByteLength: 64, + leaseDuration: TimeSpan.FromMinutes(1)); + publisher.TryAcquireWrite(out var write).Should().BeTrue(); + using (write) + { + BitConverter.TryWriteBytes(write.Span, 42.5f).Should().BeTrue(); + var descriptor = write.Commit(sizeof(float), DateTimeOffset.UtcNow); + using var read = new CultMeshSharedMemoryBodyAdapter().OpenReadOnly( + descriptor, + Request(descriptor, DateTimeOffset.UtcNow)); + + descriptor.Sequence.Should().Be(0); + descriptor.SemanticHash.Should().NotBeNullOrWhiteSpace(); + read.ReadSingle(0).Should().Be(42.5f); + } + } + + [Test] + public void FramePublisher_AbandonedWriteDoesNotAdvanceGenerationOrHoldSlot() + { + using var publisher = new CultMeshFrameBodyPublisher( + "aetheria:entities", + "gamecult.eve.entity_soa.body.v2", + layoutVersion: 2, + capacity: 128, + producerEpoch: 7, + slotByteLength: 64); + publisher.TryAcquireWrite(out var abandoned).Should().BeTrue(); + abandoned.Dispose(); + + publisher.TryAcquireWrite(out var replacement).Should().BeTrue(); + using (replacement) + { + replacement.Span[0] = 23; + var descriptor = replacement.Commit(1, DateTimeOffset.UtcNow); + descriptor.Sequence.Should().Be(0); + } + } + + [Test] + public void BodyDemand_SelectsSharedMemoryLocallyAndNetworkRemotelyWithoutDuplicateFallbackWork() + { + using var tracker = new CultMeshBodyDemandTracker(); + tracker.Observe(Demand("unity-local", "local", sameMachine: true)); + tracker.Observe(Demand("unity-remote", "remote", sameMachine: false)); + + var mixed = tracker.Plan("eve:entity-soa:aetheria.daemon:pilot"); + + mixed.HasConsumers.Should().BeTrue(); + mixed.RequiresSharedMemory.Should().BeTrue(); + mixed.RequiresNetwork.Should().BeTrue(); + mixed.Consumers.Should().ContainSingle(value => + value.ConsumerRuntimeId == "unity-local" && + value.Transport == CultMeshBodyTransportKind.SharedMemory); + mixed.Consumers.Should().ContainSingle(value => + value.ConsumerRuntimeId == "unity-remote" && + value.Transport == CultMeshBodyTransportKind.Network); + + tracker.Observe(Demand("unity-remote", "remote", sameMachine: false, active: false)); + var localOnly = tracker.Plan("eve:entity-soa:aetheria.daemon:pilot"); + localOnly.RequiresSharedMemory.Should().BeTrue(); + localOnly.RequiresNetwork.Should().BeFalse(); + } + + [Test] + public void BodyDemand_UnknownLocalityFailsTowardNetworkAndUnsupportedDemandCreatesNoRoute() + { + using var tracker = new CultMeshBodyDemandTracker(); + tracker.Observe(Demand("unknown", "network", sameMachine: false)); + tracker.Observe(new CultNetDatabaseSubscriptionDemand( + "unsupported", + "unsupported", + new[] { "eve:entity-soa:aetheria.daemon:pilot" }, + new[] { CultMeshBodyTransportKind.SharedMemory.ToString() }, + sameMachine: false, + active: true)); + + var plan = tracker.Plan("eve:entity-soa:aetheria.daemon:pilot"); + + plan.RequiresSharedMemory.Should().BeFalse(); + plan.RequiresNetwork.Should().BeTrue(); + plan.Consumers.Should().ContainSingle().Which.ConsumerRuntimeId.Should().Be("unknown"); + } + private static CultMeshBodyValidationRequest Request(CultMeshBodyDescriptor descriptor, DateTimeOffset now) => new() { BodyId = descriptor.BodyId, @@ -275,6 +369,22 @@ public sealed class CultMeshBodyTransportTests NowUtc = now }; + private static CultNetDatabaseSubscriptionDemand Demand( + string runtimeId, + string subscriptionId, + bool sameMachine, + bool active = true) => new( + runtimeId, + subscriptionId, + new[] { "eve:entity-soa:aetheria.daemon:pilot" }, + new[] + { + CultMeshBodyTransportKind.SharedMemory.ToString(), + CultMeshBodyTransportKind.Network.ToString() + }, + sameMachine, + active); + private static CultMeshBodyDescriptor NetworkRepresentation(CultMeshBodyDescriptor descriptor) => new() { BodyId = descriptor.BodyId, diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index e93da93c..05956d0d 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -723,6 +723,40 @@ public sealed class CultMeshStreamingTests observed.Revision.Should().Be(2); } + [Test] + public async Task DocumentHandle_Field_ProjectsTypedReactiveStateWithoutRepublishingSurfaceTrees() + { + var subject = new Subject(); + var route = new CultMeshRouteHint(CultMeshLocalityKind.SharedMemory, "co-located document slab"); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "initial", + Revision = 1 + }; + var handle = CultMesh.Document( + "mesh.note.current", + CultMesh.Verse("starbridge", "unity-pilot", route), + _ => Task.FromResult(current), + _ => subject, + routeHint: route); + + var text = handle.Field(value => value.Text); + + text.PointerId.Should().Be("mesh.note.current#Text"); + text.RouteHint.Kind.Should().Be(CultMeshLocalityKind.SharedMemory); + (await text.ResolveAsync()).Should().Be("initial"); + string observed = null!; + using var subscription = text.Watch().Subscribe(value => observed = value); + subject.OnNext(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "reactive", + Revision = 2 + }); + observed.Should().Be("reactive"); + } + [Test] public void DocumentHandle_RejectsAliasTypesWithDifferentSchemaIdentity() { diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 480f9d42..8ec4e8d4 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1425,6 +1425,68 @@ namespace GameCult.Networking.Tests Assert.That(update.Document!.RecordKey, Is.EqualTo(recordKey)); } + [Test] + public async Task DatabaseSubscriptionServer_ProjectsBodyDemandAndWithdrawalFromExactSubscription() + { + var database = new CultNetDatabase(new CultCache()); + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "database-body-demand-server", + Socket = BindUdpSocket() + }); + using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database); + var active = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + var withdrawn = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + subscriptions.DemandChanged += demand => + { + if (demand.Active) active.TrySetResult(demand); + else withdrawn.TrySetResult(demand); + }; + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + }) { IsBackground = true }; + serverThread.Start(); + + using var client = CultNetSchemaClients.CreateRudp("database-body-demand-client"); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected, TimeSpan.FromSeconds(2)); + client.SendCultNet(new CultNetDatabaseSubscribeMessage + { + MessageId = "subscribe-body", + SubscriptionId = "world-body", + RecordKeys = ["world:entities", "mesh:body:world:latest"], + ConsumerRuntimeId = "eve-unity", + BodyIds = ["world"], + SupportedBodyTransports = ["SharedMemory", "Network"], + IncludeSnapshot = false + }); + + var observed = await AwaitWithTimeout(active.Task, TimeSpan.FromSeconds(2)); + Assert.That(observed.ConsumerRuntimeId, Is.EqualTo("eve-unity")); + Assert.That(observed.SubscriptionId, Is.EqualTo("world-body")); + Assert.That(observed.BodyIds, Is.EqualTo(new[] { "world" })); + Assert.That(observed.SupportedBodyTransports, Is.EqualTo(new[] { "SharedMemory", "Network" })); + Assert.That(observed.SameMachine, Is.True); + + client.SendCultNet(new CultNetDatabaseUnsubscribeMessage + { + MessageId = "unsubscribe-body", + SubscriptionId = "world-body" + }); + var removed = await AwaitWithTimeout(withdrawn.Task, TimeSpan.FromSeconds(2)); + cancellation.Cancel(); + Assert.That(removed.ConsumerRuntimeId, Is.EqualTo("eve-unity")); + Assert.That(removed.Active, Is.False); + } + [Test] public async Task DatabaseSubscriptionClient_ReplicatesInitialAndLiveTypedDocumentsOverRudp() { diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index 65a8a822..6524d5ea 100644 --- a/unity/org.gamecult.cultlib/package.json +++ b/unity/org.gamecult.cultlib/package.json @@ -1,7 +1,7 @@ { "name": "org.gamecult.cultlib", "displayName": "CultLib", - "version": "1.0.19", + "version": "1.0.20", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",