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

Route snapshot sessions through CultMesh sessions

Browse source
This commit is contained in:
Meta Vens 2026-07-12 20:26:36 +01:00
commit fb73c7beff
3 changed files with 100 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

21
src/GameCult.Mesh/CultMeshSessions.cs
View file

@ -160,6 +160,7 @@ namespace GameCult.Mesh
public sealed class CultMeshSession : IDisposable
{
private readonly Subject<CultMeshSessionState> _states = new();
private readonly ConcurrentBag<IDisposable> _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<CultMeshSessionState> WatchState() => _states;
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage => Channel.SendCultNet(message);
public IDisposable OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
var subscription = new SessionSubscription<T>(callback);
Channel.OnCultNet<T>(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<T> : IDisposable where T : ICultNetSchemaMessage
{
private Action<T>? _callback;
public SessionSubscription(Action<T> 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

33
src/GameCult.Mesh/CultMeshSnapshots.cs
View file

@ -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<CultNetSnapshotResponseRawMessage>(OnResponse);
_errorSubscription = session.OnCultNet<CultNetErrorMessage>(OnError);
}
public static async Task<CultMeshSnapshotSession> 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);
}
/// <summary>Fetches one raw snapshot while retaining the underlying endpoint connection.</summary>
public async Task<CultNetSnapshotResponseRawMessage> FetchSnapshotAsync(
IReadOnlyList<string>? 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();
}

48
tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs
View file

@ -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<ICultNetSchemaClient>(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<string[]> 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<Action<CultNetSnapshotResponseRawMessage>> _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>(T message) where T : ICultNetSchemaMessage
{
SentCount++;
var request = (CultNetSnapshotRequestMessage)(object)message;
var response = new CultNetSnapshotResponseRawMessage
{
MessageId = request.MessageId,
Documents = Array.Empty<CultNetRawDocumentRecord>()
};
foreach (var handler in _handlers.ToArray()) handler(response);
}
public void OnCultNet<T>(Action<T> 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);