< Summary

Information
Class: Services.ChannelTaskServices.ChannelTaskProcessor
Assembly: Services
File(s): /home/runner/work/Studio2Prod/Studio2Prod/AppRepository/ChannelTaskServices/ChannelTaskProcessor.cs
Line coverage
92%
Covered lines: 113
Uncovered lines: 9
Coverable lines: 122
Total lines: 201
Line coverage: 92.6%
Branch coverage
97%
Covered branches: 33
Total branches: 34
Branch coverage: 97%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
.ctor(...)100%11100%
Submit(...)100%44100%
TryGet(...)100%11100%
TryGetCompletedFile(...)91.66%121292.3%
RegisterCompletedFile(...)100%11100%
UpdateTask(...)100%22100%
StartAsync(...)100%11100%
StopAsync()100%2271.42%
Dispose()100%22100%
ProcessQueueAsync()100%2262.5%
RunWorkItemAsync()100%11100%
CleanupExpired()100%121075%

File(s)

/home/runner/work/Studio2Prod/Studio2Prod/AppRepository/ChannelTaskServices/ChannelTaskProcessor.cs

#LineLine coverage
 1using System.Collections.Concurrent;
 2using System.Text.Json;
 3using System.Threading.Channels;
 4using Microsoft.Extensions.Hosting;
 5using Microsoft.Extensions.Logging;
 6
 7namespace Services.ChannelTaskServices;
 8
 9public sealed class ChannelTaskProcessor : IChannelTaskService, IHostedService, IDisposable
 10{
 111    internal static readonly string ExportDirectory = Path.Combine(Path.GetTempPath(), "st2-channel-tasks");
 112    private static readonly TimeSpan TaskTtl = TimeSpan.FromHours(4);
 113    private static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCas
 14
 15    private readonly Channel<ChannelTaskWorkItem> _channel;
 1816    private readonly ConcurrentDictionary<string, ChannelTaskState> _tasks = new(StringComparer.Ordinal);
 1817    private readonly ConcurrentDictionary<string, string> _completedFiles = new(StringComparer.Ordinal);
 18    private readonly ChannelTaskExecutor _executor;
 19    private readonly ILogger<ChannelTaskProcessor> _logger;
 1820    private readonly CancellationTokenSource _cts = new();
 1821    private readonly SemaphoreSlim _semaphore = new(4, 4);
 1822    private readonly object _cleanupLock = new();
 23    private Task? _processingTask;
 24
 1825    public ChannelTaskProcessor(ChannelTaskExecutor executor, ILogger<ChannelTaskProcessor> logger)
 26    {
 1827        _executor = executor;
 1828        _logger = logger;
 1829        var options = new BoundedChannelOptions(200)
 1830        {
 1831            FullMode = BoundedChannelFullMode.Wait,
 1832            SingleReader = true,
 1833            SingleWriter = false
 1834        };
 1835        _channel = Channel.CreateBounded<ChannelTaskWorkItem>(options);
 1836    }
 37
 38    public ChannelTaskState Submit(ChannelTaskType type, object payload)
 39    {
 21840        CleanupExpired();
 21841        var taskId = Guid.NewGuid().ToString("N");
 21842        var payloadJson = JsonSerializer.Serialize(payload, JsonOptions);
 21843        var state = new ChannelTaskState
 21844        {
 21845            TaskId = taskId,
 21846            Type = type,
 21847            Status = ChannelTaskStages.Queued,
 21848            Stage = ChannelTaskStages.Queued,
 21849            Message = "Задача поставлена в очередь",
 21850            Percent = 0,
 21851            ReportId = payload is BitrixSyncReportTaskPayload bitrixPayload ? bitrixPayload.ReportId : null
 21852        };
 53
 21854        _tasks[taskId] = state;
 55
 21856        if (!_channel.Writer.TryWrite(new ChannelTaskWorkItem
 21857            {
 21858                TaskId = taskId,
 21859                Type = type,
 21860                PayloadJson = payloadJson
 21861            }))
 62        {
 163            _tasks.TryRemove(taskId, out _);
 164            throw new InvalidOperationException("Очередь задач переполнена. Повторите позже.");
 65        }
 66
 21767        return state;
 68    }
 69
 70    public bool TryGet(string taskId, out ChannelTaskState? state)
 471        => _tasks.TryGetValue(taskId, out state);
 72
 73    public bool TryGetCompletedFile(string taskId, out string? filePath, out string? fileName)
 74    {
 1075        filePath = null;
 1076        fileName = null;
 1077        if (!_tasks.TryGetValue(taskId, out var state))
 078            return false;
 79
 1080        if (state.Status != ChannelTaskStages.Completed
 1081            || state.ResultKind != ChannelTaskResultKind.File
 1082            || string.IsNullOrWhiteSpace(state.FileName))
 883            return false;
 84
 285        if (!_completedFiles.TryGetValue(taskId, out var path) || !File.Exists(path))
 186            return false;
 87
 188        filePath = path;
 189        fileName = state.FileName;
 190        return true;
 91    }
 92
 93    private void RegisterCompletedFile(string taskId, string filePath)
 1394        => _completedFiles[taskId] = filePath;
 95
 96    private void UpdateTask(string taskId, Action<ChannelTaskState> update)
 97    {
 1398        if (_tasks.TryGetValue(taskId, out var state))
 1399            update(state);
 13100    }
 101
 102    public Task StartAsync(CancellationToken cancellationToken)
 103    {
 2104        _processingTask = ProcessQueueAsync(_cts.Token);
 2105        return Task.CompletedTask;
 106    }
 107
 108    public async Task StopAsync(CancellationToken cancellationToken)
 109    {
 2110        _channel.Writer.TryComplete();
 2111        _cts.Cancel();
 2112        if (_processingTask != null)
 113        {
 4114            try { await _processingTask.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken); }
 0115            catch (TimeoutException) { _logger.LogWarning("Channel task processor stop timeout"); }
 0116            catch (OperationCanceledException) { }
 117        }
 2118    }
 119
 120    public void Dispose()
 121    {
 19122        if (!_cts.IsCancellationRequested)
 16123            _cts.Cancel();
 19124        _cts.Dispose();
 19125        _semaphore.Dispose();
 19126    }
 127
 128    private async Task ProcessQueueAsync(CancellationToken ct)
 129    {
 130        try
 131        {
 8132            await foreach (var item in _channel.Reader.ReadAllAsync(ct))
 133            {
 2134                await _semaphore.WaitAsync(ct);
 4135                _ = RunWorkItemAsync(item, ct).ContinueWith(_ => _semaphore.Release(), CancellationToken.None);
 2136            }
 2137        }
 0138        catch (OperationCanceledException)
 139        {
 0140            _logger.LogInformation("Channel task processor stopped");
 0141        }
 2142    }
 143
 144    private async Task RunWorkItemAsync(ChannelTaskWorkItem item, CancellationToken ct)
 145    {
 146        try
 147        {
 2148            UpdateTask(item.TaskId, s =>
 2149            {
 2150                s.Status = ChannelTaskStages.Queued;
 2151                s.Stage = ChannelTaskStages.Queued;
 2152                s.Message = "Задача в обработке";
 2153                s.StartedAtUtc = DateTime.UtcNow;
 4154            });
 155
 2156            await _executor.ExecuteAsync(
 2157                item.TaskId,
 2158                item.Type,
 2159                item.PayloadJson,
 2160                UpdateTask,
 2161                RegisterCompletedFile,
 2162                ct);
 1163        }
 1164        catch (Exception ex)
 165        {
 1166            _logger.LogError(ex, "Channel task failed. taskId={TaskId}, type={Type}", item.TaskId, item.Type);
 1167            UpdateTask(item.TaskId, s =>
 1168            {
 1169                s.Status = ChannelTaskStages.Failed;
 1170                s.Stage = ChannelTaskStages.Failed;
 1171                s.Error = ex.Message;
 1172                s.Message = "Ошибка выполнения задачи";
 1173                s.CompletedAtUtc = DateTime.UtcNow;
 2174            });
 1175        }
 2176    }
 177
 178    private void CleanupExpired()
 179    {
 218180        lock (_cleanupLock)
 181        {
 218182            var border = DateTime.UtcNow - TaskTtl;
 40642183            foreach (var pair in _tasks)
 184            {
 20103185                var finishedAt = pair.Value.CompletedAtUtc ?? pair.Value.CreatedAtUtc;
 20103186                if (finishedAt >= border)
 187                    continue;
 188
 1189                _tasks.TryRemove(pair.Key, out _);
 1190                if (_completedFiles.TryRemove(pair.Key, out var filePath))
 191                {
 3192                    try { if (File.Exists(filePath)) File.Delete(filePath); }
 0193                    catch (Exception exception)
 194                    {
 0195                        _logger.LogWarning(exception, "Не удалось удалить временный файл завершённой channel-задачи {Tas
 0196                    }
 197                }
 198            }
 199        }
 218200    }
 201}
Generated by: ReportGenerator 5.5.11.0
10/25/2026 - 07:43:48
GitHub | reportgenerator.io

Methods/Properties

.cctor()
.ctor(Services.ChannelTaskServices.ChannelTaskExecutor,Microsoft.Extensions.Logging.ILogger`1<Services.ChannelTaskServices.ChannelTaskProcessor>)
Submit(Services.ChannelTaskServices.ChannelTaskType,System.Object)
TryGet(System.String,Services.ChannelTaskServices.ChannelTaskState&)
TryGetCompletedFile(System.String,System.String&,System.String&)
RegisterCompletedFile(System.String,System.String)
UpdateTask(System.String,System.Action`1<Services.ChannelTaskServices.ChannelTaskState>)
StartAsync(System.Threading.CancellationToken)
StopAsync()
Dispose()
ProcessQueueAsync()
RunWorkItemAsync()
CleanupExpired()