From d34b8cbc419b34b2e38e2331b1b51c39ff204d21 Mon Sep 17 00:00:00 2001 From: slothdpal <16717792+SlothDpal@users.noreply.github.com> Date: Sun, 20 Sep 2026 21:00:23 +0300 Subject: [PATCH] Improved queue by locking _isProcessing flag. Using one static HttpClient. Added processing 429 (rate-limit) from Discord server. --- Discord/DiscordWebHook.cs | 145 +++++++++++++++++++++++++++++-------- Properties/AssemblyInfo.cs | 6 +- 2 files changed, 117 insertions(+), 34 deletions(-) diff --git a/Discord/DiscordWebHook.cs b/Discord/DiscordWebHook.cs index 2c2b733..3ec3fc7 100644 --- a/Discord/DiscordWebHook.cs +++ b/Discord/DiscordWebHook.cs @@ -2,6 +2,7 @@ using System.Collections.Concurrent; using System.Diagnostics; using System.IO; +using System.Net; using System.Net.Http; using System.Text; using System.Threading; @@ -9,16 +10,41 @@ using System.Threading.Tasks; namespace Discord.Webhook { + // Результат одной попытки отправки. Позволяет разобраться в причине + // неудачи (в т.ч. rate limit от Discord) + public sealed class DiscordSendResult + { + public bool Success { get; private set; } + public HttpStatusCode? StatusCode { get; private set; } + public string ErrorBody { get; private set; } + public TimeSpan? RetryAfter { get; private set; } + public Exception Exception { get; private set; } + + public static DiscordSendResult Ok() => new DiscordSendResult { Success = true }; + public static DiscordSendResult Fail(HttpStatusCode? statusCode = null, string errorBody = null, TimeSpan? retryAfter = null, Exception exception = null) + => new DiscordSendResult + { + Success = false, + StatusCode = statusCode, + ErrorBody = errorBody, + RetryAfter = retryAfter, + Exception = exception + }; + } + public class DiscordWebhook { public string Url { get; set; } public int queueRetryCount = 3; public int sendTimeoutSeconds = 5; - + // используем общий экземпляр HttpClient для всех запросов, чтобы избежать проблем с исчерпанием сокетов + // HttpClient предназначен для повторного использования и потокобезопасен. + private static readonly HttpClient _httpClient = new HttpClient(); private UInt64 totalMessages = 0; private ConcurrentQueue<(UInt64 num, DiscordMessage message, FileInfo[] files)> _queue = new ConcurrentQueue<(UInt64 num, DiscordMessage, FileInfo[])>(); private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1); private bool _isProcessing; + private readonly object _isProcessingFlagLock = new object(); private CancellationTokenSource _cts = new CancellationTokenSource(); private int queueErrorCounter = 0; private int queueSuppressedCounter = 0; @@ -31,14 +57,13 @@ namespace Discord.Webhook public bool IsProcessing => _isProcessing; public HttpRequestException LastHpptEx => lastHpptEx; - public async Task SendAsync(DiscordMessage message, params FileInfo[] files) + public async Task SendAsync(DiscordMessage message, params FileInfo[] files) { if (string.IsNullOrEmpty(Url)) throw new ArgumentNullException("Invalid Webhook URL."); string boundary = "------------------------" + DateTime.Now.Ticks.ToString("x"); - using (var client = new HttpClient() { Timeout = TimeSpan.FromSeconds(sendTimeoutSeconds) }) using (var content = new MultipartFormDataContent(boundary)) { // Добавляем JSON payload @@ -65,31 +90,75 @@ namespace Discord.Webhook } } - try + // эмулируем таймаут соединения, используя CancellationTokenSource с заданным временем ожидания + // если запрос не завершится за указанное время, будет выброшено исключение TaskCanceledException + // так как HttpClient не имеет встроенного таймаута для отдельных запросов, мы используем + // CancellationTokenSource для управления временем ожидания + using (var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(sendTimeoutSeconds))) + using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token, timeoutCts.Token)) { - var response = await client.PostAsync(Url, content, _cts.Token); - response.EnsureSuccessStatusCode(); - } - catch (HttpRequestException ex) - { - Debug.WriteLine($"SendAsync: Discord webhook request failed: {ex.Message}"); - lastHpptEx = ex; - return false; - } - catch (TaskCanceledException ex) - { - Debug.WriteLine($"SendAsync: Discord webhook request cancelled: {ex.Message}"); - return false; + HttpResponseMessage response; + try + { + response = await _httpClient.PostAsync(Url, content, linkedCts.Token); + } + catch (HttpRequestException ex) + { + Debug.WriteLine($"SendAsync: Discord webhook request failed: {ex.Message}"); + lastHpptEx = ex; + return DiscordSendResult.Fail(exception: ex); + } + catch (TaskCanceledException ex) + { + bool wasTimeout = timeoutCts.IsCancellationRequested && !_cts.IsCancellationRequested; + Debug.WriteLine(wasTimeout + ? $"SendAsync: Discord webhook request timed out after {sendTimeoutSeconds}s." + : $"SendAsync: Discord webhook request cancelled: {ex.Message}"); + return DiscordSendResult.Fail(exception: ex); + } + + using (response) + { + if (response.IsSuccessStatusCode) + { + return DiscordSendResult.Ok(); + } + + // Discord возвращает 429 при превышении rate limit и указывает, + // сколько секунд подождать, в заголовке Retry-After. + if (response.StatusCode == (HttpStatusCode)429) + { + TimeSpan? retryAfter = response.Headers.RetryAfter?.Delta; + string body429 = await SafeReadBodyAsync(response); + Debug.WriteLine($"SendAsync: rate limited by Discord. Retry-After: {retryAfter?.TotalSeconds ?? -1}s. Body: {body429}"); + return DiscordSendResult.Fail(statusCode: response.StatusCode, errorBody: body429, retryAfter: retryAfter); + } + + string body = await SafeReadBodyAsync(response); + Debug.WriteLine($"SendAsync: Discord webhook request failed ({(int)response.StatusCode}): {body}"); + return DiscordSendResult.Fail(statusCode: response.StatusCode, errorBody: body); + } } } - return true; + } + + private static async Task SafeReadBodyAsync(HttpResponseMessage response) + { + try + { + return await response.Content.ReadAsStringAsync(); + } + catch + { + return string.Empty; + } } private async Task ProcessQueueAsync() { queueErrorCounter = 0; - while (_queue.TryPeek(out var item)) + while (_queue.TryPeek(out var queueItem)) { if (_cts.Token.IsCancellationRequested) { @@ -97,10 +166,12 @@ namespace Discord.Webhook break; } await _semaphore.WaitAsync(); + DiscordSendResult result; try { - Debug.WriteLine($"ProcessQueueAsync: Processing message {item.num}. Queue size: {_queue.Count}"); - if (await SendAsync(item.message, item.files)) + Debug.WriteLine($"ProcessQueueAsync: Processing message {queueItem.num}. Queue size: {_queue.Count}"); + result = await SendAsync(queueItem.message, queueItem.files); + if (result.Success) { _queue.TryDequeue(out var deqItem); queueErrorCounter = 0; @@ -123,7 +194,10 @@ namespace Discord.Webhook } try { - await Task.Delay(1000, _cts.Token); // Discord rate limit: 1 message per second + // Если Discord вернул 429 с Retry-After — ждём именно столько, + // иначе используем стандартный лимит 1 сообщение в секунду. + TimeSpan delay = result.RetryAfter ?? TimeSpan.FromSeconds(1); + await Task.Delay(delay, _cts.Token); } catch (TaskCanceledException) { @@ -131,7 +205,7 @@ namespace Discord.Webhook break; } } - if ( _cts.IsCancellationRequested) + if (_cts.IsCancellationRequested) { Debug.WriteLine($"ProcessQueueAsync: Discord queue processing cancelled. Was {_queue.Count} messages in queue."); Debug.WriteLine("Clearing queue."); @@ -139,22 +213,31 @@ namespace Discord.Webhook Interlocked.Exchange(ref _queue, _newqueue); } Debug.WriteLine($"ProcessQueueAsync: Discord queue processing finished."); - _isProcessing = false; + lock (_isProcessingFlagLock) + { + _isProcessing = false; + } } public void Send(DiscordMessage message, params FileInfo[] files) { _queue.Enqueue((totalMessages++, message, files)); - Debug.WriteLine($"Message {totalMessages-1} added. Queue size: {_queue.Count}"); - if (_isProcessing) + Debug.WriteLine($"Message {totalMessages - 1} added. Queue size: {_queue.Count}"); + + lock (_isProcessingFlagLock) { - Debug.WriteLine("Already processing queue."); - return; + if (_isProcessing) + { + Debug.WriteLine("Already processing queue."); + return; + } + + _isProcessing = true; + _cts.Dispose(); + _cts = new CancellationTokenSource(); } + Debug.WriteLine("Run ProcessQueueAsync"); - _cts.Dispose(); - _cts = new CancellationTokenSource(); - _isProcessing = true; Task.Run(ProcessQueueAsync); return; } diff --git a/Properties/AssemblyInfo.cs b/Properties/AssemblyInfo.cs index fb42d28..d01d108 100644 --- a/Properties/AssemblyInfo.cs +++ b/Properties/AssemblyInfo.cs @@ -29,7 +29,7 @@ using System.Runtime.InteropServices; // Номер сборки // Редакция // -[assembly: AssemblyVersion("1.6.1.2")] -[assembly: AssemblyFileVersion("1.6.1.2")] +[assembly: AssemblyVersion("1.6.1.7")] +[assembly: AssemblyFileVersion("1.6.1.7")] -[assembly: AssemblyInformationalVersion("1.6.1.2")] \ No newline at end of file +[assembly: AssemblyInformationalVersion("1.6.1.7")] \ No newline at end of file