窗口操作符不能与自身一起工作？

Question

我有一些代码希望以某种方式工作，但事实并非如此，我想知道我做错了什么：

class Program
{
    static void Main(string[] args)
    {
      var ints = Observable.Interval(TimeSpan.FromMilliseconds(1000));
      var windowed = ints.Window(() => ints.Select(i => i / 3).DistinctUntilChanged());

      windowed.Subscribe(HandleNewWindow);

      Console.ReadLine();
    }

    public static void HandleNewWindow(IObservable<long> ints)
    {
      Console.WriteLine("New sequence received");
      ints.Subscribe(Console.WriteLine);
    }
  }

输出应该是：

收到新序列
0
1
2
收到新序列
3
4
5
收到新序列
6
7
8
...

但它是：

收到新序列
0
收到新序列
1
收到新序列
2
收到新序列
3
收到新序列
4
收到新序列
5
收到新序列
6
...

请注意，如果我使用不同的行来定义我的窗口，例如 :

var windowed = ints.Window(() => Observable.Interval(TimeSpan.FromMilliseconds(3000)));

那么一切正常。

Window 在使用从 Observable 派生的窗口关闭（它是窗口）时是否有问题，或者我在这里遗漏了一些重要的东西？

Answer 1

您需要使用Publish运算符来创建一个可以共享源订阅的可观察对象。看起来每次关闭窗口时，它都会在内部设置对源的新订阅。使用发布可确保您不会每次都开始新的间隔。

您还需要将窗口关闭选择器更改为仅在您希望关闭窗口时触发。

class Program
{
    static void Main(string[] args)
    {
        var ints = Observable.Interval(TimeSpan.FromMilliseconds(1000))
            .Publish(new Subject<long>());

        var closeOnValues = ints.Where(ShouldClose);

        var windowed = ints.Window(() => closeOnValues);

        windowed.Subscribe(HandleNewWindow);

        Console.ReadLine();
    }

    public static void HandleNewWindow(IObservable<long> ints)
    {
        Console.WriteLine("New sequence received");
        ints.Subscribe(Console.WriteLine);
    }

    public static bool ShouldClose(long index)
    {
        var notZero = index != 0;
        var countIsMultipleOfThree = (index + 1) % 3 == 0;

        return notZero && countIsMultipleOfThree;
    }
}

Answer 2

我有一些看起来更像我的原始代码并产生预期值的东西。
我仍然不明白为什么这段代码有效，而另一段代码无效，但我认为 James Hay 说得很好，他说某种重新订阅发生在幕后。

class Program
{
    static void Main(string[] args)
    {
        var ints = Observable.Interval(TimeSpan.FromMilliseconds(1000));
        var windowClosings = ints
            .Select(i => i / 3)
            .DistinctUntilChanged()
            .SkipWhile((i) => i == 0)
            .Publish(new Subject<long>());
        var windowed = ints.Window(() => windowClosings);

        windowed.Subscribe(HandleNewWindow);

        Console.ReadLine();
    }

    public static void HandleNewWindow(IObservable<long> ints)
    {
        Console.WriteLine("New sequence received");
        ints.Subscribe(Console.WriteLine);
    }
}

这里的主要区别是，除了仅删除第一个 windowClosing 的 SkipWhile 之外，我必须发布 windowClosings（而不是原始的 Observable）。

仍然不是 100% 确定为什么我必须这样做。