どのようにObservable.ToAsyncをIEnumerableで使用する

Question

Rxで問題が発生しました。私は、新しいスレッドの各項目を処理した後、結果をメインスレッドに送る必要があります。私は他の方法でそれをやった。 Rxでこのタスクをどのように解決できますか？ここにコードは

Observable.ToAsync<string, IEnumerable<UserState>>(Run)(path) 
.ObserveOnDispatcher<IEnumerable<UserState>>() 
.Subscribe(
(o) => 
{ // need to run in Main Thread 
    foreach (var item in o) 
    { 
    WriteLog(item.StatusCode, item.Url); 
    }       
}, 
(ex) =>{  }, 
() =>{  }); 

    // need to run in New Thread 
    private IEnumerable<UserState> Run(string sitemap) 
    { 
.... 
foreach (var url in urls) 
{ 
    var result = new UserState 
    { 
     Url = url.Value, 
     TotalItems = urls.Count 
    }; 
    .... 
    yield return result; 
} 
    } 

Answer 1

あなたが他のいくつかのバックグラウンドスレッドでのIEnumerableを生成し、この理解は、あなたがこのような何かを行うことができます正しいかどうか、各ユーザーは、メインUIスレッド上で、この列挙でオブジェクトを処理する：

var users = Run(path); //NOTE: This doesn't execute your run method yet, Run will only execute when you start enumerating the users values 
users.ToObservable(System.Concurrency.Scheduler.ThreadPool) //The enumerator will be scheduled on separate thread 
.ObserveOn(frm) //Observe on UI thread of win form 
.Subscribe(s => {}) //This will run in UI thread for each user object 

Answer 2

あなたが持っている問題を説明することができればよいです。私は現時点ではあなたを伝えることができますどのような

は、あなたが潜在的な問題のカップルを持っているということです。 ObserveOnDispatcherを使用してまず

は、（通常、WPFを使用して作成）System.Windows.Threading.Dispatcherで動作するように意図されています。 WPFの外部でコードを実行している場合は、「現在のスレッド」を意味し、現在のスレッドがビジー状態の場合、サブスクリプションを実行できなくなる可能性があります。つまり、デッドロックを作成する可能性があります。

私は、WPFとLINQPadの両方でコードを実行し、それがWPFに罰金働いたが、LINQPadでデッドロック。私が別のスレッドで観察した場合、LINQPadで正常に動作し、WPFで失敗しました。

第二に、あなたは、観察非同期にiteratorメソッドを回していると、あなたが期待するよう、それは動作しません。反復子は、実際には列挙型を反復処理するまで、実際にはコードを実行しません。基本的に、あなたはほぼ瞬時にRunから復帰し、あなただけのSubscribeコードであなたの体Runメソッドを実行 - それは間違っているスレッドです！あなたがする必要がどのような

可算の力即時実行である - 非常に少なくとも、 - このように見えるようにコードを変更します。

private UserState[] Run(string sitemap) 
{ 
    ... 
    Func</* url type */, UserState> create = url => 
    { 
     var result = new UserState 
     { 
      Url = url.Value, 
      TotalItems = urls.Count 
     }; 
     .... 
     return result; 
    }; 
    return (from url in urls select create(url)).ToArray(); 
} 

あなたの主なコードは少しクリーンアップ持っている必要があります：

Observable.ToAsync<string, UserState[]>(Run)(path) 
    .ObserveOnDispatcher() 
    .Subscribe(o => 
    { 
     foreach (var item in o) 
     { 
      WriteLog(item.StatusCode, item.Url); 
     }       
    }); 

は、このいずれかが助けなら、私に教えてください。

---

EDIT：コメントにOPリクエストごとにサンプルFromEventPatternを追加しました。

ここでは、Windowsフォームの使用例をFromEventPatternとしています。最初の部分は、フォームが終了したときにサブスクリプションをクリーンアップする方法を作成します。

public partial class Form1 : Form 
{ 
    public Form1() 
    { 
     InitializeComponent(); 

     // Create a collection of IDisposable 
     // to allow clean-up of subscriptions 
     var subscriptions = 
      new System.Reactive.Disposables.CompositeDisposable(); 

     var formClosings = Observable 
      .FromEventPattern<FormClosingEventHandler, FormClosingEventArgs>(
       h => this.FormClosing += h, 
       h => this.FormClosing -= h); 

     // Add a subscription that cleans up subscriptions 
     // when the form closes 
     subscriptions.Add(
      formClosings 
       .Subscribe(ea => subscriptions.Dispose())); 

この次の部分では、画像ボックスのマウスドラッグを監視し、ドラッグした距離をユーザーに知らせるメッセージを作成します。

 var pictureBox1MouseDowns = Observable 
      .FromEventPattern<MouseEventHandler, MouseEventArgs>(
       h => pictureBox1.MouseDown += h, 
       h => pictureBox1.MouseDown -= h); 

     var pictureBox1MouseMoves = Observable 
      .FromEventPattern<MouseEventHandler, MouseEventArgs>(
       h => pictureBox1.MouseMove += h, 
       h => pictureBox1.MouseMove -= h); 

     var pictureBox1MouseUps = Observable 
      .FromEventPattern<MouseEventHandler, MouseEventArgs>(
       h => pictureBox1.MouseUp += h, 
       h => pictureBox1.MouseUp -= h); 

     var pictureBox1MouseDrags = 
      from md in pictureBox1MouseDowns 
      from mm in pictureBox1MouseMoves.TakeUntil(pictureBox1MouseUps) 
      let dx = mm.EventArgs.Location.X - md.EventArgs.Location.X 
      let dy = mm.EventArgs.Location.Y - md.EventArgs.Location.Y 
      select new Point(dx, dy); 

     var pictureBox1MouseDragMessages = 
      from md in pictureBox1MouseDrags 
      let f = "You've dragged ({0}, {1}) from your starting point" 
      select String.Format(f, md.X, md.Y); 

次の部分は、ボタンがクリックされた回数を追跡し、ユーザーに表示するメッセージを作成します。

 var button1ClickCount = 0; 

     var button1Clicks = Observable 
      .FromEventPattern(
       h => button1.Click += h, 
       h => button1.Click -= h); 

     var button1ClickCounts = 
      from c in button1Clicks 
      select ++button1ClickCount; 

     var button1ClickMessages = 
      from cc in button1ClickCounts 
      let f = "You clicked the button {0} time{1}" 
      select String.Format(f, cc, cc == 1 ? "" : "s"); 

最後に、2つのメッセージobervableがマージされ、サブスクライブされ、メッセージをラベルに入れます。

 var messages = pictureBox1MouseDragMessages 
      .Merge(button1ClickMessages); 

     // Add a subscription to display the 
     // merged messages in the label 
     subscriptions.Add(
      messages 
       .Subscribe(m => label1.Text = m)); 
    } 
} 

このすべては、フォームのコンストラクタに常駐し、何のモジュールレベルのフィールドやプロパティが使用されていないと、フォームが閉じたときに、すべてのイベント・ハンドラが削除されていることを覚えておいてください。非常にきちんとしたもの。