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

Use reactive extension to pair request and response

场景:

我向服务器队列发送请求(我正在使用 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 吗?

您可以使用 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 将 return 在响应返回之前。既然你说这是 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{});