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

问题描述

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

I have been using Rx for a while now for Events on my projects and dedicatedly for Socket programming and the good part is its doing well. Managing my code, performance advantage and much better to execute and interpret.

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

Lately I have to modify my project's process flow where i need to dump all the incoming data (from socket operations) into queues (using MSMQ implementation as decided for queueing).

因为 MSMQ 提供了异步调用以从队列中取出消息(但以一种奇怪的模式).我现在一直在努力将 Rx 用于此目的，但已启用.

As MSMQ provides async call for dequeing messages from the queue (but in an wierd pattern).I have been struggling to use Rx for this purpose now, but enable to do so.

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

Question : Can some one give me a clean code example to implement Rx for message receiving from queue using Async pattern.

我需要类似于这样的 MSMQ 异步操作符实现

I need the async operator implementation for MSMQ analogous to something like this

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

提前致谢.*为 Rx 和 .NET 干杯*

Thanks in advance. *cheers* to Rx and .NET

推荐答案

就这么简单:

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

这篇关于使用异步模式 (queue.BeginReceive,queue.EndReceive) 为 MSMQ 消息接收使用反应式扩展 (Rx)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！