使用响应式扩展(Rx)为MSMQ消息接收使用异步模式(queue.BeginReceive,queue.EndRecei

Question

我已经在我的项目中使用Rx一段时间了，专门用于Socket编程，好的部分是它做得很好。管理我的代码，性能优势和更好的执行和解释。

最近我不得不修改我的项目的流程流，我需要转储所有传入的数据(从套接字操作)到队列(使用MSMQ实现决定排队)。

As MSMQ为从队列中取出消息提供异步调用(但以一种奇怪的模式)。我现在一直在努力使用Rx来实现这个目的，但是可以这样做。

问题:有人能给我一个干净的代码示例来实现Rx的消息接收队列使用异步模式。

我需要MSMQ的异步操作符实现，类似于以下内容

var data = Observable.FromAsyncPattern<byte[]>(
                        this.receiverSocket.BeginReceive,
                        this.receiverSocket.EndReceive(some parameters);

提前感谢。 * 欢呼声* Rx和。net

Answer 1

就像这样简单:

var queue = new System.Messaging.MessageQueue("test");
var fun = Observable.FromAsyncPattern((cb, obj) => queue.BeginReceive(TimeSpan.FromMinutes(10),obj,cb), a => queue.EndReceive(a));
var obs = fun();