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

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

Answer 1

这看起来像是某种分页自动滚动实现？从概念上讲，它可以帮助拆分您的 observable：

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），因此您可以取消该方法。

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