Uso de la extensión reactiva (Rx) para recibir mensajes MSMQ utilizando un patrón asíncrono (queue.BeginReceive, queue.EndReceive)

Question

He estado utilizando Rx por un tiempo ahora para eventos en mis proyectos y dedicado a la programación de Socket y la parte buena es su haciendo bien. Administrar mi código, ventaja de rendimiento y mucho mejor para ejecutar e interpretar.

Últimamente tengo que modificar el flujo de procesos de mi proyecto donde necesito volcar todos los datos entrantes (desde operaciones de socket) en colas (usando la implementación MSMQ como se decidió para hacer cola).

Como MSMQ proporciona una llamada asincrónica para eliminar mensajes de la cola (pero en un patrón extraño). He estado luchando para usar Rx para este propósito ahora, pero lo habilito para hacerlo.

Pregunta: ¿Alguien puede darme un ejemplo de código limpio para implementar Rx para la recepción de mensajes de la cola utilizando el patrón Async.

necesito la ejecución del operador asíncrono para MSMQ análoga a algo como esto

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

Gracias de antemano. * aplausos * a Rx y .NET

Answer 1

Sería tan simple como:

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