using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;
using GameCult.Logging;
namespace GameCult.Networking
{
///
/// Result of a bounded non-blocking RUDP schema-server drain.
///
public readonly struct RudpCultNetSchemaPollResult
{
public RudpCultNetSchemaPollResult(int transportItemsConsumed, int messagesDispatched)
{
TransportItemsConsumed = transportItemsConsumed;
MessagesDispatched = messagesDispatched;
}
/// Gets transport work consumed, including ACK and connection control packets.
public int TransportItemsConsumed { get; }
/// Gets application schema messages dispatched.
public int MessagesDispatched { get; }
}
///
/// Options for hosting schema-v0 messages over the native CultNet RUDP listener.
///
public sealed class RudpCultNetSchemaServerOptions
{
///
/// Gets or sets the runtime id advertised by the RUDP listener.
///
public string RuntimeId { get; set; } = "csharp-rudp-schema-server";
///
/// Gets or sets the UDP socket used by the listener. When omitted, a loopback socket is created.
///
public Socket? Socket { get; set; }
///
/// Gets or sets the accepted RUDP connection id.
///
public uint ConnectionId { get; set; } = 0x43554c54;
///
/// Gets or sets the first local packet sequence for each accepted peer. Unset, each peer's
/// session draws its own at random.
///
public uint? InitialSequence { get; set; }
///
/// Gets or sets the resend delay in milliseconds.
///
public long ResendDelayMs { get; set; } = 25;
///
/// Gets or sets the advertised transport id.
///
public string TransportId { get; set; } = "schema-rudp";
///
/// Gets or sets the maximum fragment size for schema-channel payloads.
///
public int? MaxFragmentBytes { get; set; } = 1024;
///
/// Gets or sets the maximum pending reliable packet count.
///
public int? MaxPendingReliablePackets { get; set; }
///
/// Gets or sets the payload sent with accept packets.
///
public byte[]? AcceptPayload { get; set; }
}
///
/// Peer context for a schema-v0 RUDP server.
///
public sealed class RudpCultNetSchemaServerPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation
{
private readonly RudpCultNetSchemaServer _server;
internal RudpCultNetSchemaServerPeer(
RudpCultNetSchemaServer server,
CultNetRudpSocketServerPeer transportPeer)
{
_server = server;
TransportPeer = transportPeer;
}
///
/// Gets the underlying RUDP transport peer.
///
public CultNetRudpSocketServerPeer TransportPeer { get; }
///
/// Gets the UDP endpoint for this peer.
///
public EndPoint RemoteEndPoint => TransportPeer.RemoteEndPoint;
///
/// Sends a schema-v0 response to this peer.
///
public void SendCultNet(TMessage message)
where TMessage : ICultNetSchemaMessage
{
_server.SendCultNet(TransportPeer, message);
}
}
///
/// Multi-peer schema-v0 host over the native CultNet RUDP transport.
///
public sealed class RudpCultNetSchemaServer : ICultNetSchemaServer, ICultNetSchemaServerPeerLifecycle, IDisposable
{
private readonly Socket _socket;
private readonly CultNetRudpSocketTransportServer _transport;
private readonly ConcurrentDictionary _handlers = new();
private readonly ConcurrentDictionary _schemaPeerAdapters = new();
private bool _disposed;
private ILogger _logger = new NullLogger();
///
/// Gets or sets the logger used by the server. R-AM: the dispatch backstop in DispatchAsync
/// logs every untyped fault through this instead of swallowing it silently.
///
public ILogger Logger
{
get => _logger;
set => _logger = value ?? new NullLogger();
}
///
/// Creates a schema-v0 RUDP server.
///
public RudpCultNetSchemaServer(RudpCultNetSchemaServerOptions? options = null)
{
options ??= new RudpCultNetSchemaServerOptions();
var socket = options.Socket;
if (socket == null)
{
socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
}
_socket = socket;
_transport = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
{
RuntimeId = options.RuntimeId,
Socket = socket,
ConnectionId = options.ConnectionId,
InitialSequence = options.InitialSequence,
ResendDelayMs = options.ResendDelayMs,
TransportId = options.TransportId,
MaxFragmentBytes = options.MaxFragmentBytes,
MaxPendingReliablePackets = options.MaxPendingReliablePackets,
AcceptPayload = options.AcceptPayload
});
_transport.PeerDisconnected += OnTransportPeerDisconnected;
}
///
/// Gets the underlying RUDP listener profile.
///
public CultNetTransportProfile Profile => _transport.Profile;
///
/// Gets the local UDP endpoint.
///
public IPEndPoint LocalEndPoint => (IPEndPoint)_socket.LocalEndPoint!;
///
/// Gets a snapshot of the current transfer counters.
///
public CultNetTransportStats Stats => _transport.Stats;
// The listener this server drives, for tests that inject send failures through its seams.
internal CultNetRudpSocketTransportServer Transport => _transport;
///
/// Gets the currently tracked RUDP peers.
///
public IReadOnlyCollection Peers => _transport.Peers;
///
public event Action? PeerDisconnected;
///
/// Registers a schema-v0 handler.
///
public void OnCultNet(Func callback)
where TMessage : ICultNetSchemaMessage
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
_handlers.AddOrUpdate(
typeof(TMessage),
_ => callback,
(type, current) =>
{
var combined = Delegate.Combine(current, callback) as Func;
return combined ?? throw new InvalidOperationException($"Failed to combine delegates for {type.Name}.");
});
}
///
/// Registers a schema-v0 handler.
///
public void OnCultNet(Action callback)
where TMessage : ICultNetSchemaMessage
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
OnCultNet((message, peer) =>
{
callback(message, peer);
return Task.CompletedTask;
});
}
///
public void OnCultNet(Func callback)
where TMessage : ICultNetSchemaMessage
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
Func adapter = (message, peer) => callback(message, peer);
_schemaPeerAdapters[callback] = adapter;
OnCultNet(adapter);
}
///
public void RemoveCultNetMessageListener(Delegate callback)
where TMessage : ICultNetSchemaMessage
{
if (_schemaPeerAdapters.TryRemove(callback, out var adapter))
{
callback = adapter;
}
if (!_handlers.TryGetValue(typeof(TMessage), out var current))
{
return;
}
var next = Delegate.Remove(current, callback);
if (next == null)
{
_handlers.TryRemove(typeof(TMessage), out _);
}
else
{
_handlers[typeof(TMessage)] = next;
}
}
///
/// Sends a schema-v0 message to a transport peer.
///
public void SendCultNet(CultNetRudpSocketServerPeer peer, TMessage message)
where TMessage : ICultNetSchemaMessage
{
_transport.SendSchemaMessage(peer, message);
}
///
/// Polls one packet/frame and dispatches at most one schema-v0 message.
///
public async Task PollOnceAsync()
{
if (!_transport.TryReceiveOnce(out var delivered) || delivered == null)
{
return false;
}
return await DispatchAsync(delivered).ConfigureAwait(false);
}
///
/// Drains up to immediately available transport work items.
/// Control packets count as progress so callers only idle when the transport is actually empty.
///
public async Task PollAvailableAsync(int maxTransportItems)
{
if (maxTransportItems <= 0) throw new ArgumentOutOfRangeException(nameof(maxTransportItems));
var consumed = 0;
var dispatched = 0;
while (consumed < maxTransportItems && _transport.TryReceiveOnce(out var delivered))
{
consumed++;
if (delivered != null && await DispatchAsync(delivered).ConfigureAwait(false))
dispatched++;
}
return new RudpCultNetSchemaPollResult(consumed, dispatched);
}
///
/// The goodbye's reason when a handler fails on a peer's message: the server could not handle
/// it, which says nothing about whether the peer's packet was well formed.
///
internal static readonly byte[] HandlerFailedReason = Encoding.UTF8.GetBytes("server could not handle a message");
private static string SessionOutcome(bool ended) =>
ended ? "that peer's session ended." : "that peer's session had already ended.";
private async Task DispatchAsync(CultNetRudpSocketServerFrame delivered)
{
if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal))
return false;
// A peer's message that cannot be decoded (malformed bytes, an unknown schemaVersion) or
// handled ends that peer's session, never the poll: the drain goes on serving every other
// peer. The transport already acknowledged the frame, so the session cannot be kept (a
// retransmit would never be handled). The peer is told, and the drop is counted and logged.
// The C# TCP schema server (CultNetTcpSchemaTransport.ServePeerAsync) likewise ends the one
// peer whose message fails to decode or handle. A handler is still expected to answer its
// own typed refusals on the wire (HandleSnapshotRequestV1Async does, R-AM); this is the
// backstop for whatever it does not. The log names the fault's type only: its message can
// quote what the peer sent.
ICultNetSchemaMessage message;
try
{
message = CultNetSchemaMessageSerialization.Deserialize(delivered.Frame.Payload);
}
catch (Exception ex)
{
// The peer sent something no runtime can read: the refusal is the peer's.
var ended = _transport.EndPeerSession(delivered.Peer, CultNetRudpSession.RefusedPacketReason);
Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); {SessionOutcome(ended)}");
return false;
}
if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null)
{
return false;
}
try
{
var result = handler.DynamicInvoke(message, new RudpCultNetSchemaServerPeer(this, delivered.Peer));
if (result is Task task)
{
await task.ConfigureAwait(false);
}
}
catch (Exception ex)
{
// DynamicInvoke wraps a synchronously-thrown handler fault in TargetInvocationException;
// name the real fault, not that wrapper. The fault is the server's (a response too large
// to send, a handler bug) or a refusal the handler did not type, so the goodbye does not
// say the peer's packet was refused. A response that failed after its first fragment
// already ended the session over the send, and the log says so.
var reported = ex is System.Reflection.TargetInvocationException { InnerException: { } inner } ? inner : ex;
var ended = _transport.EndPeerSession(delivered.Peer, HandlerFailedReason);
Logger.LogError($"CultNet RUDP schema server's {message.GetType().Name} handler failed ({reported.GetType().Name}); {SessionOutcome(ended)}");
return false;
}
return true;
}
///
/// Sends any reliable packets whose resend timers are due.
///
public void PollResends()
{
_transport.PollResends();
}
///
public void Dispose()
{
if (_disposed)
{
return;
}
_disposed = true;
_transport.PeerDisconnected -= OnTransportPeerDisconnected;
_transport.Dispose();
}
private void OnTransportPeerDisconnected(CultNetRudpSocketServerPeer peer)
{
PeerDisconnected?.Invoke(new RudpCultNetSchemaServerPeer(this, peer));
}
}
}