使用响应式扩展来配对请求和响应

Question

场景：

我向服务器队列发送请求（我正在使用 RabbintMQ 异步处理消息）。当服务器处理请求时，它会对不同的队列产生两个响应。

我想使用 RX 来订阅响应，但也可以访问它们的相应请求。

我目前所拥有的：

我使用 EventAggregator，它使用响应式扩展并为事件流公开 IObservable：

public interface IEventAggregator
{
    void Publish<TEvent>(TEvent sampleEvent);
    IObservable<TEvent> GetEvent<TEvent>();
}

当我发送请求时，我发布了一个事件：

eventAggregator.Publish(new RequestSent { ID = Guid.NewGuid(), ... })
//I could later subscribe to the event like this:
//eventAggregator.GetEvent<RequestSent>().Subscribe(..)

当服务器响应时，响应也会发布到 EventAggregator，所以我可以订阅它们：

eventAggregator.GetEvent<ResponseFromQueue1>().Subscribe(OnResponseFromQueue1)

eventAggregator.GetEvent<ResponseFromQueue2>().Subscribe(OnResponseFromQueue2)

我也可以订阅 RequestSent

我需要什么：

private void OnResponseFromQueue1(RequestSent request, ResponseFromQueue1 response)
{
   I need access to both request and respone
}

这样会更好：

private void OnResponse(
    RequestSent request, 
    ResponseFromQueue1 response1, 
    ResponseFromQueue2 response2)
{
   //actually, this would simplify implementation of my logic a lot
}

可以使用 RX 吗？

Answer 1

您可以使用SelectMany，假设您可以使用 ID 之类的东西将请求与响应关联起来

trigger.SelectMany(requestData => {
  //We need to share this
  var id = Guid.NewGuid();

  //Publish your event
  eventAggregator.Publish(new Request { ID = id, /*...*/ });

  //Start listening for the two responses
  //Grab only the first item matching the IDs
  var left = eventAggregator.GetEvent<ResponseFromQueue1>().First(res => res.ID == id);
  var right = eventAggregator.GetEvent<ResponseFromQueue2>().First(res => res.Id == id);

  //We are done once both values have emitted.
  return left.Zip(right);
}, (request, responses) => {
  /*Here you will have access to the request and an array of the responses*/
});

要记住的一点是，此代码现在假设Publish 将在响应返回之前返回。既然您说这是 RabbitMQ，这可能是一个安全的假设，但如果您对此进行任何单元测试，请记住这一点。

编辑

在您的场景中，您实际上会：

//Set up the queue first    
eventAggregator.GetEvent<RequestSent>()
  .SelectMany(requestSent => {
     var id = requestSent.ID;
     var left = eventAggregator.GetEvent<ResponseFromQueue1>().First(res => res.ID == id);
  var right = eventAggregator.GetEvent<ResponseFromQueue2>().First(res => res.ID == id);

  return left.Zip(right);
}, (request, response) => {/**/});

//...Sometime later
eventAggregator.Publish(new Request{});