RX IObservable come tubazioni

Question

Al momento, sto usando il quadro RX di implementare una gestione dei messaggi conduttura del flusso di lavoro simile. In sostanza ho un produttore di messaggi (deserializza messaggi di rete e chiede OnNext () su un soggetto) e ho diversi consumatori.

. NOTA: Se e trasformare sono metodi di estensione che ho codificato che semplicemente restituire un IObservable

Un consumatore fa qualcosa di simile al seguente:

 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 viene consumato da un altro gasdotto simile e questo continua fino alla cima dove si conclude con qualcuno che chiama Subscribe() sul gasdotto finale. Il problema che sto avendo è che i messaggi dalla base non si propagano fino a meno che la sottoscrizione è chiamato sui messaggi direttamente da qualche parte.

Come posso spingere i messaggi fino alla cima della pila? So che questo è un approccio non ortodosso ma sento che rende il codice molto semplice da capire quello che sta avvenendo a un messaggio. Qualcuno può suggerire un altro modo di fare lo stesso se si sente questa è una pessima idea del tutto?

Answer 1

Perché dovrebbero passare attraverso la pipeline se non ci sono iscritti? Se uno dei vostri passaggi intermedi è utile per i loro effetti collaterali (Si desidera loro di correre, anche se non ci sono altri abbonati), si dovrebbe riscrivere l'operazione effetto collaterale di essere un abbonato.

Si potrebbe anche fare il passo con un effetto collaterale come un'operazione di pass-through (o T, se si vuole) se si voleva continuare la catena.