Share CultNet handlers across transports

src/GameCult.Networking/CultNetSchemaServerGroup.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace GameCult.Networking
+{
+    /// <summary>
+    /// Presents one logical CultNet schema server over several physical transports.
+    /// It owns handler fan-out and peer-lifecycle aggregation, but never owns or
+    /// disposes the transport servers themselves.
+    /// </summary>
+    public sealed class CultNetSchemaServerGroup :
+        ICultNetSchemaServer,
+        ICultNetSchemaServerPeerLifecycle,
+        IDisposable
+    {
+        private readonly IReadOnlyList<ICultNetSchemaServer> _servers;
+        private readonly IReadOnlyList<ICultNetSchemaServerPeerLifecycle> _lifecycles;
+        private bool _disposed;
+
+        /// <summary>Creates one logical server from two or more transport servers.</summary>
+        public CultNetSchemaServerGroup(params ICultNetSchemaServer[] servers)
+        {
+            if (servers == null) throw new ArgumentNullException(nameof(servers));
+            if (servers.Length < 2)
+                throw new ArgumentException("A CultNet schema server group requires at least two transports.", nameof(servers));
+            if (servers.Any(server => server == null))
+                throw new ArgumentException("A CultNet schema server group cannot contain a null transport.", nameof(servers));
+            for (var left = 0; left < servers.Length; left++)
+            for (var right = left + 1; right < servers.Length; right++)
+            {
+                if (ReferenceEquals(servers[left], servers[right]))
+                    throw new ArgumentException("A CultNet schema server group cannot contain the same transport twice.", nameof(servers));
+            }
+
+            _servers = Array.AsReadOnly(servers.ToArray());
+            _lifecycles = Array.AsReadOnly(servers.OfType<ICultNetSchemaServerPeerLifecycle>().ToArray());
+            foreach (var lifecycle in _lifecycles) lifecycle.PeerDisconnected += ForwardPeerDisconnected;
+        }
+
+        /// <inheritdoc />
+        public event Action<ICultNetSchemaServerPeer>? PeerDisconnected;
+
+        /// <inheritdoc />
+        public void OnCultNet<TMessage>(Func<TMessage, ICultNetSchemaServerPeer, Task> callback)
+            where TMessage : ICultNetSchemaMessage
+        {
+            if (callback == null) throw new ArgumentNullException(nameof(callback));
+            ThrowIfDisposed();
+            var registered = 0;
+            try
+            {
+                foreach (var server in _servers)
+                {
+                    server.OnCultNet(callback);
+                    registered++;
+                }
+            }
+            catch
+            {
+                for (var index = 0; index < registered; index++)
+                    _servers[index].RemoveCultNetMessageListener<TMessage>(callback);
+                throw;
+            }
+        }
+
+        /// <inheritdoc />
+        public void RemoveCultNetMessageListener<TMessage>(Delegate callback)
+            where TMessage : ICultNetSchemaMessage
+        {
+            if (callback == null) throw new ArgumentNullException(nameof(callback));
+            if (_disposed) return;
+            foreach (var server in _servers) server.RemoveCultNetMessageListener<TMessage>(callback);
+        }
+
+        /// <summary>
+        /// Detaches lifecycle forwarding. The physical transports retain their
+        /// independent ownership and must be disposed by their creator.
+        /// </summary>
+        public void Dispose()
+        {
+            if (_disposed) return;
+            _disposed = true;
+            foreach (var lifecycle in _lifecycles) lifecycle.PeerDisconnected -= ForwardPeerDisconnected;
+            PeerDisconnected = null;
+        }
+
+        private void ForwardPeerDisconnected(ICultNetSchemaServerPeer peer) =>
+            PeerDisconnected?.Invoke(peer);
+
+        private void ThrowIfDisposed()
+        {
+            if (_disposed) throw new ObjectDisposedException(nameof(CultNetSchemaServerGroup));
+        }
+    }
+}

tests/GameCult.Networking.Tests/CultNetSchemaServerGroupTests.cs

