c# - 拆分一个IObservable然后处理后再合并？

Question

在试验了 IObservables 之后，我决定测试它们在消息总线上处理传入消息的能力。基本上我得到一个 IObservable<Request>和 Request包含回复响应的必要功能。

在处理过程中的某一时刻，我必须反序列化数据并将其从 Request 转换为包含它实际需要执行的操作的 Command 对象。命令与请求无关。

反序列化后，我将其转换为正确的响应，但是为了发送响应，我需要原始的 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

我已经圆满地解决了这个问题，但也许还有更好的方法。我创建了一个类似 monadic 的类型，它由两种类型组成:一个值，它是要转换的数据；以及作为周围数据的上下文。

类似下面的内容:

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

我还为上下文和值定义了隐式运算符。还有一些关联的扩展方法允许您将值从一种类型转换为新类型。

如果有人有更好的方法，但我欢迎其他方法，我将暂时不回答这个问题。