From fb73c7beff463f7e58a92941b0e38e62659ee057 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 12 Jul 2026 20:26:36 +0100 Subject: [PATCH] Route snapshot sessions through CultMesh sessions --- src/GameCult.Mesh/CultMeshSessions.cs | 21 +++++++- src/GameCult.Mesh/CultMeshSnapshots.cs | 33 ++++++++++++- .../CultMeshSessionManagerTests.cs | 48 +++++++++++++++++++ 3 files changed, 100 insertions(+), 2 deletions(-) diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs index 51e32db5..00b9cf85 100644 --- a/src/GameCult.Mesh/CultMeshSessions.cs +++ b/src/GameCult.Mesh/CultMeshSessions.cs @@ -160,6 +160,7 @@ namespace GameCult.Mesh public sealed class CultMeshSession : IDisposable { private readonly Subject _states = new(); + private readonly ConcurrentBag _subscriptions = new(); private bool _disposed; internal CultMeshSession(CultMeshEndpointId endpointId, CultMeshProtocolId protocol, ICultNetSchemaClient channel, CultMeshSessionState state) { @@ -170,17 +171,35 @@ namespace GameCult.Mesh } public CultMeshEndpointId EndpointId { get; } public CultMeshProtocolId Protocol { get; } - public ICultNetSchemaClient Channel { get; } + internal ICultNetSchemaClient Channel { get; } public CultMeshSessionState State { get; private set; } public Observable WatchState() => _states; + public void SendCultNet(T message) where T : ICultNetSchemaMessage => Channel.SendCultNet(message); + public IDisposable OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (callback == null) throw new ArgumentNullException(nameof(callback)); + var subscription = new SessionSubscription(callback); + Channel.OnCultNet(subscription.Invoke); + _subscriptions.Add(subscription); + return subscription; + } internal void Transition(CultMeshSessionState state) { State = state; _states.OnNext(state); } public void Dispose() { if (_disposed) return; _disposed = true; + foreach (var subscription in _subscriptions) subscription.Dispose(); Channel.Dispose(); _states.Dispose(); } + + private sealed class SessionSubscription : IDisposable where T : ICultNetSchemaMessage + { + private Action? _callback; + public SessionSubscription(Action callback) { _callback = callback; } + public void Invoke(T message) => Volatile.Read(ref _callback)?.Invoke(message); + public void Dispose() => Interlocked.Exchange(ref _callback, null); + } } public sealed class CultMeshSessionManagerOptions diff --git a/src/GameCult.Mesh/CultMeshSnapshots.cs b/src/GameCult.Mesh/CultMeshSnapshots.cs index 3f2eb016..1e9996e1 100644 --- a/src/GameCult.Mesh/CultMeshSnapshots.cs +++ b/src/GameCult.Mesh/CultMeshSnapshots.cs @@ -69,6 +69,9 @@ namespace GameCult.Mesh private readonly string _endpoint; private readonly CultMeshSnapshotRequestOptions _defaults; private readonly ICultNetSchemaClient _client; + private readonly CultMeshSession? _sharedSession; + private readonly IDisposable? _responseSubscription; + private readonly IDisposable? _errorSubscription; private readonly CultNetDocumentRegistry _registry; private readonly SemaphoreSlim _requests = new(1, 1); private readonly object _completionLock = new(); @@ -93,6 +96,32 @@ namespace GameCult.Mesh _client.Connect(host, port); } + private CultMeshSnapshotSession( + CultMeshSession session, + CultMeshSnapshotRequestOptions options, + CultNetDocumentRegistry? registry) + { + _sharedSession = session ?? throw new ArgumentNullException(nameof(session)); + _endpoint = session.State.Path?.Endpoint ?? session.EndpointId.Value; + _defaults = CultMesh.CloneSnapshotRequestOptions(options); + _registry = registry ?? new CultNetDocumentRegistry(); + _client = session.Channel; + _responseSubscription = session.OnCultNet(OnResponse); + _errorSubscription = session.OnCultNet(OnError); + } + + public static async Task ConnectAsync( + CultMeshSessionManager sessions, + CultMeshEndpointId endpointId, + CultMeshSnapshotRequestOptions options, + CultNetDocumentRegistry? registry = null, + CancellationToken cancellationToken = default) + { + if (sessions == null) throw new ArgumentNullException(nameof(sessions)); + var session = await sessions.ConnectAsync(endpointId, CultMeshProtocols.Documents, cancellationToken).ConfigureAwait(false); + return new CultMeshSnapshotSession(session, options ?? new CultMeshSnapshotRequestOptions(), registry); + } + /// Fetches one raw snapshot while retaining the underlying endpoint connection. public async Task FetchSnapshotAsync( IReadOnlyList? schemaIds = null, @@ -165,7 +194,9 @@ namespace GameCult.Mesh _disposed = true; lock (_completionLock) _completion?.TrySetException(new ObjectDisposedException(nameof(CultMeshSnapshotSession))); - _client.Dispose(); + _responseSubscription?.Dispose(); + _errorSubscription?.Dispose(); + if (_sharedSession == null) _client.Dispose(); _requests.Dispose(); } diff --git a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs index c47944e2..21737d08 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs @@ -125,6 +125,27 @@ public sealed class CultMeshSessionManagerTests .Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.UnsupportedPath); } + [Test] + public async Task SnapshotSessionsBorrowOneManagedChannelAndDisposeRegistrationsIndependently() + { + var clock = new ManualSessionClock(); + using var discovery = Discovery(clock, () => new[] { "rudp://direct:3076" }); + var client = new RespondingSchemaClient(); + var connector = new FakeConnector((_, _) => Task.FromResult(client)); + using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock }); + var endpoint = CultMeshEndpointId.Parse("odin:aetheria"); + using var first = await CultMeshSnapshotSession.ConnectAsync(manager, endpoint, new CultMeshSnapshotRequestOptions()); + using var second = await CultMeshSnapshotSession.ConnectAsync(manager, endpoint, new CultMeshSnapshotRequestOptions()); + + (await first.FetchSnapshotAsync()).MessageId.Should().NotBeEmpty(); + first.Dispose(); + (await second.FetchSnapshotAsync()).MessageId.Should().NotBeEmpty(); + + connector.ConnectCount.Should().Be(1); + client.DisposeCount.Should().Be(0); + client.SentCount.Should().Be(2); + } + private static CultMeshDiscoveryService Discovery( ManualSessionClock clock, Func endpoints, @@ -188,6 +209,33 @@ public sealed class CultMeshSessionManagerTests public void Fail(Exception error) => _failure.TrySetResult(error); } + private sealed class RespondingSchemaClient : ICultNetSchemaClient + { + private readonly List> _handlers = new(); + private int _disposeCount; + public bool Connected => true; + public int DisposeCount => Volatile.Read(ref _disposeCount); + public int SentCount { get; private set; } + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + SentCount++; + var request = (CultNetSnapshotRequestMessage)(object)message; + var response = new CultNetSnapshotResponseRawMessage + { + MessageId = request.MessageId, + Documents = Array.Empty() + }; + foreach (var handler in _handlers.ToArray()) handler(response); + } + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (typeof(T) == typeof(CultNetSnapshotResponseRawMessage)) + _handlers.Add(message => callback((T)(object)message)); + } + public void Dispose() => Interlocked.Increment(ref _disposeCount); + } + private sealed class ManualSessionClock : ICultMeshClock { public DateTimeOffset UtcNow { get; private set; } = new DateTimeOffset(2026, 7, 12, 21, 0, 0, TimeSpan.Zero);