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

Lease schema clients from CultMesh sessions

Browse source
This commit is contained in:
Meta Vens 2026-07-12 20:33:39 +01:00
commit b4e06bc762
2 changed files with 74 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -174,6 +174,7 @@ namespace GameCult.Mesh
internal ICultNetSchemaClient Channel { get; }
public CultMeshSessionState State { get; private set; }
public Observable<CultMeshSessionState> WatchState() => _states;
public ICultNetSchemaClient OpenSchemaClient() => new SessionSchemaClient(this);
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage => Channel.SendCultNet(message);
public IDisposable OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
{
@ -200,6 +201,36 @@ namespace GameCult.Mesh
public void Invoke(T message) => Volatile.Read(ref _callback)?.Invoke(message);
public void Dispose() => Interlocked.Exchange(ref _callback, null);
}
private sealed class SessionSchemaClient : ICultNetSchemaClient
{
private readonly CultMeshSession _session;
private readonly ConcurrentBag<IDisposable> _registrations = new();
private bool _disposed;
public SessionSchemaClient(CultMeshSession session) { _session = session; }
public bool Connected => !_disposed && _session.State.Status == CultMeshSessionStatus.Online;
public void Connect(string host, int port)
{
if (_disposed) throw new ObjectDisposedException(nameof(SessionSchemaClient));
if (!Connected) throw new InvalidOperationException("The managed CultMesh session is not online.");
}
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage
{
if (_disposed) throw new ObjectDisposedException(nameof(SessionSchemaClient));
_session.SendCultNet(message);
}
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
{
if (_disposed) throw new ObjectDisposedException(nameof(SessionSchemaClient));
_registrations.Add(_session.OnCultNet(callback));
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
foreach (var registration in _registrations) registration.Dispose();
}
}
}
public sealed class CultMeshSessionManagerOptions

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

@ -166,6 +166,32 @@ public sealed class CultMeshSessionManagerTests
client.SentCount.Should().Be(2);
}
[Test]
public async Task SchemaClientLeasesDisposeHandlersWithoutClosingSession()
{
var clock = new ManualSessionClock();
using var discovery = Discovery(clock, () => new[] { "rudp://direct:3076" });
var client = new LeaseSchemaClient();
var connector = new FakeConnector((_, _) => Task.FromResult<ICultNetSchemaClient>(client));
using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock });
var session = await manager.ConnectAsync(CultMeshEndpointId.Parse("odin:aetheria"), CultMeshProtocols.Subscriptions);
using var first = session.OpenSchemaClient();
using var second = session.OpenSchemaClient();
var firstCount = 0;
var secondCount = 0;
first.OnCultNet<CultNetErrorMessage>(_ => firstCount++);
second.OnCultNet<CultNetErrorMessage>(_ => secondCount++);
client.Emit(new CultNetErrorMessage { Error = "first" });
first.Dispose();
client.Emit(new CultNetErrorMessage { Error = "second" });
firstCount.Should().Be(1);
secondCount.Should().Be(2);
client.DisposeCount.Should().Be(0);
session.State.Status.Should().Be(CultMeshSessionStatus.Online);
}
private static CultMeshDiscoveryService Discovery(
ManualSessionClock clock,
Func<string[]> endpoints,
@ -284,6 +310,23 @@ public sealed class CultMeshSessionManagerTests
public void Dispose() { }
}
private sealed class LeaseSchemaClient : ICultNetSchemaClient
{
private readonly List<Action<CultNetErrorMessage>> _handlers = new();
private int _disposeCount;
public bool Connected => true;
public int DisposeCount => Volatile.Read(ref _disposeCount);
public void Connect(string host, int port) { }
public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage { }
public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
{
if (typeof(T) == typeof(CultNetErrorMessage))
_handlers.Add(message => callback((T)(object)message));
}
public void Emit(CultNetErrorMessage message) { foreach (var handler in _handlers.ToArray()) handler(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);