如何定义两个订阅，请？

测试案例1：

测试案例2：

我觉得你是pretty的多少在正确的轨道。我将停止摆弄周围所有的异步的东西 - 那只是使事情复杂化

试试这个查询：

  VAR的查询=
    _要求
        .SelectMany（要求=＆GT;
            _mainResponses.Where（米=＆GT; m.Id == request.Id）。取（1）
                。做（米=＆GT; OnMainResponseReceived（米））
                。拉链（
                    _secondaryResponses.Where（S =＆GT; s.Id == request.Id）。取（1），
                    （M，S）=＆GT;新MainAndSecondaryResponseReceived（）
                    {
                        请求=请求，
                        主要= M，
                        二级= S
                    }））;

VAR订阅=
    query.Subscribe（X =＆GT; OnMainAndSecondaryResponseReceived（x）的）;
 

在。做（...）中的重要缺失的部分在code。它确保了 OnMainResponseReceived 在 OnMainAndSecondaryResponseReceived 被称为无论如果主要或次要响应进来第一。

我测试用：

 受试对象; RequestSent＆GT; _requestsSubject =新的受试对象; RequestSent＆GT;（）;
受试对象; MainResponseReceived＆GT; _mainResponsesSubject =新的受试对象; MainResponseReceived＆GT;（）;
受试对象; SecondResponseReceived＆GT; _secondaryResponsesSubject =新的受试对象; SecondResponseReceived＆GT;（）;

的IObservable＆LT; RequestSent＆GT; _requests = _requestsSubject.AsObservable（）;
的IObservable＆LT; MainResponseReceived＆GT; _mainResponses = _mainResponsesSubject.AsObservable（）;
的IObservable＆LT; SecondResponseReceived＆GT; _secondaryResponses = _secondaryResponsesSubject.AsObservable（）;

_requestsSubject.OnNext（新RequestSent（）{n = 42}）;
_mainResponsesSubject.OnNext（新MainResponseReceived（）{n = 42}）;
_secondaryResponsesSubject.OnNext（新SecondResponseReceived（）{n = 42}）;

_requestsSubject.OnNext（新RequestSent（）{n = 99}）;
_mainResponsesSubject.OnNext（新MainResponseReceived（）{n = 99}）;
_secondaryResponsesSubject.OnNext（新SecondResponseReceived（）{n = 99}）;
 

Situation

I have a system where one request produces two responses. The request and responses have corresponding observables:

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

it is guaranteed that RequestSent event occurs earlier than MainResponseReceived and SecondaryResponseReceived but the responses come in random order.

What I have

Originally I wanted handler that handles both responses, so I zipped the 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);

What I need

Now I need to handle also MainResponseReceived without waiting for SecondaryResponseRecieved and it must be guaranteed, that the OnMainResponseRecieved completes before OnMainAndSecondaryResponseReceived is called

How to define the two subscriptions, please?

Test case 1:

Test case 2:

I think you're pretty much on the right track. I would stop mucking around with all the Async stuff - that's just making things complicated.

Try this query:

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));

The .Do(...) is the important missing part in your code. It ensures that OnMainResponseReceived is called before OnMainAndSecondaryResponseReceived no matter if the main or secondary response comes in first.

I tested this with:

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 });