Создание наблюдаемых побочных эффектов в блоке Do

Question

С реактивными расширениями обычно предлагается оставить Select чистыми функциями, то есть без каких-либо побочных эффектов. Затем, когда необходим побочный эффект, либо поместите его в конце трубопровода с помощью Subscribe, либо в качестве последнего средства в середине трубопровода с Do.

В идеале я бы воспользовался последней опцией (Do), но в моей текущей ситуации, похожей на следующий блок кода, у меня есть Select с побочными эффектами, так как мне нужно знать результат операции чтобы позже выполнить некоторую логику. Примерами результатов могут быть:

• ли операция выполнена успешно или не выполнена, чтобы показать ошибки пользователям;

• некоторый номер, сбитый операцией, которая необходима в дальнейшем наблюдаемом трубопроводе.

Во всяком случае, вот пример кода:

IObservable<Unit> inputStream = /* generate stream... */; 

IObservable<OperationResult> operationOneStream = inputStream.Select(_ => 
{ 
    try 
    { 
    /* side effect: do something, that can also throw... */ 

    return OperationResult.Success; // use a simple Enum 
    } 
    catch (Exception ex) 
    { 
    /* compensate error: do something ... */ 

    return OperationResult.ErrorInOperationOne; 
    } 
}); 

// operationOneStream is used to build further streams down the chain, it's not "final". 

IObservable<OperationResult> operationTwoStream = operationOneStream 
    .Where(result => result == OperationResult.Success) 
    // here in between some more operations: delay, select returning a stream, switch, ... 
    .Select(_ => 
    { 
    try 
    { 
     /* side effect: do something, that can also throw... */ 

     return OperationResult.Success; 
    } 
    catch (Exception ex) 
    { 
     /* compensate error: do something ... */ 

     return OperationResult.ErrorInOperationTwo; 
    } 
    }); 

// then a third operation, running when nr. 2 is successful 

/* ... */ 

/* Then, operationOne could be merged with other similar operations, 
* so to grab their error streams and treat them in a single block. 
*/ 
IObservable<SomeType> allOperationsStream = Observable.Merge(
    operationOneStream, operationTwoStream /*, ... */); 

IDisposable subscription = allOperationsStream 
    .Where(result => result != OperationResult.Success) 
    .Subscribe(result => /* do something with error */); 

Теперь я хотел бы понять, если есть способ, чтобы перейти от Do к Subscribe для того, чтобы быть еще более категорично о той стороне -последствия. В этом случае Подписка останавливает цепочку, поэтому, чтобы захватить вывод из операции и использовать ее позже, я прибегну к теме и отправлю ее, когда операция завершится или завершится сбой. Но опять же, Субъекты, как правило, обескуражены, если это действительно необходимо.

Есть ли более выразительный код? Спасибо всем

EDIT: после различных комментариев, я добавил больше кода, чтобы показать более полный пример. Также я исправил некоторые ошибки в разоблачении текущей ситуации, которая, я думаю, запуталась.

Answer 1

Не уверен, что я полностью следую за вами, но, возможно, вы просто пытаетесь связать несколько наблюдаемых вместе с помощью selectmany?

(from a in Observable.Return(1) 
from b in Observable.Return(2) 
select b 
).Subscribe(x => {}, ex => {}) 

Вы можете также построить обработку ошибок в этом так:

(from a in Observable.Return(1) 
from result in Observable.Return(2) 
          .Select(x => OperationResult.Success) 
          .Catch<OperationResult, Exception>(ex => { return Observable.Return(OperationResult.Error);}) 
    select result) 
    .Subscribe(x => {}, ex => {});