IObservable <T>を作成する方法は、MSMQメッセージキューから読み取りますか？

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> 
{ 
} 

あなたはそれは私が最もよく合うようだ考えるものであるため、リスナーとプロセッサが観測パターンを使用していることに気づくでしょう。

私の問題は、MSMQから受け取るIRequestListenerの実装を書く方法を考え出しています。基本的にどのように適切なIObservable<T>を作成しますか？

最初に私が見つけたのはMSDN documentationで与えられた例に基づいてをゼロから作成することですが、これは多くの配管作業のようです。

もう1つの選択肢は、オブザーバブルを作成しやすくするように設計されているようなReactive Extensionsを使用することです。

• Using Reactive extension (Rx) for MSMQ message receive using async pattern (queue.BeginReceive,queue.EndReceive)
• MSMQ using Rx - Code snippet for MSMQ receive timeout problem

しかし、私は私のIRequestListenerインターフェイスにこれらの例を適用することができますかどうかはわかりません。私はMSMQでのRxを使用して見つけた最も近いが、これらのページです。

他のアイデアも歓迎します。私の基本的なデザインが適切であれば変更することもできます。

Answer 1

私は最初にFromAsyncPatternを使用しましたが、タイムアウトと被害を受けたメッセージを処理したため、クラスを作成しました。いったん開始されると、キューはとにかく暑いObservablesです。 Observable.Deferを使用して、Start/Stopではなく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; 
}