Preserve peer locality across schema transports

This commit is contained in:
Meta Vens 2026-07-19 18:50:22 +01:00
commit 6beca4a80a
6 changed files with 50 additions and 5 deletions

View file

@ -171,9 +171,9 @@ namespace GameCult.Networking
private static bool IsSameMachine(ICultNetSchemaServerPeer peer)
{
if (peer is RudpCultNetSchemaServerPeer rudp && rudp.RemoteEndPoint is IPEndPoint ip)
if (peer is ICultNetSchemaServerPeerLocation located && located.RemoteEndPoint is IPEndPoint ip)
return IPAddress.IsLoopback(ip.Address);
return peer is CultNetServerPeer liteNet && IPAddress.IsLoopback(liteNet.Peer.Address);
return false;
}
private IDisposable Watch(

View file

@ -70,7 +70,7 @@ namespace GameCult.Networking
/// <summary>
/// Peer context for a schema-v0 RUDP server.
/// </summary>
public sealed class RudpCultNetSchemaServerPeer : ICultNetSchemaServerPeer
public sealed class RudpCultNetSchemaServerPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation
{
private readonly RudpCultNetSchemaServer _server;

View file

@ -1,4 +1,5 @@
using System;
using System.Net;
using System.Threading.Tasks;
namespace GameCult.Networking
@ -33,6 +34,13 @@ namespace GameCult.Networking
where TMessage : ICultNetSchemaMessage;
}
/// <summary>Optional physical peer location used for locality-sensitive transport negotiation.</summary>
public interface ICultNetSchemaServerPeerLocation
{
/// <summary>Gets the remote physical endpoint when the transport exposes one.</summary>
EndPoint? RemoteEndPoint { get; }
}
/// <summary>
/// Optional transport-neutral peer lifetime surface for stateful service bridges.
/// </summary>

View file

@ -189,20 +189,23 @@ namespace GameCult.Networking
}
}
private sealed class TcpPeer : ICultNetSchemaServerPeer, IDisposable
private sealed class TcpPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation, IDisposable
{
private readonly TcpClient _client;
private readonly object _sendGate = new();
private readonly EndPoint? _remoteEndPoint;
public TcpPeer(TcpClient client)
{
_client = client;
_remoteEndPoint = client.Client.RemoteEndPoint;
Transport = new TcpFramedTransportConnection(
client.GetStream(),
CultNetTransportProfiles.CreateTcpFramed("csharp-tcp-schema-server"));
}
public bool Connected => _client.Connected;
public EndPoint? RemoteEndPoint => _remoteEndPoint;
public TcpFramedTransportConnection Transport { get; }
public void SendCultNet<TMessage>(TMessage message) where TMessage : ICultNetSchemaMessage

View file

@ -3,6 +3,7 @@ using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Text.RegularExpressions;
using System.Threading.Tasks;
using GameCult.Caching;
@ -820,7 +821,7 @@ namespace GameCult.Networking
/// <summary>
/// Transport-aware server peer context for built-in CultNet service bodies.
/// </summary>
public sealed class CultNetServerPeer : ICultNetSchemaServerPeer
public sealed class CultNetServerPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation
{
internal CultNetServerPeer(NetPeer peer, LiteNetLibTransportConnection transport)
{
@ -833,6 +834,9 @@ namespace GameCult.Networking
/// </summary>
public NetPeer Peer { get; }
/// <inheritdoc />
public EndPoint RemoteEndPoint => new IPEndPoint(Peer.Address, Peer.Port);
/// <summary>
/// Gets the channel-aware LiteNetLib transport adapter for this peer.
/// </summary>

View file

@ -1526,6 +1526,36 @@ namespace GameCult.Networking.Tests
Assert.That(removed.Active, Is.False);
}
[Test]
public async Task DatabaseSubscriptionServer_ClassifiesLoopbackTcpDemandAsSameMachine()
{
var database = new CultNetDatabase(new CultCache());
using var server = new TcpFramedCultNetSchemaServer(new TcpListener(IPAddress.Loopback, 0));
using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database);
var active = new TaskCompletionSource<CultNetDatabaseSubscriptionDemand>(
TaskCreationOptions.RunContinuationsAsynchronously);
subscriptions.DemandChanged += demand =>
{
if (demand.Active) active.TrySetResult(demand);
};
using var client = new TcpFramedCultNetSchemaClient();
client.Connect("127.0.0.1", server.LocalEndPoint.Port);
client.SendCultNet(new CultNetDatabaseSubscribeMessage
{
MessageId = "subscribe-tcp-body",
SubscriptionId = "world-body",
ConsumerRuntimeId = "eve-unity",
BodyIds = ["world"],
SupportedBodyTransports = ["SharedMemory", "Network"],
IncludeSnapshot = false
});
var observed = await AwaitWithTimeout(active.Task, TimeSpan.FromSeconds(2));
Assert.That(observed.SameMachine, Is.True);
Assert.That(observed.BodyIds, Is.EqualTo(new[] { "world" }));
}
[Test]
public async Task DatabaseSubscriptionServer_ProjectsExactReactiveStateDemandWithoutBodyDemand()
{