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 -44
2 changed files with 59 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit de397b95f6 - Show all commits
Telegrator/Polling/ReactiveUpdateReceiver.cs
+4 -4
View File
@@ -1,4 +1,4 @@
using Telegram.Bot; using Telegram.Bot;
using Telegram.Bot.Polling; using Telegram.Bot.Polling;
using Telegram.Bot.Requests; using Telegram.Bot.Requests;
using Telegram.Bot.Types; using Telegram.Bot.Types;
@@ -6,12 +6,12 @@ using Telegram.Bot.Types;
namespace Telegrator.Polling namespace Telegrator.Polling
{ {
/// <summary> /// <summary>
/// Reactive implementation of <see cref="IUpdateReceiver"/> for polling updates from Telegram. /// Reactive update receiver 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> /// </summary>
/// <param name="client">The Telegram bot client for making API requests.</param> /// <param name="client">The Telegram bot client for making API requests.</param>
/// <param name="options">Optional receiver options for configuring update polling behavior.</param> /// <param name="options">Optional receiver options for configuring update polling behavior.</param>
public class ReactiveUpdateReceiver(ITelegramBotClient client, ReceiverOptions? options) : IUpdateReceiver public class ReactiveUpdateReceiver(ITelegramBotClient client, ReceiverOptions? options)
{ {
/// <summary> /// <summary>
/// Gets the receiver options for configuring update polling behavior. /// Gets the receiver options for configuring update polling behavior.
@@ -63,7 +63,7 @@ namespace Telegrator.Polling
try try
{ {
request.Offset = update.Id + 1; request.Offset = update.Id + 1;
_ = updateHandler.HandleUpdateAsync(Client, update, cancellationToken); await updateHandler.HandleUpdateAsync(Client, update, cancellationToken).ConfigureAwait(continueOnCapturedContext: false);
} }
catch (Exception exception2) catch (Exception exception2)
{ {
Telegrator/Polling/UpdateHandlersPool.cs
+24 -9
View File
@@ -60,6 +60,18 @@ namespace Telegrator.Polling
/// <inheritdoc/> /// <inheritdoc/>
public async Task Enqueue(IEnumerable<DescribedHandlerInfo> handlers) 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; Result? lastResult = null;
foreach (DescribedHandlerInfo handlerInfo in handlers) foreach (DescribedHandlerInfo handlerInfo in handlers)
@@ -70,14 +82,10 @@ namespace Telegrator.Polling
continue; continue;
} }
if (ExecutingHandlersSemaphore != null)
{
await ExecutingHandlersSemaphore.WaitAsync().ConfigureAwait(false);
}
try 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); HandlerExecuting?.Invoke(handlerInfo);
using (UpdateHandlerBase instance = handlerInfo.HandlerInstance) using (UpdateHandlerBase instance = handlerInfo.HandlerInstance)
@@ -87,12 +95,12 @@ namespace Telegrator.Polling
await task.ConfigureAwait(false); await task.ConfigureAwait(false);
lastResult = task.Result; lastResult = task.Result;
ExecutingHandlersSemaphore?.Release(1);
} }
if (lastResult.RouteNext) 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) catch (NotImplementedException)
@@ -106,13 +114,20 @@ namespace Telegrator.Polling
} }
catch (Exception ex) 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) if (lastResult != null && !lastResult.RouteNext)
break; break;
} }
} }
finally
{
ExecutingHandlersSemaphore?.Release(1);
}
}, GlobalCancellationToken);
}
/// <summary> /// <summary>
/// Disposes of the handlers pool and releases all resources. /// Disposes of the handlers pool and releases all resources.