Rikitav/Telegrator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 13 Wiki Activity

WIP: Make the UpdateRecevier not wait for each handler #4

Closed
XSilverTH wants to merge 3 commits from parallelhandler into master
pull from: parallelhandler
merge into: Rikitav:master
Conversation 2 Commits 3 Files Changed 2
+59 -43
2 changed files with 59 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Telegrator/Polling/ReactiveUpdateReceiver.cs
+4 -3
View File
@@ -1,4 +1,4 @@
using Telegram.Bot;
using Telegram.Bot;
using Telegram.Bot.Polling;
using Telegram.Bot.Requests;
using Telegram.Bot.Types;
@@ -7,7 +7,7 @@ namespace Telegrator.Polling
{
/// <summary>
/// Reactive implementation of <see cref="IUpdateReceiver"/> for polling updates from Telegram.
/// Provides custom update receiving logic with error handling and configuration options.
/// /// Provides custom update receiving logic with error handling and configuration options.
/// </summary>
/// <param name="client">The Telegram bot client for making API requests.</param>
/// <param name="options">Optional receiver options for configuring update polling behavior.</param>
@@ -67,7 +67,8 @@ namespace Telegrator.Polling
}
catch (Exception exception2)
{
await updateHandler.HandleErrorAsync(Client, exception2, HandleErrorSource.HandleUpdateError, cancellationToken).ConfigureAwait(false);
await updateHandler.HandleErrorAsync(Client, exception2,
HandleErrorSource.HandleUpdateError, cancellationToken).ConfigureAwait(false);
}
}
}
Telegrator/Polling/UpdateHandlersPool.cs
+24 -9
View File
@@ -60,6 +60,18 @@ namespace Telegrator.Polling
/// <inheritdoc/>
public async Task Enqueue(IEnumerable<DescribedHandlerInfo> handlers)
{
if (ExecutingHandlersSemaphore != null)
{
await ExecutingHandlersSemaphore.WaitAsync().ConfigureAwait(false);
}
// Offload the entire processing of this update's handlers to a background task.
// This allows the Receiver to continue polling for NEW updates immediately,
// while this update acts as a self-contained unit of work.
_ = Task.Run(async () =>
{
try
{
Result? lastResult = null;
foreach (DescribedHandlerInfo handlerInfo in handlers)
@@ -70,14 +82,10 @@ namespace Telegrator.Polling
continue;
}
if (ExecutingHandlersSemaphore != null)
{
await ExecutingHandlersSemaphore.WaitAsync().ConfigureAwait(false);
}
try
{
Alligator.LogDebug("Described handler '{0}' (Update {1})", handlerInfo.DisplayString, handlerInfo.HandlingUpdate.Id);
Alligator.LogDebug("Described handler '{0}' (Update {1})", handlerInfo.DisplayString,
handlerInfo.HandlingUpdate.Id);
HandlerExecuting?.Invoke(handlerInfo);
using (UpdateHandlerBase instance = handlerInfo.HandlerInstance)
@@ -87,12 +95,12 @@ namespace Telegrator.Polling
await task.ConfigureAwait(false);
lastResult = task.Result;
ExecutingHandlersSemaphore?.Release(1);
}
if (lastResult.RouteNext)
{
Alligator.LogTrace("Handler '{0}' requested route continuation (Update {1})", handlerInfo.DisplayString, handlerInfo.HandlingUpdate.Id);
Alligator.LogTrace("Handler '{0}' requested route continuation (Update {1})",
handlerInfo.DisplayString, handlerInfo.HandlingUpdate.Id);
}
}
catch (NotImplementedException)
@@ -106,13 +114,20 @@ namespace Telegrator.Polling
}
catch (Exception ex)
{
Alligator.LogError("Failed to process handler '{0}' (Update {1})", exception: ex, handlerInfo.DisplayString, handlerInfo.HandlingUpdate.Id);
Alligator.LogError("Failed to process handler '{0}' (Update {1})", exception: ex,
handlerInfo.DisplayString, handlerInfo.HandlingUpdate.Id);
}
if (lastResult != null && !lastResult.RouteNext)
break;
}
}
finally
{
ExecutingHandlersSemaphore?.Release(1);
}
}, GlobalCancellationToken);
}
/// <summary>
/// Disposes of the handlers pool and releases all resources.