我正在编写一个函数，用于检索有关主题的新闻，并通过IObservable返回值将此新闻反馈。
然而，我有几个新闻来源。我不想用Merge把这些源合并成一个。相反，我想做的是按优先级排序——
- 当我的函数被调用时，第一个新闻源被查询(它产生一个IObservable表示该源)。
- 如果新闻源的IObservable没有返回任何结果，则查询下一个新闻源。
- 如果第二个新闻源没有返回结果，则查询最终的新闻源。
- 整个行为被封装成一个可观察对象，我可以返回给用户。
这种行为是我可以完成使用内置的Rx扩展方法，还是我需要实现一个自定义类来处理这个?我该怎么做呢?
在我看来，接受的答案是不可取的，因为它使用Subject, Do，并且在第一个序列不为空时仍然订阅第二个序列。如果第二个可观察对象调用任何重要的东西，后者可能是一个大问题。我想到了下面的解决方案:
public static IObservable<T> SwitchIfEmpty<T>(this IObservable<T> @this, IObservable<T> switchTo)
{
if (@this == null) throw new ArgumentNullException(nameof(@this));
if (switchTo == null) throw new ArgumentNullException(nameof(switchTo));
return Observable.Create<T>(obs =>
{
var source = @this.Replay(1);
var switched = source.Any().SelectMany(any => any ? Observable.Empty<T>() : switchTo);
return new CompositeDisposable(source.Concat(switched).Subscribe(obs), source.Connect());
});
}
名称SwitchIfEmpty与现有的RxJava实现一致。这里正在讨论将一些RxJava操作符合并到RxNET中。
我确信一个自定义的IObservable实现会比我的更有效。你可以在这里找到一个ReactiveX成员akarnokd写的。在NuGet上也可以使用
听起来您可以使用普通的
Amb查询
编辑:根据评论，Amb不会这样做-给这个试试:
public static IObservable<T> SwitchIfEmpty<T>(
this IObservable<T> first,
Func<IObservable<T>> second)
{
return first.IsEmpty().FirstOrDefault() ? second() : first;
}
测试平台:
static Random r = new Random();
public IObservable<string> GetSource(string sourceName)
{
Console.WriteLine("Source {0} invoked", sourceName);
return r.Next(0, 10) < 5
? Observable.Empty<string>()
: Observable.Return("Article from " + sourceName);
}
void Main()
{
var query = GetSource("A")
.SwitchIfEmpty(() => GetSource("B"))
.SwitchIfEmpty(() => GetSource("C"));
using(query.Subscribe(Console.WriteLine))
{
Console.ReadLine();
}
}
一些例子如下:
Source A invoked
Article from A
Source A invoked
Source B invoked
Article from B
Source A invoked
Source B invoked
Source C invoked
Article from C
EDITEDIT:
我想，你也可以这样概括:
public static IObservable<T> SwitchIf<T>(
this IObservable<T> first,
Func<IObservable<T>, IObservable<bool>> predicate,
Func<IObservable<T>> second)
{
return predicate(first).FirstOrDefault()
? second()
: first;
}
另一种方法-与其他方法差别很大，因此我将旋转一个新的答案:
下面是各种有趣的调试行:
public static IObservable<T> FirstWithValues<T>(this IEnumerable<IObservable<T>> sources)
{
return Observable.Create<T>(obs =>
{
// these are neat - if you set it's .Disposable field, and it already
// had one in there, it'll auto-dispose it
SerialDisposable disp = new SerialDisposable();
// this will trigger our exit condition
bool hadValues = false;
// start on the first source (assumed to be in order of importance)
var sourceWalker = sources.GetEnumerator();
sourceWalker.MoveNext();
IObserver<T> checker = null;
checker = Observer.Create<T>(v =>
{
// Hey, we got a value - pass to the "real" observer and note we
// got values on the current source
Console.WriteLine("Got value on source:" + v.ToString());
hadValues = true;
obs.OnNext(v);
},
ex => {
// pass any errors immediately back to the real observer
Console.WriteLine("Error on source, passing to observer");
obs.OnError(ex);
},
() => {
// A source completed; if it generated any values, we're done;
if(hadValues)
{
Console.WriteLine("Source completed, had values, so ending");
obs.OnCompleted();
}
// Otherwise, we need to check the next source in line...
else
{
Console.WriteLine("Source completed, no values, so moving to next source");
sourceWalker.MoveNext();
disp.Disposable = sourceWalker.Current.Subscribe(checker);
}
});
// kick it off by subscribing our..."walker?" to the first source
disp.Disposable = sourceWalker.Current.Subscribe(checker);
return disp.Disposable;
});
}
用法:
var query = new[]
{
Observable.Defer(() => GetSource("A")),
Observable.Defer(() => GetSource("B")),
Observable.Defer(() => GetSource("C")),
}.FirstWithValues();
Source A invoked
Got value on source:Article from A
Article from A
Source completed, had values, so ending
Source A invoked
Source completed, no values, so moving to next source
Source B invoked
Got value on source:Article from B
Article from B
Source completed, had values, so ending
Source A invoked
Source completed, no values, so moving to next source
Source B invoked
Source completed, no values, so moving to next source
Source C invoked
Got value on source:Article from C
Article from C
Source completed, had values, so ending
编辑自原海报:
我选择了这个答案，但是把它变成了一个扩展方法——
/// <summary> Returns the elements of the first sequence, or the values in the second sequence if the first sequence is empty. </summary>
/// <param name="first"> The first sequence. </param>
/// <param name="second"> The second sequence. </param>
/// <typeparam name="T"> The type of elements in the sequence. </typeparam>
/// <returns> The <see cref="IObservable{T}"/> sequence. </returns>
public static IObservable<T> DefaultIfEmpty<T>(this IObservable<T> first, IObservable<T> second)
{
var signal = new AsyncSubject<Unit>();
var source1 = first.Do(item => { signal.OnNext(Unit.Default); signal.OnCompleted(); });
var source2 = second.TakeUntil(signal);
return source1.Concat(source2); // if source2 is cold, it won't invoke it until source1 is completed
}
原始答案:
这可能会奏效。
var signal1 = new AsyncSubject<Unit>();
var signal2 = new AsyncSubject<Unit>();
var source1 = a.Do(item => { signal1.onNext(Unit.Default); signal1.onCompleted(); });
var source2 = b.Do(item => { signal2.onNext(Unit.Default); signal2.onCompleted(); })).TakeUntil(signal1);
var source3 = c.TakeUntil(signal2.Merge(signal1));
return Observable.Concat(source1, source2, source3);
编辑:哎呀，第二个源需要一个单独的信号，第三个源不需要任何信号。Edit2:哎呀…类型。我已经习惯使用RxJs了:)
注:也有更少的RX-y方法，这可能需要更少的输入:
var gotResult = false;
var source1 = a();
var source2 = Observable.Defer(() => return gotResult ? Observable.Empty<T>() : b());
var source3 = Observable.Defer(() => return gotResult ? Observable.Empty<T>() : c());
return Observable.Concat(source1, source2, source3).Do(_ => gotResult = true;);
这是JerKimball的SwitchIfEmpty操作符的非阻塞版本。
/// <summary>Returns the elements of the first sequence, or the elements of the
/// second sequence if the first sequence is empty.</summary>
public static IObservable<T> SwitchIfEmpty<T>(this IObservable<T> first,
IObservable<T> second)
{
return Observable.Defer(() =>
{
bool isEmpty = true;
return first
.Do(_ => isEmpty = false)
.Concat(Observable.If(() => isEmpty, second));
});
}
下面是同一操作符的一个版本，它接受多个序列，并返回第一个非空序列的元素:
/// <summary>Returns the elements of the first non-empty sequence.</summary>
public static IObservable<T> SwitchIfEmpty<T>(params IObservable<T>[] sequences)
{
return Observable.Defer(() =>
{
bool isEmpty = true;
return sequences
.Select(s => s.Do(_ => isEmpty = false))
.Select(s => Observable.If(() => isEmpty, s))
.Concat();
});
}
Observable.Defer操作符用于防止多个订阅共享相同的bool isEmpty状态(更多信息在这里)。