.net - Rx .NET : Filter observable until task is done

Question

我正在学习.NET的Rx，一位同事给我提供了一个简单的示例作为开始，但是我不喜欢某些丑陋的东西。

代码:

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

我可以在输出日志中看到“事件已触发”，但一旦进入source，似乎第一个StartWith(Unit.Default)元素(WithLatestFrom)丢失了。

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合并以确保对于给定索引只有一个请求，然后启动获取。

现在，我们有了一个截然不同的请求流，我们需要订阅和更新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)，因此您可以消除该方法。

所有这些都可以在构造函数中进行，因此您可以在其中隐藏所有状态更改。