主題に関するニュースを取得し、IObservable の戻り値を介してこのニュースをフィードバックする関数を作成しています。
ただし、いくつかのニュースソースがあります。Mergeこれらのソースを 1 つに結合するために使用したくありません。代わりに、私がやりたいのは、それらを優先順位で並べることです -
この種の動作は、組み込みの Rx 拡張メソッドを使用して達成できるものですか?それとも、これを処理するためにカスタム クラスを実装する必要がありますか? どちらかを行うにはどうすればよいですか？
私はこの答えに行きましたが、これを拡張メソッドに変えました -
/// <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);
編集：おっと、2番目のソースには別の信号が必要で、3番目のソースには何も信号を送る必要はありません。Edit2: おっと...タイプ。私はRxJに慣れています:)
PSまた、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;);
単純な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
編集編集:
これを一般化することもできます。
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