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

Route peer exchange through CultMesh sessions

Browse source
This commit is contained in:
Meta Vens 2026-07-12 20:30:55 +01:00
commit 1277c6bd2e
2 changed files with 100 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

52
src/GameCult.Mesh/CultMeshPeerExchange.cs
View file

@ -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.
/// </summary>
public Func<ICultNetSchemaClient>? CreateClient { get; set; }
/// <summary>Gets or sets the shared CultMesh session owner.</summary>
public CultMeshSessionManager? Sessions { get; set; }
/// <summary>Gets or sets the clock used by managed response deadlines.</summary>
public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance;
}
/// <summary>
@ -401,6 +408,36 @@ namespace GameCult.Mesh
return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false);
}
/// <summary>Fetches peer cards through the shared session owner.</summary>
public async Task<CultMeshPeerExchangeResponseMessage> 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<CultMeshPeerExchangeResponseMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
using var responseSubscription = session.OnCultNet<CultMeshPeerExchangeResponseMessage>(response =>
{
if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) completion.TrySetResult(response);
});
using var errorSubscription = session.OnCultNet<CultNetErrorMessage>(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);
}
/// <summary>
/// Fetches peer cards from endpoints and upserts every response into a local peer catalog.
/// </summary>
@ -465,5 +502,20 @@ namespace GameCult.Mesh
return await responseTask.ConfigureAwait(false);
}
private async Task<CultMeshPeerExchangeResponseMessage> WaitForManagedResponseAsync(
Task<CultMeshPeerExchangeResponseMessage> 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);
}
}
}

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

@ -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<ICultNetSchemaClient>(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<string[]> endpoints,
@ -236,6 +256,34 @@ public sealed class CultMeshSessionManagerTests
public void Dispose() => Interlocked.Increment(ref _disposeCount);
}
private sealed class PeerExchangeSchemaClient : ICultNetSchemaClient
{
private readonly List<Action<CultMeshPeerExchangeResponseMessage>> _handlers = new();
public bool Connected => true;
public int SentCount { get; private set; }
public void Connect(string host, int port) { }
public void SendCultNet<T>(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<T>(Action<T> 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);