diff --git a/src/GameCult.Networking/CultNetSchemaServerGroup.cs b/src/GameCult.Networking/CultNetSchemaServerGroup.cs new file mode 100644 index 00000000..91a8484d --- /dev/null +++ b/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 +{ + /// + /// 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. + /// + public sealed class CultNetSchemaServerGroup : + ICultNetSchemaServer, + ICultNetSchemaServerPeerLifecycle, + IDisposable + { + private readonly IReadOnlyList _servers; + private readonly IReadOnlyList _lifecycles; + private bool _disposed; + + /// Creates one logical server from two or more transport servers. + 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().ToArray()); + foreach (var lifecycle in _lifecycles) lifecycle.PeerDisconnected += ForwardPeerDisconnected; + } + + /// + public event Action? PeerDisconnected; + + /// + public void OnCultNet(Func 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(callback); + throw; + } + } + + /// + public void RemoveCultNetMessageListener(Delegate callback) + where TMessage : ICultNetSchemaMessage + { + if (callback == null) throw new ArgumentNullException(nameof(callback)); + if (_disposed) return; + foreach (var server in _servers) server.RemoveCultNetMessageListener(callback); + } + + /// + /// Detaches lifecycle forwarding. The physical transports retain their + /// independent ownership and must be disposed by their creator. + /// + 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)); + } + } +} diff --git a/tests/GameCult.Networking.Tests/CultNetSchemaServerGroupTests.cs b/tests/GameCult.Networking.Tests/CultNetSchemaServerGroupTests.cs new file mode 100644 index 00000000..79e7e14f --- /dev/null +++ b/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(); + Func 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(handler); + + Assert.That(received, Is.EqualTo(new[] { "tcp", "websocket" })); + Assert.That(tcp.HasHandler(), Is.False); + Assert.That(webSocket.HasHandler(), 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(); + 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 _handlers = new(); + public bool Disposed { get; private set; } + public event Action? PeerDisconnected; + + public void OnCultNet(Func callback) + where TMessage : ICultNetSchemaMessage => + _handlers[typeof(TMessage)] = callback; + + public void RemoveCultNetMessageListener(Delegate callback) + where TMessage : ICultNetSchemaMessage + { + if (_handlers.TryGetValue(typeof(TMessage), out var current) && current == callback) + _handlers.Remove(typeof(TMessage)); + } + + public bool HasHandler() where TMessage : ICultNetSchemaMessage => + _handlers.ContainsKey(typeof(TMessage)); + + public Task DispatchAsync(TMessage message) where TMessage : ICultNetSchemaMessage => + ((Func)_handlers[typeof(TMessage)])(message, new FakePeer()); + + public void Disconnect(ICultNetSchemaServerPeer peer) => PeerDisconnected?.Invoke(peer); + } + + private sealed class FakePeer : ICultNetSchemaServerPeer + { + public void SendCultNet(TMessage message) where TMessage : ICultNetSchemaMessage + { + } + } + } +}