diff --git a/src/GameCult.Mesh/CultMeshPeerExchange.cs b/src/GameCult.Mesh/CultMeshPeerExchange.cs index 67e294fd..109df07d 100644 --- a/src/GameCult.Mesh/CultMeshPeerExchange.cs +++ b/src/GameCult.Mesh/CultMeshPeerExchange.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using GameCult.Networking; using R3; @@ -343,6 +344,12 @@ namespace GameCult.Mesh /// Defaults to endpoint selection: rudp:// uses RUDP and cultnet:// uses LiteNetLib. /// public Func? CreateClient { get; set; } + + /// Gets or sets the shared CultMesh session owner. + public CultMeshSessionManager? Sessions { get; set; } + + /// Gets or sets the clock used by managed response deadlines. + public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance; } /// @@ -401,6 +408,36 @@ namespace GameCult.Mesh return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false); } + /// Fetches peer cards through the shared session owner. + public async Task FetchAsync( + CultMeshEndpointId endpointId, + CultMeshPeerExchangeRequestMessage request, + CancellationToken cancellationToken = default) + { + if (endpointId == null) throw new ArgumentNullException(nameof(endpointId)); + if (request == null) throw new ArgumentNullException(nameof(request)); + if (string.IsNullOrWhiteSpace(request.VerseId)) throw new ArgumentException("Peer exchange requires a verseId.", nameof(request)); + var sessions = _options.Sessions ?? throw new InvalidOperationException("Managed peer exchange requires a CultMeshSessionManager."); + var session = await sessions.ConnectAsync(endpointId, CultMeshProtocols.PeerExchange, cancellationToken).ConfigureAwait(false); + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var responseSubscription = session.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) completion.TrySetResult(response); + }); + using var errorSubscription = session.OnCultNet(error => + completion.TrySetException(new InvalidOperationException(error.Error))); + session.SendCultNet(new CultMeshPeerExchangeRequestMessage + { + MessageId = messageId, + VerseId = request.VerseId, + Roles = request.Roles, + KnownPeerIds = request.KnownPeerIds, + Limit = request.Limit + }); + return await WaitForManagedResponseAsync(completion.Task, endpointId, cancellationToken).ConfigureAwait(false); + } + /// /// Fetches peer cards from endpoints and upserts every response into a local peer catalog. /// @@ -465,5 +502,20 @@ namespace GameCult.Mesh return await responseTask.ConfigureAwait(false); } + + private async Task WaitForManagedResponseAsync( + Task responseTask, + CultMeshEndpointId endpointId, + CancellationToken cancellationToken) + { + var timeout = _options.Clock.DelayAsync(_options.ResponseTimeout, cancellationToken); + var completed = await Task.WhenAny(responseTask, timeout).ConfigureAwait(false); + if (completed != responseTask) + { + cancellationToken.ThrowIfCancellationRequested(); + throw new TimeoutException($"Timed out waiting for peer exchange response from {endpointId}."); + } + return await responseTask.ConfigureAwait(false); + } } } diff --git a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs index 21737d08..84bdd6d9 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs @@ -146,6 +146,26 @@ public sealed class CultMeshSessionManagerTests client.SentCount.Should().Be(2); } + [Test] + public async Task PeerExchangeBorrowsManagedProtocolSessionAcrossRequests() + { + var clock = new ManualSessionClock(); + using var discovery = Discovery(clock, () => new[] { "rudp://direct:3076" }); + var client = new PeerExchangeSchemaClient(); + var connector = new FakeConnector((_, _) => Task.FromResult(client)); + using var manager = new CultMeshSessionManager(discovery, new[] { connector }, new CultMeshSessionManagerOptions { Clock = clock }); + var exchange = new CultMeshPeerExchangeClient(new CultMeshPeerExchangeClientOptions { Sessions = manager, Clock = clock }); + var endpoint = CultMeshEndpointId.Parse("odin:aetheria"); + + var first = await exchange.FetchAsync(endpoint, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" }); + var second = await exchange.FetchAsync(endpoint, new CultMeshPeerExchangeRequestMessage { VerseId = "aetheria" }); + + first.Peers.Single().PeerId.Should().Be("peer-a"); + second.Peers.Single().PeerId.Should().Be("peer-a"); + connector.ConnectCount.Should().Be(1); + client.SentCount.Should().Be(2); + } + private static CultMeshDiscoveryService Discovery( ManualSessionClock clock, Func endpoints, @@ -236,6 +256,34 @@ public sealed class CultMeshSessionManagerTests public void Dispose() => Interlocked.Increment(ref _disposeCount); } + private sealed class PeerExchangeSchemaClient : ICultNetSchemaClient + { + private readonly List> _handlers = new(); + public bool Connected => true; + public int SentCount { get; private set; } + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + SentCount++; + var request = (CultMeshPeerExchangeRequestMessage)(object)message; + var response = new CultMeshPeerExchangeResponseMessage + { + MessageId = request.MessageId, + Peers = new[] + { + new CultMeshPeerCard("peer-a", request.VerseId, new[] { "rudp://peer-a:3076" }).ToMessage() + } + }; + foreach (var handler in _handlers.ToArray()) handler(response); + } + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (typeof(T) == typeof(CultMeshPeerExchangeResponseMessage)) + _handlers.Add(message => callback((T)(object)message)); + } + public void Dispose() { } + } + private sealed class ManualSessionClock : ICultMeshClock { public DateTimeOffset UtcNow { get; private set; } = new DateTimeOffset(2026, 7, 12, 21, 0, 0, TimeSpan.Zero);