使用反应式扩展以正确的顺序处理多个响应
本文关键字:顺序处理 响应 反应式 扩展 | 更新日期: 2023-09-27 18:18:27
>情况
我有一个系统,一个请求产生两个响应。请求和响应具有相应的可观察量:
IObservable<RequestSent> _requests;
IObservable<MainResponseReceived> _mainResponses;
IObservable<SecondResponseReceived> _secondaryResponses;
可以保证RequestSent
事件发生早于MainResponseReceived
和SecondaryResponseReceived
但响应以随机顺序出现。
我有什么
最初我想要处理两个响应的处理程序,所以我压缩了可观察量:
_requests
.SelectMany(async request =>
{
var main = _mainResponses.FirstAsync(m => m.Id == request.Id);
var secondary = _secondaryResponses.FirstAsync(s => s.Id == request.Id);
var zippedResponse = main.Zip(secondary, (m, s) => new MainAndSecondaryResponseReceived {
Request = request,
Main = m,
Secondary = s
});
return await zippedResponse.FirstAsync(); ;
})
.Subscribe(OnMainAndSecondaryResponseReceived);
<小时 />我需要什么
现在我还需要处理MainResponseReceived
,而无需等待 SecondaryResponseRecieved,并且必须保证 OnMainResponseRecieved 在调用 OnMainAndSecondaryResponseReceived
之前完成
请问如何定义这两个订阅?
测试用例 1:
- 发生
RequestSent
-
MainResponseReceived
发生 -> OnMainResponseReceived 被调用 -
SecondaryResponseReceive
d 发生 -> 调用 OnMainAndSecondaryResponseReceived
测试用例 2:
- 发生
RequestSent
- 发生
SecondaryResponseReceived
-
MainResponseReceived occurs
-> OnMainResponseReceived 被称为 -> OnMainAndSecondaryResponseReceived 被调用
我认为你走在正确的轨道上。我会停止在所有异步的东西上胡思乱想 - 那只会让事情变得复杂。
请尝试以下查询:
var query =
_requests
.SelectMany(request =>
_mainResponses.Where(m => m.Id == request.Id).Take(1)
.Do(m => OnMainResponseReceived(m))
.Zip(
_secondaryResponses.Where(s => s.Id == request.Id).Take(1),
(m, s) => new MainAndSecondaryResponseReceived()
{
Request = request,
Main = m,
Secondary = s
}));
var subscription =
query.Subscribe(x => OnMainAndSecondaryResponseReceived(x));
.Do(...)
是代码中缺少的重要部分。它确保在OnMainAndSecondaryResponseReceived
之前调用OnMainResponseReceived
,无论主响应还是次要响应是先传入。
我用以下方法对此进行了测试:
Subject<RequestSent> _requestsSubject = new Subject<RequestSent>();
Subject<MainResponseReceived> _mainResponsesSubject = new Subject<MainResponseReceived>();
Subject<SecondResponseReceived> _secondaryResponsesSubject = new Subject<SecondResponseReceived>();
IObservable<RequestSent> _requests = _requestsSubject.AsObservable();
IObservable<MainResponseReceived> _mainResponses = _mainResponsesSubject.AsObservable();
IObservable<SecondResponseReceived> _secondaryResponses = _secondaryResponsesSubject.AsObservable();
_requestsSubject.OnNext(new RequestSent() { Id = 42 });
_mainResponsesSubject.OnNext(new MainResponseReceived() { Id = 42 });
_secondaryResponsesSubject.OnNext(new SecondResponseReceived() { Id = 42 });
_requestsSubject.OnNext(new RequestSent() { Id = 99 });
_mainResponsesSubject.OnNext(new MainResponseReceived() { Id = 99 });
_secondaryResponsesSubject.OnNext(new SecondResponseReceived() { Id = 99 });