🦙 Move long polling to a separate task

Author: database64128

CubicBot.Telegram.App/CubicBot.Telegram.App.csproj

@@ -13,6 +13,7 @@
 
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Http" Version="10.0.0" />
   </ItemGroup>
 
   <ItemGroup>

CubicBot.Telegram.App/LongPollingBotService.cs

@@ -0,0 +1,81 @@
+using Telegram.Bot;
+using Telegram.Bot.Types;
+using Telegram.Bot.Types.Enums;
+
+namespace CubicBot.Telegram.App;
+
+public sealed partial class LongPollingBotService(ILogger<LongPollingBotService> logger, IHttpClientFactory httpClientFactory) : BotService(logger, httpClientFactory.CreateClient())
+{
+    private Task? _pollUpdatesTask;
+
+    public override async Task StartAsync(CancellationToken cancellationToken)
+    {
+        (TelegramBotClient bot, CancellationTokenSource cts) = await StartBotAsync(cancellationToken);
+        _pollUpdatesTask = PollUpdatesAsync(bot, cts.Token);
+    }
+
+    private async Task PollUpdatesAsync(ITelegramBotClient botClient, CancellationToken cancellationToken = default)
+    {
+        int? offset = null;
+
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            try
+            {
+                Update[] updates = await botClient.GetUpdates(offset, allowedUpdates: [UpdateType.Message], cancellationToken: cancellationToken);
+
+                if (updates.Length > 0)
+                {
+                    offset = updates[^1].Id + 1;
+
+                    foreach (Update update in updates)
+                    {
+                        await UpdateWriter.WriteAsync(update, cancellationToken);
+                    }
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                return;
+            }
+            catch (Exception ex)
+            {
+                LogFailedToGetUpdates(ex);
+
+                try
+                {
+                    await Task.Delay(s_delayOnError, cancellationToken);
+                }
+                catch (TaskCanceledException)
+                {
+                    return;
+                }
+            }
+        }
+    }
+
+    private static readonly TimeSpan s_delayOnError = TimeSpan.FromSeconds(5);
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "Failed to get updates")]
+    private partial void LogFailedToGetUpdates(Exception ex);
+
+    public override async Task StopAsync(CancellationToken cancellationToken)
+    {
+        if (Cts is null)
+        {
+            return;
+        }
+
+        try
+        {
+            await base.StopAsync(cancellationToken);
+        }
+        finally
+        {
+            if (_pollUpdatesTask is not null)
+            {
+                await _pollUpdatesTask.WaitAsync(cancellationToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+            }
+        }
+    }
+}

CubicBot.Telegram.App/Program.cs

@@ -1,7 +1,8 @@
-using CubicBot.Telegram;
+using CubicBot.Telegram.App;
 
 var builder = Host.CreateApplicationBuilder(args);
-builder.Services.AddHostedService<BotService>();
+builder.Services.AddHttpClient();
+builder.Services.AddHostedService<LongPollingBotService>();
 
 var host = builder.Build();
 host.Run();

CubicBot.Telegram.App/appsettings.json

@@ -2,7 +2,8 @@
   "Logging": {
     "LogLevel": {
       "Default": "Information",
-      "Microsoft.Hosting.Lifetime": "Information"
+      "Microsoft.Hosting.Lifetime": "Information",
+      "System.Net.Http.HttpClient.Default": "Warning"
     }
   }
 }

CubicBot.Telegram.App/appsettings.systemd.json

