Split an IObservable and then combine after processing?

Question

After experimenting with IObservables, I've decided to test them for the processing of incoming messages over a message bus. Essentially I get an IObservable<Request> and the Request contains the necessary functions to reply to the response.

At a point during processing I have to deserialize the data and convert it from a Request to a Command object that contains what it actually needs to do. Command is not related to Request.

After deserializing it I transform it into the proper response, however in order to send the response I need the original Request object. I want to try and achieve this while maintaining high code readability. So far I've used extension methods and lambda expressions to get the following (where requests is the 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));  

My question is, since all the commands before the Zip function take time to process, will the Zip operate on the same input object (ie. the original input, and also on the seperate processed input) if there are a constant stream of messages. How can I test this?

Is there a better way of doing this?

Answer 1

I've solved it satisfactorily however there may be a better method. I created a monadic-like type that composes two types: a value which is the data being transformed; and a context which is the surrounding data.

It is something like the following:

 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 also defined implicit operators for both context and value. There are also some associated extension methods that allow you to transform the value from one type to a new type.

If anyone has a better method though I welcome alternatives, and I'm going to leave this unanswered for a while.