using System; using System.Threading; using System.Threading.Tasks; namespace GameCult.Networking { /// /// A peer refused a write: the typed cultnet.error.v0 it answered with. /// public sealed class CultNetWriteRefusedException : Exception { /// Creates the exception from the refusal frame. public CultNetWriteRefusedException(CultNetErrorMessage refusal) : base(refusal?.Error ?? throw new ArgumentNullException(nameof(refusal))) { Code = refusal.Code; Refusal = refusal; } /// Gets the machine-readable refusal code, e.g. write_not_authorized or message_invalid. public string? Code { get; } /// Gets the refusal frame as received. public CultNetErrorMessage Refusal { get; } } /// /// A write nothing answered within its window: neither admitted nor refused (write-receipt.md, "unconfirmed"). /// public sealed class CultNetWriteUnconfirmedException : Exception { /// Creates the exception for the write . public CultNetWriteUnconfirmedException(string messageId) : base($"No admission or refusal arrived for write '{messageId}'.") { MessageId = messageId; } /// Gets the id of the unconfirmed write. public string MessageId { get; } } /// /// A forwarded write that provably never left this process: the connection failed, or the send failed before any byte was /// written. It is the only forwarding failure after which the same id may be forwarded again. A send port that cannot prove /// nothing was written must not throw this; every other failure of a forward is unconfirmed. /// public sealed class CultNetWriteNotSentException : Exception { /// Creates the exception for the write ; is for logs only. public CultNetWriteNotSentException(string messageId, Exception? cause = null) : base($"Write '{messageId}' was not sent.", cause) { MessageId = messageId; } /// Gets the id of the write that was not sent. public string MessageId { get; } } /// /// Completes a write on the peer's result frame, never on a send's return or a transport acknowledgement: /// an ack proves delivery, not admission. /// public static class CultNetWriteCompletion { /// How long after a send a refusal may still arrive when the caller names no window. public static readonly TimeSpan DefaultResultWindow = TimeSpan.FromMilliseconds(500); /// How long a receipt-capable peer has to answer a write (write-receipt.md, RECEIPT_RESULT_WINDOW_MS). public static readonly TimeSpan ReceiptResultWindow = TimeSpan.FromMilliseconds(5000); /// /// Sends one write to a receipt-capable peer and waits for its answer: a /// write_admitted naming returns, an error.v1 naming it throws /// , and an id-less error (v0, or v1 with no id) is attributed to this write /// because it is the only one in flight on the connection. Silence throws . /// public static async Task SendAndAwaitReceiptAsync( Func, Action, Action, IDisposable?> subscribe, Action send, string messageId, TimeSpan resultWindow, CancellationToken cancellationToken = default) { if (subscribe == null) throw new ArgumentNullException(nameof(subscribe)); if (send == null) throw new ArgumentNullException(nameof(send)); var answer = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using var registration = subscribe( admitted => { if (string.Equals(admitted.MessageId, messageId, StringComparison.Ordinal)) answer.TrySetResult(null); }, error => answer.TrySetResult(error), error => { if (error.MessageId == null || string.Equals(error.MessageId, messageId, StringComparison.Ordinal)) answer.TrySetResult(error.ToV0()); }); send(); var window = Task.Delay(resultWindow, cancellationToken); await Task.WhenAny(answer.Task, window).ConfigureAwait(false); if (!answer.Task.IsCompleted) { await window.ConfigureAwait(false); throw new CultNetWriteUnconfirmedException(messageId); } var refusal = await answer.Task.ConfigureAwait(false); if (refusal != null) throw new CultNetWriteRefusedException(refusal); } /// /// Sends one write and waits for the peer's answer. A refusal frame rejects with /// whatever its order against the transport acknowledgement. A peer /// that admits a write sends no success frame, so the end of the window is the interim bound on a refusal /// (until peers send one); the error frame carries no message id, so any refusal inside the window is this /// write's. /// /// Registers the refusal handler and returns its registration; called before the send. /// Sends the write. /// How long a refusal may take to arrive. /// Cancels the wait. public static async Task SendAndAwaitResultAsync( Func, IDisposable?> subscribe, Action send, TimeSpan resultWindow, CancellationToken cancellationToken = default) { if (subscribe == null) throw new ArgumentNullException(nameof(subscribe)); if (send == null) throw new ArgumentNullException(nameof(send)); var refused = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using var registration = subscribe(error => refused.TrySetResult(error)); send(); var window = Task.Delay(resultWindow, cancellationToken); await Task.WhenAny(refused.Task, window).ConfigureAwait(false); if (refused.Task.IsCompleted) throw new CultNetWriteRefusedException(await refused.Task.ConfigureAwait(false)); await window.ConfigureAwait(false); } } }