@@ -2,7 +2,8 @@
   "Logging": {
     "LogLevel": {
       "Default": "Information",
-      "Microsoft.Hosting.Lifetime": "Information"
+      "Microsoft.Hosting.Lifetime": "Information",
+      "System.Net.Http.HttpClient.Default": "Warning"
     }
   },
   "Console": {

CubicBot.Telegram/BotService.cs

@@ -1,16 +1,39 @@
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using System.Threading.Channels;
 using Telegram.Bot;
 using Telegram.Bot.Exceptions;
 using Telegram.Bot.Types;
 
 namespace CubicBot.Telegram;
 
-public sealed partial class BotService(ILogger<BotService> logger) : BackgroundService
+public partial class BotService : IHostedService
 {
-    protected override Task ExecuteAsync(CancellationToken stoppingToken) => RunBotAsync(stoppingToken);
+    private readonly ILogger<BotService> _logger;
+    private readonly HttpClient _httpClient;
+    private readonly ChannelReader<Update> _updateReader;
 
-    private async Task RunBotAsync(CancellationToken cancellationToken = default)
+    public BotService(ILogger<BotService> logger, HttpClient httpClient)
+    {
+        _logger = logger;
+        _httpClient = httpClient;
+        Channel<Update> updateChannel = Channel.CreateBounded<Update>(new BoundedChannelOptions(64)
+        {
+            SingleReader = true,
+        });
+        _updateReader = updateChannel.Reader;
+        UpdateWriter = updateChannel.Writer;
+    }
+
+    public ChannelWriter<Update> UpdateWriter { get; }
+
+    protected CancellationTokenSource? Cts { get; private set; }
+    private Task? _saveDataTask;
+    private Task? _handleUpdatesTask;
+
+    public virtual Task StartAsync(CancellationToken cancellationToken) => StartBotAsync(cancellationToken);
+
+    protected async Task<(TelegramBotClient, CancellationTokenSource)> StartBotAsync(CancellationToken cancellationToken = default)
     {
         Config config;
 
@@ -45,8 +68,7 @@ private async Task RunBotAsync(CancellationToken cancellationToken = default)
         {
             RetryCount = 7,
         };
-        using HttpClient httpClient = new();
-        TelegramBotClient bot = new(options, httpClient);
+        TelegramBotClient bot = new(options, _httpClient);
         User me;
 
         while (true)
@@ -58,47 +80,43 @@ private async Task RunBotAsync(CancellationToken cancellationToken = default)
             }
             catch (RequestException ex)
             {
-                logger.LogWarning(ex, "Failed to get bot info, retrying in 30 seconds");
-                await Task.Delay(30000, cancellationToken);
+                _logger.LogWarning(ex, "Failed to get bot info, retrying in 30 seconds");
+                await Task.Delay(s_delayOnError, cancellationToken);
             }
         }
 
         if (string.IsNullOrEmpty(me.Username))
             throw new Exception("Bot username is null or empty.");
 
-        var updateHandler = new UpdateHandler(me.Username, config, data, logger);
+        var updateHandler = new UpdateHandler(me.Username, config, data, _logger, bot);
 
         while (true)
         {
             try
             {
-                await updateHandler.RegisterCommandsAsync(bot, cancellationToken);
+                await updateHandler.RegisterCommandsAsync(cancellationToken);
                 break;
             }
             catch (RequestException ex)
             {
-                logger.LogWarning(ex, "Failed to register commands, retrying in 30 seconds");
-                await Task.Delay(30000, cancellationToken);
+                _logger.LogWarning(ex, "Failed to register commands, retrying in 30 seconds");
+                await Task.Delay(s_delayOnError, cancellationToken);
             }
         }
 
         LogStartedBot(me.Username, me.Id);
 
-        var saveDataTask = SaveDataHourlyAsync(data, cancellationToken);
+        Cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        _saveDataTask = SaveDataHourlyAsync(data, Cts.Token);
+        _handleUpdatesTask = updateHandler.RunAsync(_updateReader, Cts.Token);
 
-        try
-        {
-            await updateHandler.RunAsync(bot, cancellationToken);
-        }
-        finally
-        {
-            await saveDataTask;
-        }
+        return (bot, Cts);
     }
 
     [LoggerMessage(Level = LogLevel.Information, Message = "Started Telegram bot: @{BotUsername} ({BotId})")]
     private partial void LogStartedBot(string botUsername, long botId);
 
+    private static readonly TimeSpan s_delayOnError = TimeSpan.FromSeconds(30);
     private static readonly TimeSpan s_saveDataInterval = TimeSpan.FromHours(1);
 
     private async Task SaveDataHourlyAsync(Data data, CancellationToken cancellationToken = default)
@@ -116,11 +134,36 @@ private async Task SaveDataHourlyAsync(Data data, CancellationToken cancellation
             try
             {
                 await data.SaveAsync(CancellationToken.None);
-                logger.LogDebug("Saved data");
+                _logger.LogDebug("Saved data");
             }
             catch (Exception ex)
             {
-                logger.LogError(ex, "Failed to save data");
+                _logger.LogError(ex, "Failed to save data");
+            }
+        }
+    }
+
+    public virtual async Task StopAsync(CancellationToken cancellationToken)
+    {
+        if (Cts is null)
+        {
+            return;
+        }
+
+        try
+        {
+            Cts.Cancel();
+        }
+        finally
+        {
+            if (_saveDataTask is not null)
+            {
+                await _saveDataTask.WaitAsync(cancellationToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+            }
+
+            if (_handleUpdatesTask is not null)
+            {
+                await _handleUpdatesTask.WaitAsync(cancellationToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
             }
         }
     }