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

Question

我有一个 Rx 管道，其中一些实体是根据传入数据创建的，然后应保留这些实体，并在它们完成时打印状态。

我试过下面的代码。问题是它多次重新运行该操作，而不是一次，所以我在数据库中获得了多个实体副本。基本上使用下面的代码我只想打印一次 "persisting test1"。

        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();

这确保了 observable 被正确共享而不是重新订阅

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

How to avoid rerunning actions in Rx pipeline

.net

c#

reactive-programming

system.reactive