合并多个可观察对象并更新现有订阅者？

Question

如何将可能有订阅者也可能没有订阅者的热 Observable 聚合到一个新的 Observable 中，并继续向现有订阅者提供所有新数据？

举个例子，假设我们有这样的 class：


class SomeClass
{

  IObservable<string> Actions { get; set; } = Observable.Empty<string>();

  void AddActionCreator(IObservable<string> creator)
  {
    Actions = Actions.Merge(creator);
  }
}

我运行遇到的问题是，如果 AddActionCreator 添加新的操作流，那么 SomeClass.Actions 的任何先前订阅者在新流合并之前订阅的将永远不会得到新动作。

Answer 1

做自己想做的事很容易。这里你需要的是 SelectMany 和 Subject<IObservabe<string>>.

这是您需要的class：

public class SomeClass
{
    private Subject<IObservable<string>> _sources = new Subject<System.IObservable<string>>();
    public IObservable<string> Actions { get; private set; } = null;

    public SomeClass()
    {
        this.Actions = _sources.SelectMany(x => x);
    }

    public void AddActionCreator(IObservable<string> creator)
    {
        _sources.OnNext(creator);
    }
}

现在你可以这样使用了：

var sc = new SomeClass();
sc.Actions.Subscribe(x => Console.WriteLine($"1:{x}"));
sc.AddActionCreator(Observable.Return("Hello"));
sc.Actions.Subscribe(x => Console.WriteLine($"2:{x}"));
sc.AddActionCreator(Observable.Range(0, 3).Select(x => $"{x}"));
sc.Actions.Subscribe(x => Console.WriteLine($"3:{x}"));
sc.AddActionCreator(Observable.Return("World"));

你会得到这个输出：

1:Hello
1:0
1:1
1:2
2:0
2:1
2:2
1:World
2:World
3:World

您可以看到新的可观察对象已添加到现有订阅者中。

合并多个可观察对象并更新现有订阅者？

Merging multiple observables and updating existing subscribers?

c#

reactive-programming

system.reactive

rx.net

reactive