我有这个代码：
var s1 = new Subject<Unit>();
var s2 = new Subject<Unit>();
var ss = s1.Merge(s2).Finally(() => Console.WriteLine("Finished!"));
ss.Subscribe(_ => Console.WriteLine("Next"));
s1.OnNext(new Unit());
s2.OnNext(new Unit());
s1.OnCompleted(); // I wish ss finished here.
s2.OnCompleted(); // Yet it does so here. =(
我已经使用OnError(new OperationCanceledException())解决了我的问题,但我想要一个更好的解决方案(必须有一个组合器吗？).
解决方法
或者这个,这也很整洁：
public static class Ext
{
public static IObservable<T> MergeWithCompleteOnEither<T>(this IObservable<T> source,IObservable<T> right)
{
return Observable.CreateWithDisposable<T>(obs =>
{
var compositeDisposable = new CompositeDisposable();
var subject = new Subject<T>();
compositeDisposable.Add(subject.Subscribe(obs));
compositeDisposable.Add(source.Subscribe(subject));
compositeDisposable.Add(right.Subscribe(subject));
return compositeDisposable;
});
}
}
这使用了一个主题,它将确保只有一个OnCompleted被推送到CreateWithDisposable()中的观察者;