c# - 响应式扩展 .MaxBy

Question

为什么

var a = Observable.Interval(TimeSpan.FromSeconds(1))
    .Publish();

a.Subscribe(o =>
{
    Console.WriteLine("Test");
});
a.Connect();

火，但不火

var a = Observable.Interval(TimeSpan.FromSeconds(1))
    .MaxBy(o=>o)
    .Publish();

a.Subscribe(o =>
{
    Console.WriteLine("Test");
});
a.Connect();

我正在尝试在不同的场景中使用 MaxBy，但甚至无法使上述工作正常。

这是我更复杂的例子

var _telemetryBatchObservable = Observable.FromEventPattern<DeviceStateStreamEventArg>(
        ev => DeviceStateStreamEvent += ev,
        ev => DeviceStateStreamEvent -= ev)
    .Synchronize()
    .GroupBy(o => o.EventArgs.DeviceId)
    .Select(o => o.MaxBy(i => i.EventArgs.DateTimeOffset))
    .SelectMany(o => o.Select(i => i))
    .SelectMany(o => o.Select(i => i))
    .Buffer(TimeSpan.FromMilliseconds(5000), 100)
    .Publish();

Answer 1

Max 和 MaxBy 仅在源可观察对象终止时发出单个值。如果你有一个非终止源，它们将永远不会发射。

尝试将此代码作为反例:

var a = Observable.Interval(TimeSpan.FromSeconds(1))
    .Take(3) //causes termination after 3 values
    .MaxBy(o => o)
    .Publish();