Dividere un IObservable e poi unire dopo l'elaborazione?

Question

Dopo aver sperimentato con IObservables, ho deciso di provare a loro per l'elaborazione dei messaggi in arrivo su un bus di messaggi. Essenzialmente ottengo un IObservable<Request> e il Request contiene le funzioni necessarie per rispondere alla risposta.

Ad un certo punto durante l'elaborazione devo deserializzare i dati e convertirlo da una richiesta a un oggetto Command che contiene ciò che in realtà deve fare. Il comando non è legato alla richiesta.

Dopo la deserializzazione Mi trasformarlo in risposta corretta, tuttavia al fine di inviare la risposta che ho bisogno l'oggetto richiesta originale. Voglio cercare di raggiungere questo obiettivo, pur mantenendo alta la leggibilità del codice. Finora ho usato metodi di estensione e le espressioni lambda ottenere le seguenti (dove requests è il IObservable<Request>):

requestProcessor = requests
            .Deserialize<IdentityRequest>()
            .Where(idRequest => idRequest.Address != null)
            .Select(idRequest => new IdentityResponse() {Identity = identityTable[idRequest.Address.Address]})
            .Serialize()
            .Zip(requests, (response, request) => new {request = request, response = response})
            .Subscribe(data => data.request.Respond(data.response, ObjectTypes.IdentityResponse));  

mia domanda è, dato che tutti i comandi prima che la funzione Zip richiedono tempo per elaborare, sarà il Zip operare sull'oggetto stesso ingresso (es. L'ingresso originale, e anche sull'ingresso trasformati separata) se ci sono un flusso costante di messaggi. Come posso provare questo?

C'è un modo migliore di fare questo?

Answer 1

Ho risolto in modo soddisfacente tuttavia ci può essere un metodo migliore. Ho creato un tipo monadic-like che compone di due tipi: un valore che sia i dati che vengono trasformati; e un contesto che è il dato circostante.

Si tratta di qualcosa di simile al seguente:

 public class ComposedType<TValue, TContext>
 {
       public TValue Value { get; protected set; }
       public TContext Context { get; protected set; }

       public ComposedType(TValue value, TContext context)
       {
            Value = value;
            Context = context;
       }
  }

Ho anche definito operatori impliciti sia per contesto e valore. Ci sono anche alcuni metodi di estensione associati che permettono di trasformare il valore da un tipo a un nuovo tipo.

Se qualcuno ha un metodo migliore se Accolgo con favore le alternative, e ho intenzione di lasciare questo senza risposta per un po '.