RX IObservable comme un pipeline

Question

À l'heure actuelle, j'utilise le cadre RX pour mettre en œuvre un pipeline de traitement des messages comme flux de travail. Essentiellement, j'ai un producteur de messages (désérialise messages du réseau et les appels OnNext () sur un sujet) et j'ai plusieurs consommateurs.

NOTE:. Si et transformer des méthodes d'extension, j'ai codé qui retourne simplement un IObservable

Un consommateur fait quelque chose comme ce qui suit:

 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 est alors consommé par un autre pipeline similaire et cela continue jusqu'au sommet où il se termine par une personne appelant Subscribe() sur le pipeline final. La question que je vais avoir est que les messages de la base ne se propagent pas à moins que souscrire est appelé sur les messages directement quelque part.

Comment puis-je pousser les messages au sommet de la pile? Je sais que c'est une approche peu orthodoxe mais je pense qu'il rend le code très simple à comprendre ce qui se passe à un message. Quelqu'un peut-il suggérer une autre façon de faire la même chose si vous pensez que cela est une idée tout à fait terrible?

Answer 1

Pourquoi devraient-ils passer par le pipeline s'il n'y a pas d'abonnés? Si l'une de vos étapes intermédiaires est utile pour leurs effets secondaires (Vous voulez qu'ils fonctionnent même s'il n'y a pas d'autres abonnés), vous devez réécrire l'opération effet secondaire d'être abonné.

Vous pouvez aussi faire l'étape avec un effet secondaire comme une opération pass-through (ou tee, si vous voulez) si vous voulez continuer la chaîne.