파이프 라인으로 rx iobservable

Question

현재 RX 프레임 워크를 사용하여 워크 플로와 같은 메시지 처리 파이프 라인을 구현하고 있습니다. 본질적으로 나는 메시지 생산자 (주제에 네트워크 메시지를 제외하고 onnext () 호출)을 가지고 있으며 여러 소비자가 있습니다.

참고 : 변환은 단순히 iobservable을 반환하는 코딩 된 확장 방법이면.

소비자는 다음과 같은 작업을 수행합니다.

 var commerceRequest = messages.Transform(x => GetSomethingFromDatabase(x)
                              .Where(y => y.Value > 5)
                              .Select(y => y.ComplexObject)
                              .If(z => z.IsPaid, respond(z))
                              .Do(z => SendError(z));

commerceRequest 그런 다음 또 다른 유사한 파이프 라인에 의해 소비되며 이것은 누군가가 전화로 끝날 때까지 계속됩니다. Subscribe() 최종 파이프 라인에서. 내가 가진 문제는 구독이 어딘가에 직접 메시지를 요청하지 않으면베이스의 메시지가 전파되지 않는다는 것입니다.

메시지를 스택 상단까지 어떻게 밀어 넣을 수 있습니까? 나는 이것이 정통적인 접근법이라는 것을 알고 있지만, 코드가 메시지에 무슨 일이 일어나고 있는지 이해하는 데 매우 간단하게 생각합니다. 이것이 완전히 끔찍한 생각이라고 생각한다면 다른 방법을 똑같이하는 방법을 제안 할 수 있습니까?

Answer 1

가입자가없는 경우 왜 파이프 라인을 통과해야합니까? 중간 단계 중 하나가 부작용에 유용한 경우 (다른 가입자가 없어도 실행되기를 원합니다), 가입자가되도록 부작용 작업을 다시 작성해야합니다.

체인을 계속하려면 통과 작업 (또는 원한 경우 TEE)으로 부작용으로 단계를 수행 할 수 있습니다.