RX IObservable als Pipeline

Question

Derzeit ist ich den RX-Framework mit einer Workflow-ähnlichen Nachrichtenbehandlung Pipeline zu implementieren. Im Grunde habe ich eine Nachricht Produzenten (deserialisiert Netzwerk Nachrichten und Anrufe OnNext () auf einem Gegenstand) und ich habe mehrere Verbraucher.

Hinweis: Wenn und Transformation sind Erweiterungsmethoden I codiert haben, dass einfach eine IObservable zurückkehren

.

Ein Verbraucher macht etwas wie folgt aus:

 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 wird dann durch eine andere ähnliche Pipeline verbraucht und dies wird fortgesetzt, bis die Spitze, wo es endet mit jemandem Subscribe() auf der letzten Pipeline-Aufruf. Das Problem, das ich habe, ist, dass die Nachrichten von der Basis propagieren nicht, es sei denn subscribe auf Nachrichten direkt irgendwo genannt wird.

Wie kann ich die Push-Nachrichten an die Spitze des Stapels nach oben? Ich weiß, dass dies ein unorthodoxer Ansatz ist, aber ich fühle es, den Code sehr einfach macht, um zu verstehen, was mit einer Nachricht auftritt. Kann jemand empfehlen, einen anderen Weg, um das gleiche zu tun, wenn Sie denken, dass dies eine ganz schreckliche Idee ist?

Answer 1

Warum sollten sie durch die Pipeline gehen, wenn es keine Abonnenten? Wenn einer Ihrer Zwischenschritte für ihre Nebenwirkungen nützlich ist (Sie wollen, dass sie selbst laufen, wenn es keine anderen Teilnehmer sind), sollten Sie den Nebeneffekt, den Betrieb neu zu schreiben Abonnent sein.

Sie können auch den Schritt mit Nebenwirkung als Passthrough-Betrieb machen (oder T-Stück, wenn man so will), wenn Sie die Kette fortsetzen wollen.