拆分iObservable，然后在处理后合并？

Question

在尝试了iObservables之后，我决定测试它们，以通过消息总线处理传入的消息。本质上我得到了 IObservable<Request> 和 Request 包含必要的功能来回复响应。

在处理过程中，我必须对数据进行估算，并将其从请求转换为包含实际需要做的命令对象。命令与请求无关。

估算后，我将其转换为适当的响应，但是为了发送响应，我需要原始的请求对象。我想在保持高码可读性的同时尝试实现这一目标。到目前为止，我使用了扩展方法和lambda表达式来获取以下内容 requests 是个 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));  

我的问题是，因为所有命令都在 Zip 功能需要时间来处理，将 Zip 如果有恒定的消息流，请在相同的输入对象（即原始输入，也是单独的处理输入）上操作。我该如何测试？

有更好的方法吗？

Answer 1

我已经令人满意地解决了它，但是可能有更好的方法。我创建了一种单层状类型，该类型构成了两种类型：一个值是要转换的数据；以及周围数据的上下文。

这是以下内容：

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

我还为上下文和价值定义了隐式操作员。还有一些关联的扩展方法，使您可以将值从一种类型转换为新类型。

如果有人有更好的方法，我欢迎其他选择，我将在一段时间内没有得到答复。