c# – 使用被动扩展以正确的顺序处理多个响应

情况

我有一个系统,其中一个请求产生两个响应.请求和响应具有相应的可观察量：

IObservable<RequestSent> _requests;
IObservable<MainResponseReceived> _mainResponses;
IObservable<SecondResponseReceived> _secondaryResponses;

保证RequestSent事件早于MainResponseReceived和SecondaryResponseReceived发生,但响应按随机顺序排列.

是)我有的

最初我想要处理两个响应的处理程序,所以我压缩了observables：

_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
>发生SecondaryResponseReceived – >调用OnMainAndSecondaryResponseReceived

测试案例2：

> RequestSent发生
>发生SecondaryResponseReceived
> MainResponseReceived发生 – > OnMainResponseReceived被称为 – >调用OnMainAndSecondaryResponseReceived

解决方法

我认为你几乎走在正确的轨道上.我会停止使用所有Async的东西 – 这只是让事情复杂化.

试试这个查询：

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),s) => new MainAndSecondaryResponseReceived()
                    {
                        Request = request,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 });