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

Add C# interop RUDP serve dial path

Browse source
This commit is contained in:
Meta Vens 2026-06-17 11:22:17 +01:00
commit cd3f3c28f6
4 changed files with 521 additions and 74 deletions
Download patch file Download diff file Expand all files Collapse all files

7
docs/cultnet-transport-parity.md
View file

@ -28,7 +28,7 @@ CultNet owns cross-runtime transport semantics:
| Runtime | Current data transport | Discovery | Transport authority today |
| --- | --- | --- | --- |
| C# `GameCult.Networking` | LiteNetLib UDP `NetManager` / `NetPeer`; sends legacy union messages and schema-v0 messages with `DeliveryMethod.ReliableOrdered`; single-peer RUDP socket transport exists in the library | LiteNetLib connection requests and app-level peer/catalog surfaces | Production LiteNetLib adapter profile plus UDP socket binding for the shared RUDP reliability owner |
| C# interop peer | TCP stream with 4-byte length-prefixed MessagePack frames | UDP multicast probe/announce | Test harness only |
| C# interop peer | Schema-v0 MessagePack over TCP-framed or shared RUDP transport | UDP multicast probe/announce with TCP and RUDP transport profiles | Test harness transport parity |
| TypeScript `cultnet-ts` | `CultNetPeer` over any Node `Duplex`, TCP-framed transport, or single-peer RUDP socket transport; interop uses TCP | UDP multicast probe/announce in the interop peer | First UDP socket binding for the shared RUDP reliability owner |
| Rust `cultnet-rs` | Interop example serves and dials schema-v0 MessagePack over TCP-framed or shared RUDP transport; single-peer RUDP socket transport exists in the library | UDP multicast probe/announce with TCP and RUDP transport profiles | UDP socket binding for the shared RUDP reliability owner |
| Python `cultcache-py` | TCP sockets with 4-byte length-prefixed MessagePack frames for local CultMesh/CultNet server and client; single-peer RUDP socket transport exists in the library | Endpoint lists and CultMesh peer/Verse catalogs | UDP socket binding for the shared RUDP reliability owner |
@ -145,7 +145,10 @@ Current progress:
- C# now has the shared `tcp_framed` transport profile helper plus a
`TcpFramedTransportConnection` with schema-channel `SendAsync`,
`ReceiveAsync`, and transfer stats. The C# interop peer advertises and uses
that shared port instead of owning raw TCP frame I/O directly.
that shared port instead of owning raw TCP frame I/O directly. The C# interop
peer now also advertises and serves the full schema-v0 flow over
`interop-rudp`, and its dial path can use the shared RUDP transport while
keeping TCP-framed as compatibility.
- C# now has a `litenetlib` transport profile helper, and the production
`Client`/`Server` expose profiles for the LiteNetLib lane. The profile names
both the modern reliable ordered `schema` channel and the legacy reliable

2
docs/runtime-parity-scope.md
View file

@ -30,7 +30,7 @@ CultMesh.
| Runtime | Role | Claimed parity | Not claimed |
| --- | --- | --- | --- |
| C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, schema-client transport port for service request/response clients, CultMesh node/session/catalog/stream/simulation/RUDP surfaces, authorized peer lookup | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption |
| C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, C# interop RUDP serve/dial path, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, schema-client transport port for service request/response clients, CultMesh node/session/catalog/stream/simulation/RUDP surfaces, authorized peer lookup | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption |
| Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup, shard catalogs/logs, schema-transport factory for raw client request/subscription flows, local TCP and RUDP CultMesh server listeners, Python interop RUDP serve/dial path, session/prediction/simulation fact helpers, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe |
| TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, advertised schema/shard catalog wire contracts plus request/sync helpers, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP interop serve/dial path, RUDP reconnect policy/controller, RUDP reconnect-loop helper, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup | Full C# game session/server stack |
| Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, schema and shard catalog helpers, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, Rust interop RUDP serve/dial path, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for schema registry, shard catalog, peer cards, authority leases, authorized peer lookup, RUDP endpoint parsing, and peer-card/authorized-peer RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server |

100
packages/cultnet-ts/test/interop/cultnet-interop.test.ts
View file

