RX IObserver подписки времени

Question

Я новичок в Rx в .net, но начали использовать его с некоторым успехом подведению некоторых Коммуникационная сеть

Очень упрощенный пример:

IObservable<Result> SendRequest(Request request) 
    { 
     return Observable.Create<Result>(observer => 
     { 
      NetworkComms.SendReqeust(request, 
       result => 
       { 
        observer.OnNext(result); 
        observer.OnCompleted(); 
       }); 
      return Disposable.Empty; 
     }).SubscribeOn(_commsScheduler); 
    } 

Проблема у меня есть заключается в том, что команда (NetworkComms.SendRequest) фактически не отправляется до тех пор, пока вызывающий абонент не присоединится к возвращенному IObservable. В некоторых случаях запрос является командой «огонь-и-забыть», и результат довольно бессмыслен, поэтому для абонента не имеет смысла фактически подписываться на возвращаемый IObservable.

функциональность мне нужно:

1. Команда отправляется немедленно, даже если абонент никогда не присоединяется к IObservable
2. Если и когда клиент выписывает, они получат результат, даже если они поздняя подписка
3. Команда отправляется только один раз, но все подписки должны получать одинаковые результаты.

Я попытался сделать это с использованием .Replay(). RefCount() и выполнить функцию Subscribe() перед возвратом IObservable. Это почти работает, но в том случае, когда клиент подписывает после результат, полученный (и, следовательно, после автоматического удаления последовательности после завершения), он вызывает повторный вызов кода подписки, отправляя команду во второй раз.

Есть ли простое расширение Rx, которое может справиться с этим сценарием для меня, или мне нужно катиться самостоятельно?

Answer 1

Похоже, что вы хотите использовать AsyncSubject<T>:

IObservable<Result> SendRequest(Request request) 
{ 
    var subject = new AsyncSubject<Result>(); 
    NetworkComms.SendReqeust(request, result => 
    { 
     subject.OnNext(result); 
     subject.OnCompleted(); 
    }); 
    return subject.AsObservable(); 
} 

Я хотел бы добавить, что интерфейс вы хотите немного странно. Если семантика - это «В тот момент, когда я вызываю этот метод, запрос делается, и любой, кто хочет получить ответ позже», тогда рассмотрим только использование Task<Result> вместо IObservable<Result>.

Task<Result> SendRequestAsync(Request request) 
{ 
    var tcs = new TaskCompletionSource<Result>(); 
    NetworkComms.SendReqeust(request, result => tcs.SetResult(result)); 
    return tcs.Task; 
} 

Кроме того, рассмотреть вопрос о создании этого Task<Result> SendRequestAsync(Request request) метод непосредственно на NetworkComms класса.