Fractionner un objet IObservable, puis le combiner après le traitement?

Question

Après avoir expérimenté avec IObservables, j'ai décidé de les tester pour le traitement des messages entrants sur un bus de messages. Essentiellement, je reçois un IObservable<Request> et le Request contient les fonctions nécessaires pour répondre à la réponse.

À un moment du traitement, je dois désérialiser les données et les convertir d'un objet Request en un objet Command contenant ce qu'il doit réellement faire. La commande n'est pas liée à la demande. Après la désérialisation, je la transforme en la bonne réponse, mais pour envoyer la réponse, j'ai besoin de l'objet Request original. Je veux essayer d'y parvenir tout en conservant une lisibilité élevée du code. Jusqu'à présent, je l'ai utilisé des méthodes d'extension et les expressions lambda pour obtenir ce qui suit (où requests est le 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)); 

Ma question est, puisque toutes les commandes avant l'heure de prise de fonction Zip pour traiter, sera le Zip fonctionner sur le même objet d'entrée (c'est-à-dire l'entrée d'origine, et aussi sur l'entrée traitée séparée) s'il y a un flux constant de messages. Comment puis-je tester cela?

Y a-t-il une meilleure façon de procéder?

Answer 1

Je l'ai résolu de manière satisfaisante, mais il pourrait y avoir une meilleure méthode. J'ai créé un type monadique qui compose deux types: une valeur qui est la donnée en cours de transformation; et un contexte qui est les données environnantes.

Il est quelque chose comme ce qui suit:

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; 
     } 
    } 

I également décrit les opérateurs implicites à la fois le contexte et la valeur. Il existe également des méthodes d'extension associées qui vous permettent de transformer la valeur d'un type en un nouveau type.

Si quelqu'un a une meilleure méthode bien que je salue les alternatives, et je vais laisser cela sans réponse pendant un certain temps.