< Summary

Information
Class: TeleFlow.Telegram.TelegramLongPollingClient
Assembly: TeleFlow.Telegram.LongPolling
File(s): /_/src/TeleFlow.Telegram.LongPolling/TelegramLongPollingClient.cs
Line coverage
89%
Covered lines: 183
Uncovered lines: 22
Coverable lines: 205
Total lines: 438
Line coverage: 89.2%
Branch coverage
89%
Covered branches: 61
Total branches: 68
Branch coverage: 89.7%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
.ctor(...)100%11100%
RunAsync()92.85%201469.23%
GetUpdatesAsync()95%202094.23%
GetUpdatesBatchAsync()100%11100%
LogStart(...)100%22100%
GetElapsedMilliseconds(...)100%210%
DelayAsync(...)50%22100%
HandleDecodeFailureAsync()60%101093.1%
CopyAllowedUpdates(...)100%22100%
IsPollingTransient(...)100%66100%
GetPollingRetryDelay(...)100%44100%

File(s)

/_/src/TeleFlow.Telegram.LongPolling/TelegramLongPollingClient.cs

#LineLine coverage
 1using System.Runtime.CompilerServices;
 2using Microsoft.Extensions.Logging;
 3using TeleFlow.Telegram.Internal;
 4using TeleFlow.Telegram.Schema.Methods;
 5using TeleFlow.Telegram.Schema.Types;
 6
 7namespace TeleFlow.Telegram;
 8
 9/// <summary>
 10/// Receives Telegram updates through getUpdates, preserves acknowledgement ordering, and applies explicit schema decode
 11/// </summary>
 12public sealed partial class TelegramLongPollingClient : ITelegramLongPollingClient
 13{
 14    private readonly ITelegramUpdateBatchReceiver _batchReceiver;
 15    private readonly ITelegramUpdateDecodeFailurePolicy _decodeFailurePolicy;
 16    private readonly TimeProvider _timeProvider;
 17    private readonly ILogger<TelegramLongPollingClient> _logger;
 18
 19    public TelegramLongPollingClient(
 20        ITelegramClient telegramClient,
 21        TimeProvider timeProvider,
 22        ILoggerFactory loggerFactory)
 723        : this(
 724            new TypedTelegramUpdateBatchReceiver(telegramClient),
 725            StopTelegramUpdateDecodeFailurePolicy.Instance,
 726            timeProvider,
 727            loggerFactory)
 28    {
 729    }
 30
 31    internal TelegramLongPollingClient(
 32        ITelegramUpdateBatchReceiver batchReceiver,
 33        ITelegramUpdateDecodeFailurePolicy decodeFailurePolicy,
 34        TimeProvider timeProvider,
 35        ILoggerFactory loggerFactory)
 36    {
 3137        ArgumentNullException.ThrowIfNull(batchReceiver);
 3138        ArgumentNullException.ThrowIfNull(decodeFailurePolicy);
 3139        ArgumentNullException.ThrowIfNull(timeProvider);
 3140        ArgumentNullException.ThrowIfNull(loggerFactory);
 41
 3142        _batchReceiver = batchReceiver;
 3143        _decodeFailurePolicy = decodeFailurePolicy;
 3144        _timeProvider = timeProvider;
 3145        _logger = loggerFactory.CreateLogger<TelegramLongPollingClient>();
 3146    }
 47
 48    public async Task RunAsync(
 49        Func<Update, CancellationToken, Task> updateHandler,
 50        TelegramRawLongPollingOptions? options = null,
 51        CancellationToken cancellationToken = default)
 52    {
 1053        ArgumentNullException.ThrowIfNull(updateHandler);
 54
 1055        options ??= new TelegramRawLongPollingOptions();
 1056        TelegramRawLongPollingOptionsValidator.Validate(options);
 57
 858        long? offset = null;
 859        var backoff = new TelegramRawLongPollingBackoff(options.Backoff);
 860        var allowedUpdates = CopyAllowedUpdates(options.AllowedUpdates);
 861        var connected = false;
 862        var recoveringFromPollingFailure = false;
 63
 864        LogStart(options, allowedUpdates);
 65
 1566        while (!cancellationToken.IsCancellationRequested)
 67        {
 1168            var updates = await GetUpdatesBatchAsync(
 1169                offset,
 1170                options,
 1171                allowedUpdates,
 1172                backoff,
 1173                () =>
 1174                {
 1175                    if (!connected)
 1176                    {
 877                        LogConnected(_logger);
 878                        connected = true;
 1179                    }
 1180
 1181                    if (recoveringFromPollingFailure)
 1182                    {
 283                        LogGetUpdatesRecovered(_logger);
 284                        recoveringFromPollingFailure = false;
 1185                    }
 1186                },
 387                () => recoveringFromPollingFailure = true,
 1188                cancellationToken).ConfigureAwait(false);
 89
 4090            for (var index = 0; index < updates.Count; index++)
 91            {
 1392                var item = updates[index];
 1393                if (!item.IsSuccess)
 94                {
 495                    await HandleDecodeFailureAsync(item, cancellationToken).ConfigureAwait(false);
 196                    offset = item.UpdateId + 1;
 197                    continue;
 98                }
 99
 9100                var update = item.Update!;
 101
 9102                if (_logger.IsEnabled(LogLevel.Debug))
 103                {
 0104                    var updateType = TelegramRawLongPollingLogFormatter.GetUpdateType(update);
 0105                    var processingStarted = _timeProvider.GetTimestamp();
 106
 0107                    LogUpdateReceived(
 0108                        _logger,
 0109                        update.UpdateId,
 0110                        updateType,
 0111                        index + 1,
 0112                        updates.Count);
 113
 0114                    await updateHandler(update, cancellationToken).ConfigureAwait(false);
 0115                    offset = update.UpdateId + 1;
 116
 0117                    LogUpdateAcknowledgedByHandler(
 0118                        _logger,
 0119                        update.UpdateId,
 0120                        updateType,
 0121                        GetElapsedMilliseconds(processingStarted));
 0122                    continue;
 123                }
 124
 9125                await updateHandler(update, cancellationToken).ConfigureAwait(false);
 8126                offset = update.UpdateId + 1;
 8127            }
 7128        }
 4129    }
 130
 131    public async IAsyncEnumerable<TelegramPolledUpdate> GetUpdatesAsync(
 132        TelegramRawLongPollingOptions? options = null,
 133        [EnumeratorCancellation] CancellationToken cancellationToken = default)
 134    {
 19135        options ??= new TelegramRawLongPollingOptions();
 19136        TelegramRawLongPollingOptionsValidator.Validate(options);
 137
 19138        long? offset = null;
 19139        var backoff = new TelegramRawLongPollingBackoff(options.Backoff);
 19140        var allowedUpdates = CopyAllowedUpdates(options.AllowedUpdates);
 19141        var connected = false;
 19142        var recoveringFromPollingFailure = false;
 143
 19144        LogStart(options, allowedUpdates);
 145
 38146        while (!cancellationToken.IsCancellationRequested)
 147        {
 24148            var updates = await GetUpdatesBatchAsync(
 24149                offset,
 24150                options,
 24151                allowedUpdates,
 24152                backoff,
 24153                () =>
 24154                {
 23155                    if (!connected)
 24156                    {
 18157                        LogConnected(_logger);
 18158                        connected = true;
 24159                    }
 24160
 23161                    if (recoveringFromPollingFailure)
 24162                    {
 4163                        LogGetUpdatesRecovered(_logger);
 4164                        recoveringFromPollingFailure = false;
 24165                    }
 23166                },
 5167                () => recoveringFromPollingFailure = true,
 24168                cancellationToken).ConfigureAwait(false);
 169
 82170            for (var index = 0; index < updates.Count; index++)
 171            {
 22172                var item = updates[index];
 22173                if (!item.IsSuccess)
 174                {
 0175                    await HandleDecodeFailureAsync(item, cancellationToken).ConfigureAwait(false);
 0176                    offset = item.UpdateId + 1;
 0177                    continue;
 178                }
 179
 22180                var update = item.Update!;
 22181                var polledUpdate = new TelegramPolledUpdate(update, index + 1, updates.Count);
 182
 22183                var debugEnabled = _logger.IsEnabled(LogLevel.Debug);
 22184                var updateType = debugEnabled
 22185                    ? TelegramRawLongPollingLogFormatter.GetUpdateType(update)
 22186                    : string.Empty;
 187
 22188                if (debugEnabled)
 189                {
 2190                    LogStreamUpdateReceived(
 2191                        _logger,
 2192                        update.UpdateId,
 2193                        updateType);
 194                }
 195
 22196                yield return polledUpdate;
 197
 19198                if (!polledUpdate.IsAcknowledged)
 199                {
 1200                    throw new InvalidOperationException(
 1201                        "Telegram polled updates must be acknowledged with AcknowledgeAsync before requesting the next u
 202                }
 203
 18204                offset = update.UpdateId + 1;
 205
 18206                if (debugEnabled)
 207                {
 1208                    LogStreamUpdateAcknowledged(
 1209                        _logger,
 1210                        update.UpdateId,
 1211                        updateType);
 212                }
 18213            }
 19214        }
 17215    }
 216
 217    private async Task<IReadOnlyList<TelegramUpdateDecodeResult>> GetUpdatesBatchAsync(
 218        long? offset,
 219        TelegramRawLongPollingOptions options,
 220        IReadOnlyList<string>? allowedUpdates,
 221        TelegramRawLongPollingBackoff backoff,
 222        Action onSuccess,
 223        Action onTransientFailure,
 224        CancellationToken cancellationToken)
 225    {
 226        while (true)
 227        {
 228            try
 229            {
 43230                var updates = await _batchReceiver.ReceiveAsync(
 43231                    new GetUpdates
 43232                    {
 43233                        Offset = offset,
 43234                        Limit = options.Limit,
 43235                        Timeout = options.TimeoutSeconds,
 43236                        AllowedUpdates = allowedUpdates
 43237                    },
 43238                    cancellationToken).ConfigureAwait(false);
 239
 34240                onSuccess();
 34241                backoff.Reset();
 34242                return updates;
 243            }
 9244            catch (Exception exception) when (IsPollingTransient(exception) && !cancellationToken.IsCancellationRequeste
 245            {
 8246                var delay = GetPollingRetryDelay(exception, backoff);
 8247                onTransientFailure();
 248
 8249                LogGetUpdatesFailed(
 8250                    _logger,
 8251                    exception,
 8252                    delay);
 253
 8254                await DelayAsync(delay, cancellationToken).ConfigureAwait(false);
 8255            }
 256        }
 34257    }
 258
 259    private void LogStart(
 260        TelegramRawLongPollingOptions options,
 261        IReadOnlyList<string>? allowedUpdates)
 262    {
 27263        if (!_logger.IsEnabled(LogLevel.Information))
 264        {
 25265            return;
 266        }
 267
 2268        LogStarting(
 2269            _logger,
 2270            TelegramRawLongPollingLogFormatter.FormatAllowedUpdates(allowedUpdates),
 2271            options.TimeoutSeconds,
 2272            options.Limit);
 2273    }
 274
 275    private double GetElapsedMilliseconds(long startingTimestamp)
 276    {
 0277        return _timeProvider.GetElapsedTime(startingTimestamp).TotalMilliseconds;
 278    }
 279
 280    private ValueTask DelayAsync(TimeSpan delay, CancellationToken cancellationToken)
 281    {
 8282        return delay <= TimeSpan.Zero
 8283            ? ValueTask.CompletedTask
 8284            : new ValueTask(Task.Delay(delay, _timeProvider, cancellationToken));
 285    }
 286
 287    private async ValueTask HandleDecodeFailureAsync(
 288        TelegramUpdateDecodeResult item,
 289        CancellationToken cancellationToken)
 290    {
 4291        var exception = item.Exception
 4292            ?? throw new InvalidOperationException("A failed Telegram update decode result did not contain an exception.
 4293        var failure = new TelegramUpdateDecodeFailure(
 4294            TelegramUpdateTransport.LongPolling,
 4295            item.UpdateId,
 4296            item.RawPayloadJson
 4297                ?? throw new InvalidOperationException("A failed Telegram update decode result did not contain raw paylo
 4298            item.PayloadSha256
 4299                ?? throw new InvalidOperationException("A failed Telegram update decode result did not contain a payload
 4300            exception);
 301
 4302        var decision = await _decodeFailurePolicy
 4303            .DecideAsync(failure, cancellationToken)
 4304            .ConfigureAwait(false);
 305
 306        switch (decision)
 307        {
 308            case TelegramUpdateDecodeFailureDecision.Stop:
 1309                LogUpdateDecodeStopped(
 1310                    _logger,
 1311                    exception,
 1312                    item.UpdateId,
 1313                    exception.JsonPath,
 1314                    exception.PayloadSha256);
 1315                throw exception;
 316            case TelegramUpdateDecodeFailureDecision.Skip:
 1317                LogUpdateDecodeSkipped(
 1318                    _logger,
 1319                    item.UpdateId,
 1320                    exception.JsonPath,
 1321                    exception.PayloadSha256);
 1322                return;
 323            default:
 0324                throw new InvalidOperationException(
 0325                    $"Unsupported Telegram update decode failure decision '{decision}'.");
 326        }
 1327    }
 328
 329    private static string[]? CopyAllowedUpdates(IReadOnlyList<string>? allowedUpdates)
 330    {
 27331        return allowedUpdates is null ? null : allowedUpdates.ToArray();
 332    }
 333
 334    private static bool IsPollingTransient(Exception exception)
 335    {
 9336        return exception is TelegramNetworkException or
 9337            TelegramServerException or
 9338            TelegramRetryAfterException;
 339    }
 340
 341    private static TimeSpan GetPollingRetryDelay(Exception exception, TelegramRawLongPollingBackoff backoff)
 342    {
 8343        return exception is TelegramRetryAfterException { RetryAfter: { } retryAfter }
 8344            ? retryAfter
 8345            : backoff.NextDelay();
 346    }
 347
 348    [LoggerMessage(
 349        EventId = 1,
 350        Level = LogLevel.Information,
 351        Message = "Starting raw Telegram long polling. allowed_updates={AllowedUpdates}, timeout={TimeoutSeconds}s, limi
 352    private static partial void LogStarting(
 353        ILogger logger,
 354        string allowedUpdates,
 355        int timeoutSeconds,
 356        int limit);
 357
 358    [LoggerMessage(
 359        EventId = 2,
 360        Level = LogLevel.Information,
 361        Message = "Raw Telegram long polling connected.")]
 362    private static partial void LogConnected(ILogger logger);
 363
 364    [LoggerMessage(
 365        EventId = 3,
 366        Level = LogLevel.Information,
 367        Message = "Raw Telegram long polling getUpdates recovered after transient failures.")]
 368    private static partial void LogGetUpdatesRecovered(ILogger logger);
 369
 370    [LoggerMessage(
 371        EventId = 4,
 372        Level = LogLevel.Debug,
 373        Message = "Raw Telegram update received. update_id={UpdateId}, type={UpdateType}, batch_index={BatchIndex}/{Batc
 374    private static partial void LogUpdateReceived(
 375        ILogger logger,
 376        long updateId,
 377        string updateType,
 378        int batchIndex,
 379        int batchCount);
 380
 381    [LoggerMessage(
 382        EventId = 5,
 383        Level = LogLevel.Debug,
 384        Message = "Raw Telegram update acknowledged by handler. update_id={UpdateId}, type={UpdateType}, total_ms={Total
 385    private static partial void LogUpdateAcknowledgedByHandler(
 386        ILogger logger,
 387        long updateId,
 388        string updateType,
 389        double totalElapsedMilliseconds);
 390
 391    [LoggerMessage(
 392        EventId = 6,
 393        Level = LogLevel.Debug,
 394        Message = "Raw Telegram update received. update_id={UpdateId}, type={UpdateType}.")]
 395    private static partial void LogStreamUpdateReceived(
 396        ILogger logger,
 397        long updateId,
 398        string updateType);
 399
 400    [LoggerMessage(
 401        EventId = 7,
 402        Level = LogLevel.Debug,
 403        Message = "Raw Telegram update acknowledged. update_id={UpdateId}, type={UpdateType}.")]
 404    private static partial void LogStreamUpdateAcknowledged(
 405        ILogger logger,
 406        long updateId,
 407        string updateType);
 408
 409    [LoggerMessage(
 410        EventId = 8,
 411        Level = LogLevel.Warning,
 412        Message = "Raw Telegram long polling getUpdates failed. Retrying in {Delay}.")]
 413    private static partial void LogGetUpdatesFailed(
 414        ILogger logger,
 415        Exception exception,
 416        TimeSpan delay);
 417
 418    [LoggerMessage(
 419        EventId = 9,
 420        Level = LogLevel.Error,
 421        Message = "Telegram update decoding failed and polling was stopped. update_id={UpdateId}, json_path={JsonPath}, 
 422    private static partial void LogUpdateDecodeStopped(
 423        ILogger logger,
 424        Exception exception,
 425        long updateId,
 426        string? jsonPath,
 427        string payloadSha256);
 428
 429    [LoggerMessage(
 430        EventId = 10,
 431        Level = LogLevel.Warning,
 432        Message = "Telegram update decoding failed and the update was skipped by application policy. update_id={UpdateId
 433    private static partial void LogUpdateDecodeSkipped(
 434        ILogger logger,
 435        long updateId,
 436        string? jsonPath,
 437        string payloadSha256);
 438}
Generated by: ReportGenerator 5.5.10.0
09/18/2026 - 04:43:19
GitHub | reportgenerator.io

Methods/Properties

.ctor(TeleFlow.Telegram.ITelegramClient,System.TimeProvider,Microsoft.Extensions.Logging.ILoggerFactory)
.ctor(TeleFlow.Telegram.Internal.ITelegramUpdateBatchReceiver,TeleFlow.Telegram.ITelegramUpdateDecodeFailurePolicy,System.TimeProvider,Microsoft.Extensions.Logging.ILoggerFactory)
RunAsync()
GetUpdatesAsync()
GetUpdatesBatchAsync()
LogStart(TeleFlow.Telegram.TelegramRawLongPollingOptions,System.Collections.Generic.IReadOnlyList`1<System.String>)
GetElapsedMilliseconds(System.Int64)
DelayAsync(System.TimeSpan,System.Threading.CancellationToken)
HandleDecodeFailureAsync()
CopyAllowedUpdates(System.Collections.Generic.IReadOnlyList`1<System.String>)
IsPollingTransient(System.Exception)
GetPollingRetryDelay(System.Exception,TeleFlow.Telegram.Internal.TelegramRawLongPollingBackoff)