Вот простой пример кода с использованием TransformBlock
и ActionBlock
. Я использую библиотеку Polly, чтобы помочь с логикой повтора. Проблема в том, что как только я вручную открываю CircuitBreaker
, а затем снова закрываю его, связь между downloadBlock
и actionBlock
разрывается. Этого не произойдет, если я опущу предикат установки в вызове LinkTo
:
private readonly TransformBlock<DataClass, DataClass> downloadBlock;
private readonly ActionBlock<DataClass> actionBlock;
private readonly AsyncCircuitBreakerPolicy circuitBreaker;
private readonly AsyncRetryPolicy retryPolicy;
retryPolicy = Policy.Handle<WebException>().RetryAsync(4);
circuitBreaker = Policy.Handle<WebException>().CircuitBreakerAsync(10, TimeSpan.FromSeconds(10));
downloadBlock = new TransformBlock<DataClass, DataClass>(async (data) =>
{
var finalPolicy = retryPolicy.WrapAsync(circuitBreaker);
try
{
await finalPolicy.ExecuteAsync(async () =>
{
//await DoSomething();
data.Status = Status.Completed;
});
}
catch (WebException we)
{
data.Status = Status.Failed;
//Do logging
}
catch (BrokenCircuitException)
{
data.Status = Status.Failed;
//Do logging
}
return data;
},
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 });
actionBlock = new ActionBlock<DataClass>((data) =>
{
//DoSomething(data);
}});
//Here, if I set data => data.Status != Status.Failed and then at some point I manually call
//circuitBreaker.Isolate() and then after some time circuitBreaker.Reset() to manually close the circuit again,
//items newly pushed into the pipeline are being processed in the downloadBlock but the results
//are not propagated to actionBlock even if their status is OK. This does not happen if ommit setting the predicate.
downloadBlock.LinkTo(actionBlock,
new DataflowLinkOptions { PropagateCompletion = true }, data => data.Status != Status.Failed);
Я мог бы, конечно, проверить статус в самом actionBlock
, но я хотел бы знать, сталкивался ли кто-нибудь с таким поведением и в чем может быть причина этого?
РЕДАКТИРОВАТЬ: Вот шаги, необходимые для воспроизведения проблемы:
- помещать данные в конвейер, обрабатывать их,
- вызовите circuitBreaker.Isolate (),
- помещать новые данные в конвейер, они попадают в исключение BrokenCircuitException и по логике не переходят к следующему блоку из-за предиката. Хорошо.
- вызовите circuitBreaker.Reset (). Теперь все должно работать.
- помещает новые данные в конвейер, они обрабатываются в downloadBlock, их статус в порядке, но они просто не передаются на следующий ActionBlock
BrokenCircuitException
, поэтому, пожалуйста, уточните свое предложение, потому что непонятно, что вы здесь имели в виду. - person Peter Csala   schedule 14.07.2020circuitBreaker.Isolate()
, 3) отправить новые данные в конвейер, они попадают вBrokenCircuitException
и по логике не переходят к следующему блоку из-за предиката. Хорошо. 4) звонитеcircuitBreaker.Reset()
. Теперь все должно работать. 5) поместите новые данные в конвейер, они будут обработаны вdownloadBlock
, их статус в порядке, но они просто не передаются на следующийActionBlock
. - person niks   schedule 14.07.2020SleepDuration
в номенклатуре Полли. Пожалуйста, добавьте логирование в политики Retry и CB. Здесь я кратко описал, как это можно сделать. - person Peter Csala   schedule 15.07.2020WaitAndRetryAsync(2, retryAttempt => TimeSpan.FromSeconds(2)
Но проблема все еще существует. - person niks   schedule 15.07.2020onRetryAsync
, чтобы проверить, какое исключение было сгенерировано и когда? - person Peter Csala   schedule 15.07.2020WebException
. (Думал стреляет по любомуException
). Не могли бы вы проверитьOutputAvailableAsync
1 до и после вызоваReset
вручную? - person Peter Csala   schedule 15.07.2020Result
является ложным до и после вызоваReset
. Понятия не имею, почему это происходит. Я начинаю думать, что мне нужно просто опустить предикат и позволить всему переходить от блока к блоку, а затем просто проверить статус элемента внутри следующего блока. Я думаю, что это не так уж важно. - person niks   schedule 16.07.2020predicate
будет указывать на использование классаFilteredLinkPropagator
. ЕгоOfferMessage
метод будет решать, распространять ли данные или нет. У меня нет времени по-настоящему вникать в это, но есть случай, когда он возвращается сDataflowMessageStatus.Declined
, что для меня подозрительно. - person Peter Csala   schedule 16.07.2020