如何避免在 Rx 管道中重新运行操作

Question

我有一个 Rx 管道，其中的一些实体是基于传入数据创建的，然后这些实体应该被保留下来，当它们完成时，应该打印一个状态。

我尝试了以下代码。问题是它会多次重新运行该操作，而不仅仅是一次，因此我获得了数据库中实体的多个副本。基本上，使用下面的代码，我只想打印一次"持久测试1"。

        var test1Observable = Observable.Return("test1");
        var test2Observable = Observable.Return("test2");
        var createTest1Observable = test1Observable.Select(name => name);
        var test1 = createTest1Observable.Do(x => Console.WriteLine("persisting test1"));
        var createTest3Observable = Observable.CombineLatest<string, string, Tuple<string,string>>(test1, test2Observable, (t1, t2) => new Tuple<string,string>(t1,t2));
        var persistTest3Observable = createTest3Observable.Do(x => Console.WriteLine("persisting test 3"));
        var createTest4Observable = test1.Select(x => "test4");
        var persistTest4Observable = createTest4Observable.Do(x => Console.WriteLine("persisting test 4"));
        var createTest5Observable = test1.Select(x => "test5");
        var persistTest5Observable = createTest5Observable.Do(x => Console.WriteLine("persisting test5"));
        var OnSuccesObservable = Observable.CombineLatest<string, Tuple<string,string>, string, string, bool>(test1, persistTest3Observable, persistTest4Observable, persistTest5Observable, (t1, t2, t3, t4) =>
        {
            if (t1 == null || t2 == null || t3 == null || t4 == null)
                return false;
            else
                return true;
        });
        OnSuccesObservable.Subscribe(
           o =>
           {
               Console.WriteLine("Finished");
               Console.ReadLine();
           });

Answer 1

尝试将 test1 设为可连接的可观察对象

var test1 = createTest1Observable.Do(x => Console.WriteLine("persisting test1")).Publish();

然后在所有其他代码之后连接到它，如下所示

var connected = test1.Connect();

这可确保正确共享可观察量，并且不会重新订阅