ストリームのリスナーをIObservableに変換する

Question

私のサービスメソッドはリスナーを取り、それをデータストリームで呼び出す。このストリームをIObservable<T>に変換しようとしています。これまでのところ、これは私がやったことです：

public class MessageListener : IMessageListener 
{ 

    private readonly Subject<string> stream = new Subject<string>(); 

    public IObservable<string> MessageStream 
    { 
     get 
     { 
      return this.stream; 
     } 
    } 

    public void OnMessageAdded(string message) 
    { 
     this.stream.OnNext(message); 
    } 

} 

//Calling code  
public IObservable<string> GetMessage() 
{ 
    var listener = new MessageListener(); 
    service.Subscribe(listener); 
    listener.MessageStream.SubscribeOn(Scheduler.NewThread); 
} 

私はこれで十分ですか分かりません。私はSubscribeOnへの呼び出しを信じて、新しいスレッドでサブスクリプションコードだけを実行します。 OnMessageAddedが新しいスレッドによって受信されたことをどうやって確認しますか？

Answer 1

あなた自身が既に発見したように、SubscribeOnはサブスクリプションが起こる場所のみを制御します。しかし、あなたのコードではSubscribeOnはサブスクリプションがないので、何の効果もありません。 SubscribeOnは、サブスクリプションに使用するObservableを返します。 "SubscribeOn"を呼び出してソースの中に何らかのグローバルフラグを設定することはできません。

あなたが実際にやりたいことは、サブスクリプションが行われる前にサービス内の "ObserveOn"に電話することです。 ObserveOnは、メッセージを処理するスレッドを定義します。

public IObservable<string> MessageStream 
{ 
    get 
    { 
     return this.stream.ObserveOn(Scheduler.ThreadPool).AsObservable(); 
    } 
} 

もう一つ

が指摘されるように、あなたのMessageStreamプロパティに、より良い被写体を直接返すのではなく this.stream.AsObservable()を呼び出します。それは次のようになりますように、別のオプションは、MessageListenerのに直接 ObserveOnを書くことであろう。