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

Question

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

例如，假设我们有这样的类：

class SomeClass
{

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

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

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

标签： c#reactive-programmingsystem.reactivereactiverx.net

Answer 1

做你想做的事很容易。你在这里需要的是 aSelectMany和 a Subject<IObservabe<string>>。

这是您需要的课程：

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：你好
1:0
1:1
1:2
2:0
2:1
2:2
1：世界
2：世界
3：世界

您可以看到新的 observables 已添加到现有订阅者中。