using System;
using System.Linq;
using System.Threading.Tasks;
namespace GameCult.Networking
{
///
/// Forwards shard writes to an authoritative owner.
///
public interface ICultNetShardWriteForwarder
{
///
/// Forwards a raw document put to the shard owner. Completes only when the owner answered that it admitted the write;
/// throws when it refused, when
/// nothing could have reached it (the only failure after which the id may be forwarded again), and any other exception
/// when the frame may have gone out and was not answered (unconfirmed, including ). The server relays to the original peer exactly what the owner answered.
///
Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message);
///
/// Forwards a raw document delete to the shard owner; completes, refuses and goes unconfirmed as
/// does.
///
Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message);
}
///
/// Options controlling shard write forwarding from a database server bridge.
///
public sealed class CultNetDatabaseServerOptions
{
///
/// Gets or sets whether non-primary writes may be forwarded to shard owners.
///
public bool ForwardNonPrimaryWrites { get; set; }
///
/// Gets or sets the forwarder used when non-primary write forwarding is enabled.
///
public ICultNetShardWriteForwarder? WriteForwarder { get; set; }
///
/// Gets or sets the authority asked for every peer put and delete. None configured refuses them all: a connection
/// key admits a peer to the transport, never to write. A write forwarded to a primary is judged here for the
/// original peer and again at the primary for the forwarder's connection.
///
public ICultNetRemoteWriteAuthority? RemoteWriteAuthority { get; set; }
}
///
/// Options for the schema-v0 client-based shard write forwarder.
///
public sealed class CultNetSchemaWriteForwarderOptions
{
///
/// Gets or sets client security options used to connect to primary endpoints.
///
public ClientSecurityOptions? Security { get; set; }
///
/// Gets or sets how long to wait for a connection before failing the forwarded write.
///
public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
///
/// Gets or sets a callback used to customize each ephemeral forwarding client.
///
public Action? ConfigureClient { get; set; }
///
/// Gets or sets the schema-v0 client factory. Defaults to an endpoint-aware adapter:
/// rudp:// endpoints use the C# RUDP schema client and cultnet:// endpoints use LiteNetLib.
///
public Func? CreateClient { get; set; }
///
/// Gets or sets how long the primary has to answer a forwarded write (from the send). Its receipt completes the forward,
/// its refusal fails it with , and silence fails it with
/// . Defaults to ; a
/// primary that predates write receipts never answers, so forwarding to one is unconfirmed after this window.
///
public TimeSpan ResultWindow { get; set; } = CultNetWriteCompletion.ReceiptResultWindow;
}
///
/// Forwards shard writes over the CultNet schema-v0 client transport.
///
public sealed class CultNetSchemaWriteForwarder : ICultNetShardWriteForwarder
{
private readonly CultNetSchemaWriteForwarderOptions _options;
///
/// Creates a schema-v0 write forwarder.
///
public CultNetSchemaWriteForwarder(CultNetSchemaWriteForwarderOptions? options = null)
{
_options = options ?? new CultNetSchemaWriteForwarderOptions();
}
///
/// Forwards a put under its original message id and completes on the primary's result frame: a receipt returns, a refusal
/// throws and silence throws .
///
public async Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message)
{
if (message == null) throw new ArgumentNullException(nameof(message));
if (shard == null) throw new ArgumentNullException(nameof(shard));
message.ShardId ??= shard.ShardId;
message.ShardEpoch ??= shard.Epoch;
await SendAsync(shard, message, message.MessageId).ConfigureAwait(false);
}
///
/// Forwards a delete under its original message id; completes, refuses and goes unconfirmed as
/// does.
///
public async Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message)
{
if (message == null) throw new ArgumentNullException(nameof(message));
if (shard == null) throw new ArgumentNullException(nameof(shard));
message.ShardId ??= shard.ShardId;
message.ShardEpoch ??= shard.Epoch;
await SendAsync(shard, message, message.MessageId).ConfigureAwait(false);
}
///
/// Parses a CultNet endpoint URI into host and port.
///
public static (string Host, int Port) ParseEndpoint(string endpoint)
{
if (string.IsNullOrWhiteSpace(endpoint))
{
throw new ArgumentException("Endpoint must be non-empty.", nameof(endpoint));
}
if (!Uri.TryCreate(endpoint, UriKind.Absolute, out var uri) ||
!IsSupportedSchemaEndpointScheme(uri.Scheme) ||
string.IsNullOrWhiteSpace(uri.Host))
{
throw new FormatException(
$"CultNet endpoint '{endpoint}' must use cultnet+tcp://host:port, cultnet://host:port, or rudp://host:port.");
}
return (uri.Host, uri.IsDefaultPort ? 3075 : uri.Port);
}
// The write frame has not been handed to the transport until this returns: any failure here is CultNetWriteNotSentException,
// the one failure the caller may retry under the same id.
private async Task ConnectBeforeSendAsync(CultNetShardDescriptor shard, string messageId)
{
ICultNetSchemaClient? client = null;
try
{
var endpoint = ResolvePrimaryEndpoint(shard);
var (host, port) = ParseEndpoint(endpoint);
client = CreateClient(endpoint);
client.Connect(host, port);
await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false);
// The hello comes first on the ordered schema channel: it opts this connection in to the primary's receipts.
client.SendCultNet(new CultNetHelloMessage
{
RuntimeId = "gamecult-networking-shard-forwarder",
RuntimeKind = "dotnet",
SupportedMessageVersions = [CultNetSchemaVersions.WriteAdmitted, CultNetSchemaVersions.ErrorV1]
});
return client;
}
catch (Exception failure)
{
client?.Dispose();
if (failure is CultNetWriteNotSentException) throw;
throw new CultNetWriteNotSentException(messageId, failure);
}
}
private async Task SendAsync(CultNetShardDescriptor shard, T message, string messageId) where T : ICultNetSchemaMessage
{
using var client = await ConnectBeforeSendAsync(shard, messageId).ConfigureAwait(false);
// From the write's send on, every failure (timeout, reset, silence, an exception from the send itself) propagates as it
// is: the frame may have gone out, so the caller treats it as unconfirmed.
await CultNetWriteCompletion.SendAndAwaitReceiptAsync(
(admitted, refuse, refuseV1) =>
{
client.OnCultNet(admitted);
client.OnCultNet(refuse);
client.OnCultNet(refuseV1);
return null;
},
() => client.SendCultNet(message),
messageId,
_options.ResultWindow).ConfigureAwait(false);
}
private ICultNetSchemaClient CreateClient(string endpoint)
{
return _options.CreateClient?.Invoke()
?? CultNetSchemaClients.CreateForEndpoint(endpoint, _options.Security, _options.ConfigureClient);
}
private static bool IsSupportedSchemaEndpointScheme(string scheme)
{
return string.Equals(scheme, "cultnet+tcp", StringComparison.OrdinalIgnoreCase) ||
string.Equals(scheme, "cultnet", StringComparison.OrdinalIgnoreCase) ||
string.Equals(scheme, "rudp", StringComparison.OrdinalIgnoreCase);
}
private async Task WaitForConnectionAsync(ICultNetSchemaClient client, string endpoint)
{
var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout;
while (!client.Connected)
{
if (DateTimeOffset.UtcNow >= deadline)
{
throw new TimeoutException($"Timed out connecting to shard primary endpoint {endpoint}.");
}
await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false);
}
}
private static string ResolvePrimaryEndpoint(CultNetShardDescriptor shard)
{
if (shard == null) throw new ArgumentNullException(nameof(shard));
return shard.PrimaryEndpoints.FirstOrDefault()
?? throw new InvalidOperationException(
$"Shard '{shard.ShardId}' does not advertise a primary endpoint.");
}
}
}