是否有更简单的方法让 IObservable 异步依赖于另一个 IObservable？

Question

我是 RX 的新手，我想要的场景运行良好，但在我看来必须有一种更简单或更优雅的方法来实现这一点。我拥有的是 IObservable，我想通过触发异步操作来订阅它，最终得到 IObservable,它为它看到的每个 T 生成一个 U。

到目前为止，我所拥有的（效果很好，但看起来很麻烦）使用中间事件流并进行如下操作：

public class Converter {
  public event EventHandler<UArgs> UDone;
  public IConnectableObservable<U> ToUs(IObservable<T> ts) {
    var us = Observable.FromEvent<UArgs>(this, "UDone").Select(e => e.EventArgs.U).Replay();
    ts.Subscribe(t => Observable.Start(() => OnUDone(new U(t))));
    return us;
  }
  private void OnUDone(U u) {
    var uDone = UDone;
    if (uDone != null) {
      uDone(this, u);
    }
  }
}

...

var c = new Converter();
IConnectableObservable<T> ts = ...;
var us = c.ToUs(ts);
us.Connect();

...

我确信我缺少一种更简单的方法来执行此操作......

Answer 1

SelectMany 应该执行您需要的操作，以展平 IO>

Observable.Range(1, 10)
        .Select(ii => Observable.Start(() => 
             string.Format("{0} {1}", ii, Thread.CurrentThread.ManagedThreadId)))
        .SelectMany(id=>id)
        .Subscribe(Console.WriteLine);

Answer 2

这正是 SelectMany 的用途：

IObservable<int> ts

IObservable<string> us = ts.SelectMany(t => StartAsync(t));

us.Subscribe(u => 
    Console.WriteLine("StartAsync completed with {0}", u));

...

private IObservable<string> StartAsync(int t)
{
    return Observable.Return(t.ToString())
        .Delay(TimeSpan.FromSeconds(1));
}

请记住，如果 StartAsync 具有可变的完成时间，您可能会以与输入值不同的顺序接收输出值。