听起来您想要这样的数据流程:
seed profiles --> source --> get related --> output
^ |
| v
-<--- transform <-----
这似乎是一个解决通用问题比解决特定问题更容易的情况,因此我将建议提供一个通用的“反馈”函数,应该能够给你需要的构建块:
编辑:修复了函数以使其完整
IObservable<TResult> Feedback<T, TResult>(this IObservable<T> seed,
Func<T, IObservable<TResult>> produce,
Func<TResult, IObservable<T>> feed)
{
return Observable.Create<TResult>(
obs =>
{
var ret = new CompositeDisposable();
Action<IDisposable> partComplete =
d =>
{
ret.Remove(d);
if (ret.Count == 0) obs.OnCompleted();
};
Action<IObservable<T>, Action<T>> ssub =
(o, n) =>
{
var disp = new SingleAssignmentDisposable();
ret.Add(disp);
disp.Disposable = o.Subscribe(n, obs.OnError, () => partComplete(disp));
};
Action<IObservable<TResult>, Action<TResult>> rsub =
(o, n) =>
{
var disp = new SingleAssignmentDisposable();
ret.Add(disp);
disp.Disposable = o.Subscribe(n, obs.OnError, () => partComplete(disp));
};
Action<T> recurse = null;
recurse = s =>
{
rsub(produce(s),
r =>
{
obs.OnNext(r);
ssub(feed(r), recurse);
});
};
ssub(seed, recurse);
return ret;
});
}
在你的情况中,
T
和
TResult
看起来是一样的,因此
feed
将成为恒等函数。
produce
将用于实现步骤2和3的函数。
以下是我使用该功能测试的一些示例代码:
void Main()
{
var seed = new int[] { 1, 2, 3, 4, 5, 6 };
var found = new HashSet<int>();
var mults = seed.ToObservable()
.Feedback(i =>
{
return Observable.Range(0, 5)
.Select(r => r * i)
.TakeWhile(v => v < 100)
.Where(v => found.Add(v));
},
i => Observable.Return(i));
using (var disp = mults.Dump())
{
Console.WriteLine("Press any key to stop");
Console.ReadKey();
}
Console.WriteLine("Press any key to exit");
Console.ReadKey();
}
static IDisposable Dump<T>(this IObservable<T> source)
{
return source.Subscribe(item => Console.WriteLine(item),
ex => Console.WriteLine("Error occurred in dump observable: " + ex.ToString()),
() => Console.WriteLine("Dump completed"));
}