From b4e06bc7627d4188af1c3d59f436dc24e7e2c2a6 Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Sun, 12 Jul 2026 20:33:39 +0100
Subject: [PATCH] Lease schema clients from CultMesh sessions

---
 src/GameCult.Mesh/CultMeshSessions.cs         | 31 +++++++++++++
 .../CultMeshSessionManagerTests.cs            | 43 +++++++++++++++++++
 2 files changed, 74 insertions(+)

diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs
index 00b9cf85..07662ab1 100644
--- a/src/GameCult.Mesh/CultMeshSessions.cs
+++ b/src/GameCult.Mesh/CultMeshSessions.cs
@@ -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
diff --git a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs
index 84bdd6d9..fa595bce 100644
--- a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs
+++ b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs
@@ -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);