@@ -0,0 +1,101 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using NUnit.Framework;
+
+namespace GameCult.Networking.Tests
+{
+    [TestFixture]
+    public sealed class CultNetSchemaServerGroupTests
+    {
+        [Test]
+        public async Task RegistersAndRemovesTheSameLogicalHandlerAcrossTransports()
+        {
+            var tcp = new FakeServer();
+            var webSocket = new FakeServer();
+            using var group = new CultNetSchemaServerGroup(tcp, webSocket);
+            var received = new List<string>();
+            Func<CultNetHelloMessage, ICultNetSchemaServerPeer, Task> handler = (message, _) =>
+            {
+                received.Add(message.RuntimeId);
+                return Task.CompletedTask;
+            };
+
+            group.OnCultNet(handler);
+            await tcp.DispatchAsync(new CultNetHelloMessage { RuntimeId = "tcp" });
+            await webSocket.DispatchAsync(new CultNetHelloMessage { RuntimeId = "websocket" });
+            group.RemoveCultNetMessageListener<CultNetHelloMessage>(handler);
+
+            Assert.That(received, Is.EqualTo(new[] { "tcp", "websocket" }));
+            Assert.That(tcp.HasHandler<CultNetHelloMessage>(), Is.False);
+            Assert.That(webSocket.HasHandler<CultNetHelloMessage>(), Is.False);
+        }
+
+        [Test]
+        public void AggregatesPeerLifetimeWithoutTakingTransportOwnership()
+        {
+            var tcp = new FakeServer();
+            var webSocket = new FakeServer();
+            var group = new CultNetSchemaServerGroup(tcp, webSocket);
+            var peer = new FakePeer();
+            var disconnected = new List<ICultNetSchemaServerPeer>();
+            group.PeerDisconnected += disconnected.Add;
+
+            tcp.Disconnect(peer);
+            webSocket.Disconnect(peer);
+            group.Dispose();
+            tcp.Disconnect(peer);
+
+            Assert.That(disconnected, Is.EqualTo(new[] { peer, peer }));
+            Assert.That(tcp.Disposed, Is.False);
+            Assert.That(webSocket.Disposed, Is.False);
+        }
+
+        [Test]
+        public void RejectsAGroupThatDoesNotActuallyAddASecondTransport()
+        {
+            var server = new FakeServer();
+            Assert.Multiple(() =>
+            {
+                Assert.That(() => new CultNetSchemaServerGroup(server), Throws.ArgumentException);
+                Assert.That(() => new CultNetSchemaServerGroup(server, server), Throws.ArgumentException);
+            });
+        }
+
+        private sealed class FakeServer :
+            ICultNetSchemaServer,
+            ICultNetSchemaServerPeerLifecycle
+        {
+            private readonly Dictionary<Type, Delegate> _handlers = new();
+            public bool Disposed { get; private set; }
+            public event Action<ICultNetSchemaServerPeer>? PeerDisconnected;
+
+            public void OnCultNet<TMessage>(Func<TMessage, ICultNetSchemaServerPeer, Task> callback)
+                where TMessage : ICultNetSchemaMessage =>
+                _handlers[typeof(TMessage)] = callback;
+
+            public void RemoveCultNetMessageListener<TMessage>(Delegate callback)
+                where TMessage : ICultNetSchemaMessage
+            {
+                if (_handlers.TryGetValue(typeof(TMessage), out var current) && current == callback)
+                    _handlers.Remove(typeof(TMessage));
+            }
+
+            public bool HasHandler<TMessage>() where TMessage : ICultNetSchemaMessage =>
+                _handlers.ContainsKey(typeof(TMessage));
+
+            public Task DispatchAsync<TMessage>(TMessage message) where TMessage : ICultNetSchemaMessage =>
+                ((Func<TMessage, ICultNetSchemaServerPeer, Task>)_handlers[typeof(TMessage)])(message, new FakePeer());
+
+            public void Disconnect(ICultNetSchemaServerPeer peer) => PeerDisconnected?.Invoke(peer);
+        }
+
+        private sealed class FakePeer : ICultNetSchemaServerPeer
+        {
+            public void SendCultNet<TMessage>(TMessage message) where TMessage : ICultNetSchemaMessage
+            {
+            }
+        }
+    }
+}