如何创建从MSMQ消息队列中读取的IObservable <T>？

Question

我正在从我们的ASP.NET站点删除我们的电子邮件系统,该站点过去常常通过系统发送电子邮件来处理单独服务中的请求,以减少网站上的工作量.我正在尝试围绕一组接口进行设计,以便我可以根据需要交换实现,但最初它将基于消息队列(MSMQ)将请求发送到队列,让服务接收传入请求然后处理它们.我目前大致定义了以下接口:

// Sends one or more requests to be processed somehow
public interface IRequestSender
{
    void Send(IEnumerable<Request> requests);
}

// Listens for incoming requests and passes them to an observer to do the real work
public interface IRequestListener : IObservable<Request>
{
    void Start();
    void Stop();
}

// Processes a request given to it by a IRequestListener
public interface IRequestProcessor : IObserver<Request>
{
}

您会注意到监听器和处理器使用可观察的模式,因为这是我认为最适合的.

我的问题是弄清楚如何编写IRequestListener从MSMQ接收的实现,基本上我该如何创建一个合适的IObservable<T>？

我找到的第一个选项是IObservable<T>根据MSDN文档给出的示例从头开始创建,但这似乎需要做很多管道工作.

另一个选择是使用Reactive Extensions,因为它似乎旨在使创建observable更容易.我发现最接近使用Rx和MSMQ的是这些页面:

• 使用异步模式(queue.BeginReceive,queue.EndReceive)使用Reactive扩展(Rx)进行MSMQ消息接收
• MSMQ使用Rx - 用于MSMQ的代码片段接收超时问题

但我不确定如何将这些示例应用到我的IRequestListener界面.

任何其他想法也是受欢迎的,如果它们合适,甚至可以改变我的基本设计.

Answer 1

我最初使用FromAsyncPattern,但最后为它编写了一个类,因为它更好地处理了超时和中毒消息.一旦开始,队列无论如何都是热的Observables.你也可以使用Observable.Defer它来使它更接近Rx而不是开始/停止.

这是QueueObservable的基本实现.你可以先打电话来ListenReceive.

Subject<T> Subject = new Subject<T>();

protected void ListenReceive()
{
    Queue.BeginReceive(MessageQueue.InfiniteTimeout, null, OnReceive);
}

protected void OnReceive(IAsyncResult ar)
{
    Message message = null;

    try
    {
        message = Queue.EndReceive(ar);
    }
    catch (TimeoutException ex)
    {
        //retry?
    }

    if (message != null)
        Subject.OnNext((T) message.Body);

    Thread.Yield();

    if (!IsDisposed)
        ListenReceive();
}    

public IObservable<T> AsObservable()
{
        return Subject;
}