c# - 如何使用 Rx 监控文件的变化？

Question

我正在使用 Rx 来监视对文件的更改。

         |  window1   | |  window2   | | window3 |
  Time   01 02 03 04 05 06 07 08 09 10 11 12 13 14 15

 Input       M  M     M     M     M              R
Output                M              M           R

 Input       M  M     M     M     M        M     R
Output                M              M     M     R         

(M - 修改，R - 重命名)

顶级示例:最后一个窗口 (11 - 14) 中没有剩余修改。

底部示例:最后一个窗口中的剩余修改(在 t = 12)。

在上面的示例中，我每 5 秒采样一次以限制对文件的修改操作。即在每 5 秒的窗口中，我只进行最新的修改操作。

我希望在重命名操作发生时立即停止采样，并且返回当前窗口中的最新修改(如果有)和重命名事件。

一些注意事项:

1. 最后一个窗口不需要 5 秒长。
2. 最后一个窗口可以返回多个事件。

这是我目前所拥有的。

var events = source
            .GroupBy(e => e.FullPath)
            .SelectMany(g => g.Sample(TimeSpan.FromSeconds(5)));

Answer 1

这是一个解决方案 - 有点仓促，所以我确信它可以整理，但它似乎有效。这使用 Nuget 包 rx-testing和 ix-main [原文如此 - IEnumerable<T>.TakeLast() 需要].我已经将您的示例映射为所展示的，但将一个时间单位表示为 100 个刻度。

请注意，您问题中的第二个场景有一个错误 - 在收到重命名事件之前不可能发出最后修改的事件。因此，我的输出在 1400 ticks 时发出了两个事件。

主要查询如下(假设actions包含FileAction事件):

发布操作以便我们可以多次订阅:

actions.Publish(source => {

以 500 个滴答间隔对操作进行窗口化，但在重命名时将窗口关闭:

  return source.Window(
    () => Observable.Timer(TimeSpan.FromTicks(500), scheduler)
                    .Merge(source.Where(x => x == FileAction.Rename)
                                 .Select(_ => 0L)))

在重命名时也停止窗口流:

    .TakeUntil(source.Where(x => x == FileAction.Rename))

将窗口展平成包含最多两个事件的列表流:

    .SelectMany(x => x.TakeLastBuffer(2))

寻找重命名作为最后一项，如果是重命名则抓取最后 2 项，否则只是最后一项。请注意，这是精心设计的，用于处理空列表或 1 项列表。另外，使用 SelectMany再次将列表展平，以便我们留下一系列操作。

    .SelectMany(l => {
      return l.LastOrDefault() == FileAction.Rename
        ? l.TakeLast(2)
        : l.TakeLast(1);
    });
})

您可以订阅它以获取请求的事件。这是一个完整的示例:

void Main()
{
    var tests = new Tests();
    tests.Scenario();
}

// Note inheritance of ReactiveTest to get OnXXX helpers
public class Tests : ReactiveTest
{
    public enum FileAction
    {
        Modify,
        Rename
    }

    public void Scenario()
    {
        var scheduler = new TestScheduler();
        var actions = scheduler.CreateHotObservable<FileAction>(
            OnNext(200, FileAction.Modify),
            OnNext(300, FileAction.Modify),
            OnNext(500, FileAction.Modify),
            OnNext(700, FileAction.Modify),
            OnNext(900, FileAction.Modify),
            // Uncomment next line for
            // scenario with mod + rename in last window
            // OnNext(1200, FileAction.Modify),
            OnNext(1400, FileAction.Rename));

        actions.Publish(source => {
            return source.Window(
                () => Observable.Timer(TimeSpan.FromTicks(500), scheduler)
                                .Merge(source.Where(x => x == FileAction.Rename)
                                            .Select(_ => 0L)))
                .TakeUntil(source.Where(x => x == FileAction.Rename))
                .SelectMany(x => x.TakeLastBuffer(2))
                .SelectMany(l => {
                    return l.LastOrDefault() == FileAction.Rename
                        ? l.TakeLast(2)
                        : l.TakeLast(1);
                });
        })
        .Timestamp(scheduler) /* to show timings */
        .Subscribe(x => Console.WriteLine(
            "Event: {0} Timestamp: {1}",
            x.Value, x.Timestamp.Ticks));

        scheduler.Start();

    }
}

这个输出:

Event: Modify Timestamp: 500
Event: Modify Timestamp: 1000
Event: Rename Timestamp: 1400

如果您取消注释第二种情况的标记行，您将获得:

Event: Modify Timestamp: 500
Event: Modify Timestamp: 1000
Event: Modify Timestamp: 1400
Event: Rename Timestamp: 1400