f# - 基于下游条件的 RX 过滤器

Question

我正在尝试根据下游条件过滤掉更多的上游项目。 mapProcess 本质上启动一个进程(脚本或exe)。该过程可能需要一些时间才能完成，我想在完成之前忽略任何进一步的上游项目。 createProcess 还返回一个 StdOut 的 Observable。我们切换到 createProcess 创建的 IObservable 并将 arg 映射到 StdOut。

例子:

let mapProcess obs =
  obs
  |> Observable.map (fun arg -> createProcess arg)
  |> Observable.switch

我试过的:这有效，但对这里的可变参数并不完全满意。

let mapProcess obs =
  let mutable processNotRunning = true
  obs
  |> Observable.filter (fun _ -> processNotRunning)
  |> Observable.map (fun arg -> processNotRunning <- false
                                createProcess arg)  
  |> Observable.switch
  |> Observable.iter (fun _ -> processNotRunning <- true)
  |> Observable.finallyDo (fun _ -> processNotRunning <- true)

我想我可能需要的是某种“switchIfSeen” Observable 函数，它只会在当前订阅的 observable 产生一个项目或完成时才会切换。通过结合一些现有的 RX 功能，我是否可能错过了一种更简单的方法？

Answer 1

[...] only switch if the currently subscribed observable has produced an item or is finished

忽略值的一种方法是将热 observable 转换为冷 observable - 这样它就可以推出项目而不管谁在听。然后，您只在需要时才听。

  var map = argn.Select(CreateProcess).Publish().RefCount();

        map.SelectMany(o => o) //flatmap
           .Take(1)
           .Repeat()
           .Subscribe(d => Console.WriteLine($"Did task which took {d * 100}msecs" ));

测试:

( CreateProcess 只是一个创建 x100 毫秒固定延迟的计时器。)

    private static void Main(string[] args)
    {
        var argn = Observable.Interval(TimeSpan.FromMilliseconds(100)).Publish().RefCount();
        argn.Subscribe(Console.WriteLine);

        var map = argn.Select(CreateProcess).Publish().RefCount();

        map.SelectMany(o => o)
           .Take(1)
           .Repeat()
           .Subscribe(d => Console.WriteLine($"Did task which took {d * 100}msecs" ));

        Console.ReadKey();
    }

    static IObservable<long> CreateProcess(long i) => Observable.Timer(TimeSpan.FromMilliseconds(i * 100)).Select(_ => i);

输出:

0
Did task which took 0msecs
1
2
Did task which took 100msecs
3
4
5
Did task which took 300msecs
6
7
8
9
10
11
Did task which took 600msecs