Diviser un IObservable puis combiner après le traitement?

Question

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

À un moment au cours du traitement, je dois désérialiser les données et le convertir à partir d'une requête à un objet de commande qui contient ce qu'il faut réellement faire. Commande n'est pas quant à la demande.

Après la désérialisation, je le transformer en la bonne réponse, mais afin d'envoyer la réponse que je besoin de l'objet de la demande initiale. Je veux essayer de parvenir tout en conservant une bonne lisibilité 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, étant donné que toutes les commandes avant que la fonction de Zip prennent du temps à traiter, seront le Zip fonctionner sur le même objet d'entrée (ie. L'entrée d'origine, et aussi sur la individuelle entrée traité) s'il y a un flux constant des messages. Comment puis-je tester cela?

Y at-il une meilleure façon de le faire?

Answer 1

Je l'ai résolu de façon satisfaisante mais il peut y avoir une meilleure méthode. Je créé un type monadique-like qui compose de deux types: une valeur qui est les données transformées; et un contexte qui correspond aux données environnant.

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

J'ai aussi défini les opérateurs implicites à la fois le contexte et la valeur. Il y a aussi quelques méthodes d'extension associés qui vous permettent de transformer la valeur d'un type à un nouveau type.

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