system.reactive - 使用异步模式 (queue.BeginReceive,queue.EndReceive) 使用响应式扩展 (Rx) 接收 MSMQ 消息

Question

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

最近我必须修改我的项目的流程，我需要将所有传入数据（来自套接字操作）转储到队列中（使用 MSMQ 实现作为排队决定）。

由于 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();