Rx .NET：タスクが完了するまで観測可能なフィルタ

Question

私はRx for .NETを学習しています。同僚は私に始めて簡単な例を送っていますが、醜いものがあります。

コード：

using System; 
using System.Reactive.Linq; 
using System.Reactive.Threading.Tasks; 
using System.Threading.Tasks; 
using System.Windows.Forms; 
using System.Collections.Generic; 

namespace WindowsFormsApplication1 
{ 
    public partial class Form1 : Form 
    { 
     public IObservable<Content> contentStream; 
     public static bool isRunning = false; 

     public Form1() 
     { 

      InitializeComponent(); 

      contentStream = Observable.FromEventPattern<ScrollEventArgs>(dataGridView1, "Scroll") // create scroll event observable 
       .Where(e => (dataGridView1.Rows.Count - e.EventArgs.NewValue < 50 && !isRunning)) //discart event if scroll is not down enough 
       //or we are already retrieving items (isRunning) 
       .Select(e => { isRunning = true; return 100; }) //transform to 100--100--100--> stream, discart next events until we finish 
       .Scan((x, y) => x + y) //get item index by accumulating stream items 
       .StartWith(0) //start with 0 before event gets triggered 
       .SelectMany(i => getContent(i).ToObservable());//create a stream with the result of an async function and merge them into just one stream 

      contentStream.Subscribe(c => invokeUpdateList(c)); //just update the control every time a item is in the contentStream 

     } 

     async private Task<Content> getContent(int index) 
     { 

      await Task.Delay(1000);//request to a web api... 
      return new Content(index);//mock the response 
     } 

     private void invokeUpdateList(Content c) 
     { 
      dataGridView1.Invoke((MethodInvoker)delegate 
      { 
       updateList(c); 
      }); 
     } 

     private void updateList(Content c) 
     { 
      foreach (var item in c.pageContent) 
      { 
       dataGridView1.Rows.Add(item); 
      } 
      isRunning = false; //unlocks event filter 
     } 

    } 

    public class Content 
    { 
     public List<string> pageContent = new List<string>(); 
     public const string content_template = "This is the item {0}."; 
     public Content() 
     { 
     } 
     public Content(int index) 
     { 

      for (int i = index; i < index + 100; i++) 
      { 
       pageContent.Add(string.Format(content_template, i)); 
      } 

     } 
    } 
} 

私は好きではない何がisRunningフィルタです。コントロールが更新されるまで、ストリーム内のいくつかのイベントを切り捨てるより良い方法はありますか？ @Shlomoアプローチは右のようですが

それが負荷に移入起動しない：

var index = new BehaviorSubject<int>(0); 

     var source = Observable.FromEventPattern<ScrollEventArgs>(dataGridView2, "Scroll") 
      .Where(e => dataGridView2.Rows.Count - e.EventArgs.NewValue < 50) 
      .Select(_ => Unit.Default) 
      .StartWith(Unit.Default) 
      .Do(i => Console.WriteLine("Event triggered")); 

     var fetchStream = source 
      .WithLatestFrom(index, (u, i) => new {unit = u,index = i }) 
      .Do(o => Console.WriteLine("Merge result" + o.unit + o.index)) 
      .DistinctUntilChanged() 
      .Do(o => Console.WriteLine("Merge changed" + o.unit + o.index)) 
      .SelectMany(i => getContent(i.index).ToObservable()); 

     var contentStream = fetchStream.WithLatestFrom(index, (c, i) => new { Content = c, Index = i }) 
      .ObserveOn(dataGridView2) 
      .Subscribe(a => 
      { 
      updateGrid(a.Content); 
      index.OnNext(a.Index + 100); 
      }); 

私が見ることができ、出力ログに「イベントをトリガー」が、私はWithLatestFromに達すると最初source要素（StartWith(Unit.Default)）が失われているようです。

Answer 1

これは、ページ分割自動スクロールの実装のようなものですか？概念的には、それはあなたの、観察を分割するのに役立つ可能性があり：

var index = new BehaviorSubject<int>(0); 

var source = Observable.FromEventPattern<ScrollEventArgs>(dataGridView1, "Scroll") 
    .Where(e => dataGridView1.Rows.Count - e.EventArgs.NewValue < 50) 
    .Select(_ => Unit.Default) 
    .StartWith(Unit.Default); 

var fetchStream = source 
    .WithLatestFrom(index, (_, i) => i) 
    .DistinctUntilChanged() 
    .SelectMany(i => getContent(i).ToObservable()); 

だからsourceは、ユーザーがリストの更新を開始することを望んでいる一連のユニット、基本的に空の通知です。 indexはダウンロードする次のインデックスを表します。 fetchstreamはsourceとindexをマージして、特定のインデックスに対する要求が1つしかないことを確認してから、フェッチを開始します。

リクエストのストリームが異なるので、UIを登録して更新する必要があります。index。

var contentStream = 
    fetchStream .WithLatestFrom(index, (c, i) => new { Content = c, Index = i }) 
    .ObserveOn(dataGridView1) 
    .Subscribe(a => 
     { 
      updateList(a.Content); 
      index.OnNext(a.Index + 100); 
     }); 

注ObserveOn(datagridView1)はあなたInvokeUpdateList方法と同じことを達成したがクリーナー形で（Nuget System.Reactive.Windows.Formsが必要）ので、あなたはその方法を排除することができます。

これはすべてコンストラクタに入れることができるので、そこにあるすべての状態の変更を非表示にすることができます。