@ -1808,6 +1808,106 @@ test("CultNet Kotlin and TypeScript exchange schema-v0 MessagePack messages over
}
});
test("CultNet TypeScript and C# full interop peers exchange schema-v0 over RUDP", async (t) => {
logInteropPhase("ts-csharp-full-rudp", "build csharp peer");
await buildCSharpInteropPeer();
cleanInteropStores([
"ts-csharp-rudp-peer",
"ts-csharp-rudp-client-dial",
]);
const csharpTcpPort = await getFreePort();
const csharpRudpPort = await getFreePort();
const tsTcpPort = await getFreePort();
const tsRudpPort = await getFreePort();
const discoveryPort = await getFreePort();
const advertiseHost = findAdvertiseHost();
const servers: RunningServeProcess[] = [];
servers.push(await spawnServeProcess("csharp-rudp", {
command: dotnetCommand,
args: [
csharpDllPath,
"serve",
"--runtime-id", "csharp-rudp-peer",
"--runtime-kind", "dotnet",
"--display-name", "CSharp RUDP Peer",
"--agent-id", "csharp-rudp-agent",
"--bind-host", "127.0.0.1",
"--advertise-host", advertiseHost,
"--tcp-port", String(csharpTcpPort),
"--rudp-port", String(csharpRudpPort),
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--schema-path", interopSchemaPath,
],
cwd: cultLibRoot,
}));
await servers[servers.length - 1].ready;
servers.push(await spawnServeProcess("ts-csharp-rudp", {
command: process.execPath,
args: [
tsPeerScript,
"serve",
"--runtime-id", "ts-csharp-rudp-peer",
"--runtime-kind", "node",
"--display-name", "TypeScript CSharp RUDP Peer",
"--agent-id", "ts-csharp-rudp-agent",
"--bind-host", "127.0.0.1",
"--advertise-host", advertiseHost,
"--tcp-port", String(tsTcpPort),
"--rudp-port", String(tsRudpPort),
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--schema-path", interopSchemaPath,
],
cwd: cultNetTsRoot,
}));
await servers[servers.length - 1].ready;
t.after(async () => {
await Promise.all(servers.map(stopProcess));
});
logInteropPhase("ts-csharp-full-rudp", "typescript dials csharp over rudp");
const tsRudpDial = await runJsonCommand("ts-csharp-rudp-dial", process.execPath, [
tsPeerScript,
"dial",
"--runtime-id", "ts-csharp-rudp-client",
"--runtime-kind", "node",
"--display-name", "TS CSharp RUDP Dialer",
"--agent-id", "ts-csharp-rudp-client-agent",
"--target-host", "127.0.0.1",
"--target-rudp-port", String(csharpRudpPort),
"--schema-path", interopSchemaPath,
], cultNetTsRoot);
assert.equal(tsRudpDial.transport, "rudp");
assert.equal(tsRudpDial.remoteHello.runtimeId, "csharp-rudp-peer");
assert.equal(tsRudpDial.hasInteropSchema, true);
assert.equal(tsRudpDial.retrievedNote.authorRuntimeId, "csharp-rudp-peer");
assert.ok(tsRudpDial.mutatedNote.tags.includes("decorated:ts-csharp-rudp-client"));
assert.equal(tsRudpDial.fireReceipt.accepted, true);
logInteropPhase("ts-csharp-full-rudp", "csharp dials typescript over rudp");
const csharpRudpDial = await runJsonCommand("csharp-rudp-dial", dotnetCommand, [
csharpDllPath,
"dial",
"--runtime-id", "csharp-rudp-client",
"--runtime-kind", "dotnet",
"--display-name", "CSharp RUDP Dialer",
"--agent-id", "csharp-rudp-client-agent",
"--target-host", "127.0.0.1",
"--target-rudp-port", String(tsRudpPort),
"--schema-path", interopSchemaPath,
], cultLibRoot);
assert.equal(csharpRudpDial.transport, "rudp");
assert.equal(csharpRudpDial.remoteHello.runtimeId, "ts-csharp-rudp-peer");
assert.equal(csharpRudpDial.hasInteropSchema, true);
assert.equal(csharpRudpDial.retrievedNote.authorRuntimeId, "ts-csharp-rudp-peer");
assert.ok(csharpRudpDial.mutatedNote.tags.includes("decorated:csharp-rudp-client"));
assert.equal(csharpRudpDial.fireReceipt.accepted, true);
});
test("CultNet RUDP reliable schema delivery survives one dropped client data packet across runtimes", async () => {
for (const peer of rudpServerInteropPeers()) {
await peer.build();

486
tests/GameCult.Networking.InteropPeer/Program.cs
View file

@ -80,6 +80,9 @@ static async Task ServeAsync(ServeConfig config)
using var udpSocket = CreateDiscoverySocket(config.DiscoveryPort, config.DiscoveryGroup);
using var tcpListener = new TcpListener(IPAddress.Parse(config.BindHost), config.TcpPort);
using var rudpSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
rudpSocket.Bind(new IPEndPoint(IPAddress.Parse(config.BindHost), config.RudpPort));
rudpSocket.ReceiveTimeout = 20;
tcpListener.Start();
var cancellationSource = new CancellationTokenSource();
@ -92,6 +95,7 @@ static async Task ServeAsync(ServeConfig config)
_ = RunDiscoveryServerAsync(udpSocket, config, cancellationSource.Token);
_ = RunTcpServerAsync(tcpListener, config, cache, documentRegistry, customSchemaRegistry, cancellationSource.Token);
_ = RunRudpServerAsync(rudpSocket, config, cache, documentRegistry, customSchemaRegistry, cancellationSource.Token);
WriteJsonLine(new
{
@ -100,6 +104,7 @@ static async Task ServeAsync(ServeConfig config)
runtimeId = config.RuntimeId,
runtimeKind = config.RuntimeKind,
tcpPort = config.TcpPort,
rudpPort = config.RudpPort,
discoveryPort = config.DiscoveryPort,
discoveryGroup = config.DiscoveryGroup.ToString()
});
@ -192,12 +197,8 @@ static async Task DialAsync(DialConfig config)
payloadDeserializer: DeserializeInteropNotePayload));
RegisterCapabilityBindings(documentRegistry);
using var client = new TcpClient();
await client.ConnectAsync(config.TargetHost, config.TargetPort);
using var stream = client.GetStream();
var transport = new TcpFramedTransportConnection(
stream,
CreateTcpFramedTransportProfile(config.RuntimeId, config.TargetHost, config.TargetPort));
using var transport = await OpenDialTransportAsync(config);
var transportName = config.TargetRudpPort.HasValue ? "rudp" : "tcp_framed";
await SendMessageAsync(transport, new CultNetHelloMessage
{
@ -209,6 +210,7 @@ static async Task DialAsync(DialConfig config)
SupportedDocumentTypes = [InteropPeerShared.InteropDocumentType],
SupportedMutationContracts = [InteropPeerShared.InteractionContract()],
SupportedMessageVersions = [InteropPeerShared.InteropSchemaVersion],
TransportProfiles = CreateDialTransportProfiles(config),
SupportsSchemaCatalog = true
});
var remoteHello = await ExpectMessageAsync<CultNetHelloMessage>(transport, CultNetSchemaVersions.Hello);
@ -245,9 +247,10 @@ static async Task DialAsync(DialConfig config)
WriteJsonLine(new
{
mode = "dial",
transport = transportName,
runtimeId = config.RuntimeId,
targetHost = config.TargetHost,
targetPort = config.TargetPort,
targetPort = config.TargetRudpPort ?? config.TargetPort,
remoteHello = new
{
schemaVersion = remoteHello.SchemaVersion,
@ -276,7 +279,7 @@ static async Task DialAsync(DialConfig config)
}
static async Task<object> MutateRemoteNoteAsync(
TcpFramedTransportConnection transport,
ISchemaMessageTransport transport,
CultCache cache,
CultNetDocumentRegistry documentRegistry,
string runtimeId,
@ -311,7 +314,7 @@ static async Task<object> MutateRemoteNoteAsync(
}
static async Task<object> FireRemoteWeaponAsync(
TcpFramedTransportConnection transport,
ISchemaMessageTransport transport,
CultCache cache,
CultNetDocumentRegistry documentRegistry,
string runtimeId,
@ -379,7 +382,7 @@ static async Task RunDiscoveryServerAsync(Socket socket, ServeConfig config, Can
TcpPort = config.TcpPort,
WireContract = CultNetWireContracts.SchemaV0,
SupportedDocumentTypes = [InteropPeerShared.InteropDocumentType],
TransportProfiles = [CreateTcpFramedTransportProfile(config.RuntimeId, config.AdvertiseHost, config.TcpPort)],
TransportProfiles = CreateInteropTransportProfiles(config.RuntimeId, config.AdvertiseHost, config.TcpPort, config.RudpPort),
SupportsSchemaCatalog = true
};
var announceBytes = MessagePackSerializer.Serialize(announce, CultNetSchemaMessageSerialization.Options);
@ -426,61 +429,22 @@ static async Task RunTcpServerAsync(
try
{
using var stream = ownedClient.GetStream();
var transport = new TcpFramedTransportConnection(
using var transport = new TcpSchemaMessageTransport(new TcpFramedTransportConnection(
stream,
CreateTcpFramedTransportProfile(config.RuntimeId, config.AdvertiseHost, config.TcpPort));
CreateTcpFramedTransportProfile(config.RuntimeId, config.AdvertiseHost, config.TcpPort)));
while (!cancellationToken.IsCancellationRequested)
{
ICultNetSchemaMessage message;
try
{
message = await ReadMessageAsync(transport, cancellationToken);
message = await transport.ReadMessageAsync(cancellationToken);
}
catch (EndOfStreamException)
{
break;
}
switch (message)
{
case CultNetHelloMessage:
await SendMessageAsync(transport, new CultNetHelloMessage
{
RuntimeId = config.RuntimeId,
RuntimeKind = config.RuntimeKind,
AgentId = config.AgentId,
Role = "peer",
DisplayName = config.DisplayName,
SupportedDocumentTypes = [InteropPeerShared.InteropDocumentType],
SupportedMutationContracts = [InteropPeerShared.InteractionContract()],
SupportedMessageVersions = [InteropPeerShared.InteropSchemaVersion],
TransportProfiles = [CreateTcpFramedTransportProfile(config.RuntimeId, config.AdvertiseHost, config.TcpPort)],
SupportsSchemaCatalog = true
}, cancellationToken);
break;
case CultNetSchemaCatalogRequestMessage catalogRequest:
await SendMessageAsync(transport, CreateCatalogResponse(customSchemaRegistry, catalogRequest), cancellationToken);
break;
case CultNetSnapshotRequestMessage snapshotRequest:
await SendMessageAsync(
transport,
documentRegistry.CreateRawSnapshotResponse(
cache,
snapshotRequest.MessageId,
snapshotRequest,
new CultNetDocumentMessageOptions
{
SourceRuntimeId = config.RuntimeId,
SourceAgentId = config.AgentId,
SourceRole = "peer",
Tags = ["interop", config.RuntimeId]
}),
cancellationToken);
break;
case CultNetDocumentPutRawMessage rawPut:
await HandleRawPutAsync(transport, config, cache, documentRegistry, rawPut, cancellationToken);
break;
}
await HandleServerMessageAsync(transport, message, config, cache, documentRegistry, customSchemaRegistry, cancellationToken);
}
}
catch (Exception error)
@ -491,8 +455,163 @@ static async Task RunTcpServerAsync(
}
}
static Task RunRudpServerAsync(
Socket socket,
ServeConfig config,
CultCache cache,
CultNetDocumentRegistry documentRegistry,
CultNetSchemaRegistry customSchemaRegistry,
CancellationToken cancellationToken)
{
return Task.Run(async () =>
{
var peers = new Dictionary<string, RudpPeerSession>(StringComparer.Ordinal);
var buffer = new byte[65535];
while (!cancellationToken.IsCancellationRequested)
{
try
{
EndPoint remote = new IPEndPoint(IPAddress.Any, 0);
var received = socket.ReceiveFrom(buffer, ref remote);
var remoteKey = remote.ToString() ?? string.Empty;
var packetBytes = new byte[received];
Array.Copy(buffer, packetBytes, received);
CultNetRudpPacket packet;
try
{
packet = CultNetRudpPacketCodec.Decode(packetBytes);
}
catch
{
continue;
}
if (packet.ConnectionId != InteropPeerShared.RudpInteropConnectionId)
{
continue;
}
if (!peers.TryGetValue(remoteKey, out var peer))
{
peer = new RudpPeerSession(remote, new CultNetRudpSession(new CultNetRudpSessionOptions
{
ConnectionId = InteropPeerShared.RudpInteropConnectionId,
InitialSequence = 100,
ResendDelayMs = InteropPeerShared.RudpInteropResendDelayMs
}));
peers[remoteKey] = peer;
}
if (packet.PacketType == CultNetRudpPacketType.Connect)
{
InteropRudpRuntime.SendPacket(socket, peer.RemoteEndPoint, peer.Session.AcceptConnect(
packet,
InteropRudpRuntime.NowMs(),
Encoding.UTF8.GetBytes("cultnet-interop-rudp")));
continue;
}
var result = peer.Session.Receive(packet, InteropRudpRuntime.NowMs());
if (result.Reply != null)
{
InteropRudpRuntime.SendPacket(socket, peer.RemoteEndPoint, result.Reply);
}
if (packet.PacketType == CultNetRudpPacketType.Data)
{
InteropRudpRuntime.SendPacket(socket, peer.RemoteEndPoint, peer.Session.CreateAck());
}
foreach (var frame in result.Delivered)
{
if (!string.Equals(frame.ChannelId, "schema", StringComparison.Ordinal))
{
continue;
}
var message = CultNetSchemaMessageSerialization.Deserialize(frame.Payload);
var transport = new RudpSessionSchemaSender(socket, peer.RemoteEndPoint, peer.Session);
await HandleServerMessageAsync(transport, message, config, cache, documentRegistry, customSchemaRegistry, cancellationToken);
}
}
catch (SocketException error) when (
error.SocketErrorCode == SocketError.TimedOut ||
error.SocketErrorCode == SocketError.WouldBlock ||
error.SocketErrorCode == SocketError.ConnectionReset)
{
var now = InteropRudpRuntime.NowMs();
foreach (var peer in peers.Values)
{
foreach (var packet in peer.Session.DueResends(now))
{
InteropRudpRuntime.SendPacket(socket, peer.RemoteEndPoint, packet);
}
}
}
catch (ObjectDisposedException)
{
break;
}
catch (Exception error)
{
WriteLog("rudpServeError", new { runtimeId = config.RuntimeId, error = error.Message });
}
}
}, cancellationToken);
}
static async Task HandleServerMessageAsync(
ISchemaMessageSender transport,
ICultNetSchemaMessage message,
ServeConfig config,
CultCache cache,
CultNetDocumentRegistry documentRegistry,
CultNetSchemaRegistry customSchemaRegistry,
CancellationToken cancellationToken)
{
switch (message)
{
case CultNetHelloMessage:
await SendMessageAsync(transport, new CultNetHelloMessage
{
RuntimeId = config.RuntimeId,
RuntimeKind = config.RuntimeKind,
AgentId = config.AgentId,
Role = "peer",
DisplayName = config.DisplayName,
SupportedDocumentTypes = [InteropPeerShared.InteropDocumentType],
SupportedMutationContracts = [InteropPeerShared.InteractionContract()],
SupportedMessageVersions = [InteropPeerShared.InteropSchemaVersion],
TransportProfiles = CreateInteropTransportProfiles(config.RuntimeId, config.AdvertiseHost, config.TcpPort, config.RudpPort),
SupportsSchemaCatalog = true
}, cancellationToken);
break;
case CultNetSchemaCatalogRequestMessage catalogRequest:
await SendMessageAsync(transport, CreateCatalogResponse(customSchemaRegistry, catalogRequest), cancellationToken);
break;
case CultNetSnapshotRequestMessage snapshotRequest:
await SendMessageAsync(
transport,
documentRegistry.CreateRawSnapshotResponse(
cache,
snapshotRequest.MessageId,
snapshotRequest,
new CultNetDocumentMessageOptions
{
SourceRuntimeId = config.RuntimeId,
SourceAgentId = config.AgentId,
SourceRole = "peer",
Tags = ["interop", config.RuntimeId]
}),
cancellationToken);
break;
case CultNetDocumentPutRawMessage rawPut:
await HandleRawPutAsync(transport, config, cache, documentRegistry, rawPut, cancellationToken);
break;
}
}
static async Task HandleRawPutAsync(
TcpFramedTransportConnection transport,
ISchemaMessageSender transport,
ServeConfig config,
CultCache cache,
CultNetDocumentRegistry documentRegistry,
@ -566,6 +685,18 @@ static CultNetSchemaCatalogResponseMessage CreateCatalogResponse(
.Concat(custom.Where(entry => builtIn.All(candidate => !string.Equals(candidate.SchemaId, entry.SchemaId, StringComparison.Ordinal))))
.ToArray();
foreach (var schema in schemas)
{
schema.SchemaVersion = string.IsNullOrWhiteSpace(schema.SchemaVersion)
? schema.SchemaId
: schema.SchemaVersion;
schema.DocumentType = string.IsNullOrWhiteSpace(schema.DocumentType)
? schema.Kind
: schema.DocumentType;
schema.Title ??= string.Empty;
schema.SchemaJson ??= string.Empty;
}
return new CultNetSchemaCatalogResponseMessage
{
MessageId = request.MessageId,
@ -575,12 +706,107 @@ static CultNetSchemaCatalogResponseMessage CreateCatalogResponse(
static CultNetTransportProfile CreateTcpFramedTransportProfile(string runtimeId, string host, int port)
{
return CultNetTransportProfiles.CreateTcpFramed(runtimeId, new TcpFramedTransportProfileOptions
return NormalizeInteropTransportProfile(CultNetTransportProfiles.CreateTcpFramed(runtimeId, new TcpFramedTransportProfileOptions
{
TransportId = "interop-tcp",
Host = host,
Port = port
});
}));
}
static CultNetTransportProfile CreateRudpTransportProfile(string runtimeId, string host, int port)
{
return NormalizeInteropTransportProfile(CultNetTransportProfiles.CreateRudp(runtimeId, new RudpTransportProfileOptions
{
TransportId = "interop-rudp",
Host = host,
Port = port,
MaxFragmentBytes = InteropPeerShared.RudpInteropMaxFragmentBytes
}));
}
static CultNetTransportProfile NormalizeInteropTransportProfile(CultNetTransportProfile profile)
{
foreach (var transport in profile.Transports)
{
transport.Host ??= string.Empty;
transport.Path ??= "/";
transport.DiscoveryGroup ??= "none";
transport.ReconnectPolicy ??= CultNetReconnectPolicies.CreateDefault();
transport.ReconnectPolicy.MaxAttempts ??= 1;
foreach (var channel in transport.Channels)
{
channel.MaxPayloadBytes ??= 65535;
channel.MaxFragmentBytes ??= 65535;
channel.MaxPendingReliablePackets ??= 1024;
}
}
return profile;
}
static CultNetTransportProfile[] CreateInteropTransportProfiles(string runtimeId, string host, int tcpPort, int rudpPort)
{
return
[
CreateTcpFramedTransportProfile(runtimeId, host, tcpPort),
CreateRudpTransportProfile(runtimeId, host, rudpPort)
];
}
static CultNetTransportProfile[] CreateDialTransportProfiles(DialConfig config)
{
var profiles = new List<CultNetTransportProfile>();
if (config.TargetPort.HasValue)
{
profiles.Add(CreateTcpFramedTransportProfile(config.RuntimeId, config.TargetHost, config.TargetPort.Value));
}
if (config.TargetRudpPort.HasValue)
{
profiles.Add(CreateRudpTransportProfile(config.RuntimeId, config.TargetHost, config.TargetRudpPort.Value));
}
return profiles.ToArray();
}
static async Task<ISchemaMessageTransport> OpenDialTransportAsync(DialConfig config)
{
if (config.TargetRudpPort.HasValue)
{
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
socket.ReceiveTimeout = 20;
var transport = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = $"{config.RuntimeId}-interop-rudp-dial",
Socket = socket,
Mode = CultNetRudpSocketMode.Client,
RemoteEndPoint = new IPEndPoint(IPAddress.Parse(config.TargetHost), config.TargetRudpPort.Value),
ConnectionId = InteropPeerShared.RudpInteropConnectionId,
InitialSequence = 1,
ResendDelayMs = InteropPeerShared.RudpInteropResendDelayMs,
TransportId = "interop-rudp",
MaxFragmentBytes = InteropPeerShared.RudpInteropMaxFragmentBytes
});
transport.Connect(Encoding.UTF8.GetBytes("cultnet-interop-rudp"));
if (!transport.AwaitConnected(TimeSpan.FromSeconds(5)))
{
transport.Dispose();
throw new TimeoutException("Timed out waiting for RUDP connect.");
}
return new RudpSocketSchemaTransport(transport);
}
if (!config.TargetPort.HasValue)
{
throw new InvalidOperationException("dial mode requires --target-port or --target-rudp-port.");
}
var client = new TcpClient();
await client.ConnectAsync(config.TargetHost, config.TargetPort.Value);
return new TcpSchemaMessageTransport(
new TcpFramedTransportConnection(
client.GetStream(),
CreateTcpFramedTransportProfile(config.RuntimeId, config.TargetHost, config.TargetPort.Value)),
client);
}
static void RudpServeOnce(Dictionary<string, string> options)
@ -842,21 +1068,21 @@ static void RequireRudpFrameBytes(CultNetTransportFrame frame, string expectedCh
}
static async Task SendMessageAsync<TMessage>(
TcpFramedTransportConnection transport,
ISchemaMessageSender transport,
TMessage message,
CancellationToken cancellationToken = default)
where TMessage : ICultNetSchemaMessage
{
var payload = CultNetSchemaMessageSerialization.Serialize(message);
await transport.SendAsync("schema", payload, cancellationToken);
await transport.SendMessageAsync(payload, cancellationToken);
}
static async Task<TMessage> ExpectMessageAsync<TMessage>(
TcpFramedTransportConnection transport,
ISchemaMessageTransport transport,
string expectedSchemaVersion)
where TMessage : class, ICultNetSchemaMessage
{
var message = await ReadMessageAsync(transport, CancellationToken.None);
var message = await transport.ReadMessageAsync(CancellationToken.None);
if (!string.Equals(message.SchemaVersion, expectedSchemaVersion, StringComparison.Ordinal))
{
throw new InvalidOperationException($"Expected {expectedSchemaVersion} but received {message.SchemaVersion}.");
@ -866,14 +1092,6 @@ static async Task<TMessage> ExpectMessageAsync<TMessage>(
?? throw new InvalidOperationException($"Expected {typeof(TMessage).Name} but received {message.GetType().Name}.");
}
static async Task<ICultNetSchemaMessage> ReadMessageAsync(
TcpFramedTransportConnection transport,
CancellationToken cancellationToken)
{
var frame = await transport.ReceiveAsync(cancellationToken);
return CultNetSchemaMessageSerialization.Deserialize(frame.Payload);
}
static Socket CreateDiscoverySocket(int port, IPAddress multicastGroup)
{
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp)
@ -1041,6 +1259,7 @@ static int ParseOptionalIntArg(Dictionary<string, string> options, string name,
static ServeConfig ParseServeConfig(Dictionary<string, string> options)
{
var tcpPort = ParseIntArg(options, "tcp-port");
return new ServeConfig(
RequireArg(options, "runtime-id"),
RequireArg(options, "runtime-kind"),
@ -1048,7 +1267,8 @@ static ServeConfig ParseServeConfig(Dictionary<string, string> options)
RequireArg(options, "agent-id"),
options.TryGetValue("bind-host", out var bindHost) ? bindHost : "127.0.0.1",
RequireArg(options, "advertise-host"),
ParseIntArg(options, "tcp-port"),
tcpPort,
ParseOptionalIntArg(options, "rudp-port", tcpPort),
ParseIntArg(options, "discovery-port"),
IPAddress.Parse(RequireArg(options, "discovery-group")),
RequireArg(options, "schema-path"));
@ -1062,7 +1282,12 @@ static DialConfig ParseDialConfig(Dictionary<string, string> options)
RequireArg(options, "display-name"),
RequireArg(options, "agent-id"),
RequireArg(options, "target-host"),
ParseIntArg(options, "target-port"),
options.TryGetValue("target-port", out var targetPort)
? int.Parse(targetPort, CultureInfo.InvariantCulture)
: null,
options.TryGetValue("target-rudp-port", out var targetRudpPort)
? int.Parse(targetRudpPort, CultureInfo.InvariantCulture)
: null,
RequireArg(options, "schema-path"));
}
@ -1085,6 +1310,120 @@ static void WriteLog(string @event, object payload)
}, InteropPeerShared.JsonOptions));
}
interface ISchemaMessageSender
{
Task SendMessageAsync(byte[] payload, CancellationToken cancellationToken = default);
}
interface ISchemaMessageTransport : ISchemaMessageSender, IDisposable
{
Task<ICultNetSchemaMessage> ReadMessageAsync(CancellationToken cancellationToken = default);
}
sealed class TcpSchemaMessageTransport : ISchemaMessageTransport
{
private readonly TcpFramedTransportConnection _transport;
private readonly TcpClient? _client;
public TcpSchemaMessageTransport(TcpFramedTransportConnection transport, TcpClient? client = null)
{
_transport = transport;
_client = client;
}
public Task SendMessageAsync(byte[] payload, CancellationToken cancellationToken = default)
{
return _transport.SendAsync("schema", payload, cancellationToken);
}
public async Task<ICultNetSchemaMessage> ReadMessageAsync(CancellationToken cancellationToken = default)
{
var frame = await _transport.ReceiveAsync(cancellationToken);
return CultNetSchemaMessageSerialization.Deserialize(frame.Payload);
}
public void Dispose()
{
_client?.Dispose();
}
}
sealed class RudpSocketSchemaTransport : ISchemaMessageTransport
{
private readonly CultNetRudpSocketTransportConnection _transport;
public RudpSocketSchemaTransport(CultNetRudpSocketTransportConnection transport)
{
_transport = transport;
}
public Task SendMessageAsync(byte[] payload, CancellationToken cancellationToken = default)
{
_transport.Send("schema", payload);
return Task.CompletedTask;
}
public Task<ICultNetSchemaMessage> ReadMessageAsync(CancellationToken cancellationToken = default)
{
var message = _transport.ReceiveSchemaMessage(TimeSpan.FromSeconds(15))
?? throw new TimeoutException("Timed out waiting for RUDP schema message.");
return Task.FromResult(message);
}
public void Dispose()
{
_transport.Dispose();
}
}
sealed class RudpSessionSchemaSender : ISchemaMessageSender
{
private readonly Socket _socket;
private readonly EndPoint _remoteEndPoint;
private readonly CultNetRudpSession _session;
public RudpSessionSchemaSender(Socket socket, EndPoint remoteEndPoint, CultNetRudpSession session)
{
_socket = socket;
_remoteEndPoint = remoteEndPoint;
_session = session;
}
public Task SendMessageAsync(byte[] payload, CancellationToken cancellationToken = default)
{
foreach (var packet in _session.SendMany(
"schema",
payload,
new CultNetRudpSendOptions
{
Reliable = true,
Ordered = true,
Sequenced = false,
NowMs = InteropRudpRuntime.NowMs()
},
InteropPeerShared.RudpInteropMaxFragmentBytes))
{
InteropRudpRuntime.SendPacket(_socket, _remoteEndPoint, packet);
}
return Task.CompletedTask;
}
}
sealed record RudpPeerSession(EndPoint RemoteEndPoint, CultNetRudpSession Session);
static class InteropRudpRuntime
{
public static void SendPacket(Socket socket, EndPoint remoteEndPoint, CultNetRudpPacket packet)
{
socket.SendTo(CultNetRudpPacketCodec.Encode(packet), remoteEndPoint);
}
public static long NowMs()
{
return DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
}
}
sealed record ServeConfig(
string RuntimeId,
string RuntimeKind,
@ -1093,6 +1432,7 @@ sealed record ServeConfig(
string BindHost,
string AdvertiseHost,
int TcpPort,
int RudpPort,
int DiscoveryPort,
IPAddress DiscoveryGroup,
string SchemaPath);
@ -1103,7 +1443,8 @@ sealed record DialConfig(
string DisplayName,
string AgentId,
string TargetHost,
int TargetPort,
int? TargetPort,
int? TargetRudpPort,
string SchemaPath);
static class InteropPeerShared
@ -1124,6 +1465,9 @@ static class InteropPeerShared
public const string FireReceiptSchemaVersion = "cultnet.interop_fire_weapon_receipt.v0";
public const string DiscoveryProbeSchemaVersion = "cultnet.discovery_probe.v0";
public const string DiscoveryAnnounceSchemaVersion = "cultnet.discovery_announce.v0";
public const uint RudpInteropConnectionId = 0x43554C54;
public const int RudpInteropMaxFragmentBytes = 1024;
public const long RudpInteropResendDelayMs = 25;
public static readonly JsonSerializerOptions JsonOptions = new()
{