/
vshmidt
/
masstransit
Обзор
Документация
Войти
/
vshmidt
/
masstransit
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
develop
src/MassTransit/Middleware/TimeoutFilter.cs
51 строка
2 KB
Chris Patterson
Updating cancellation token handling for consumers, sagas, and routing slip activities for consistent handling of consumer-generated OperationCanceledExceptions
04 апр 2023, 15:50
04 апр 2023, 15:50
849f6c0
Код
Авторство
О чём код?
namespace MassTransit.Middleware { using System; using System.Threading; using System.Threading.Tasks; public class TimeoutFilter<TContext, TResult> : IFilter<TContext> where TContext : class, ConsumeContext where TResult : TContext { readonly Func<TContext, CancellationToken, TResult> _contextFactory; readonly TimeSpan _timeout; public TimeoutFilter(Func<TContext, CancellationToken, TResult> contextFactory, TimeSpan timeout) { _contextFactory = contextFactory; _timeout = timeout; } public async Task Send(TContext context, IPipe<TContext> next) { var cts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken); try { cts.CancelAfter(_timeout); var timeoutContext = _contextFactory(context, cts.Token); await next.Send(timeoutContext).ConfigureAwait(false); await timeoutContext.ConsumeCompleted.ConfigureAwait(false); } catch (OperationCanceledException ex) when (ex.CancellationToken == cts.Token && !context.CancellationToken.IsCancellationRequested) { throw new ConsumerCanceledException("The operation was canceled by the timeout filter"); } finally { cts.Dispose(); } } public void Probe(ProbeContext context) { var scope = context.CreateFilterScope("timeout"); scope.Add("timeout", _timeout); } } }