我想合并2个可观测值并保留订单(可能基于选择器(。我也想在可观察的源头上放置压力。
因此,选择器将选择一个项目之一,以通过可观察到的可观察到,而另一个项目也会等待另一个项目来比较。
src1,src2,结果都是IObservable<T>
类型。
Src1: { 1,3,6,8,9,10 }
Src2: { 2,4,5,7,11,12 }
Result: 1,2,3,4,5,6,7,8,9,10,11,12
Timeline:
Src1: -1---3----6------8----9-10
Src2: --2-----4---5-7----11---------12
Result: --1--2--3-4-5-6--7-8--9-10-11-12
- 在上面的示例中,src1排放了1',并被阻塞直到src2散发这是第一个项目," 2"。
- 选择最小项目的选择器从src1中选择项目。
- src2现在等待下一个项目(从src1(与之比较当前项目('2'(。
- 当Src1排放下一个项目'3'时,选择再次运行,此从SRC2选择项目的时间。
- 这重复直到一个可观察到的东西完成为止。然后,剩下的可观察到的推动物品直到完成为止。
这是有可能通过现有的.NET RX方法实现的?
编辑:请注意,保证2个来源可观察到井井有条。
示例测试:
var source1 = new List<int>() { 1, 4, 6, 7, 8, 10, 14 }.AsEnumerable();
var source2 = new List<int>() { 2, 3, 5, 9, 11, 12, 13, 15 }.AsEnumerable();
var src1 = source1.ToObservable();
var src2 = source2.ToObservable();
var res = src1.SortedMerge(src2, (a, b) =>
{
if (a <= b)
return a;
else
return b;
});
res.Subscribe((x) => Console.Write($"{x}, "));
desiredresult:1,2,3,4,5,6,7,8,9,10,11,12,13,14,15
这很有趣。不得不对算法进行一些调整。它可以进一步改进。
假设:
- 有两个流,
streamA
,streamB
的CC_3。 - 分别对两个流进行分类,使得
streamA[i] < streamA[i+1]
和streamB[i] < stream[i+1]
。 - 您不能假设
streamA[i]
和streamB[i]
之间的任何关系。 - 流A&amp;b是谨慎的:同一元素不会从两者中散发出来。如果发生这种情况,我将扔
NotImplementedException
。这个情况很容易处理,但我想避免歧义。 - 类型
T
有一个函数CC_11。 - 没有关于这两个流的相对速度的假设,但是如果一个流的速度始终比另一个速度快,则背压将是一个问题。
这是我使用的算法:
- 让两个队列
qA
和qB
。 - 当您从
streamA
获得项目时,将其加入qA
。 - 当您从
streamB
获取项目时,将其加入qB
。 - 虽然
qA
和qB
中都有一个项目,但比较了两个队列的顶部项目。删除并发出这两个的最小值。如果两个队列仍然非空,请重复。 - 如果
streamA
或streamB
完成,请转储队列的内容并终止。 Note :这是懒惰的,可能应该更改为转储,然后继续返回未完成的可观察。
这是代码:
public static IObservable<T> SortedMerge<T>(this IObservable<T> source, IObservable<T> other)
{
return SortedMerge(source, other, (a, b) => Enumerable.Min(new[] { a, b}));
}
public static IObservable<T> SortedMerge<T>(this IObservable<T> source, IObservable<T> other, Func<T, T, T> min)
{
return source
.Select(i => (key: 1, value: i)).Materialize()
.Merge(other.Select(i => (key: 2, value: i)).Materialize())
.Scan((qA: ImmutableQueue<T>.Empty, qB: ImmutableQueue<T>.Empty, exception: (Exception)null, outputMessages: new List<T>()),
(state, message) =>
{
if (message.Kind == NotificationKind.OnNext)
{
var key = message.Value.key;
var value = message.Value.value;
var qA = state.qA;
var qB = state.qB;
if (key == 1)
qA = qA.Enqueue(value);
else
qB = qB.Enqueue(value);
var output = new List<T>();
while(!qA.IsEmpty && !qB.IsEmpty)
{
var aVal = qA.Peek();
var bVal = qB.Peek();
var minVal = min(aVal, bVal);
if(aVal.Equals(minVal) && bVal.Equals(minVal))
throw new NotImplementedException();
if(aVal.Equals(minVal))
{
output.Add(aVal);
qA = qA.Dequeue();
}
else
{
output.Add(bVal);
qB = qB.Dequeue();
}
}
return (qA, qB, null, output);
}
else if (message.Kind == NotificationKind.OnError)
{
return (state.qA, state.qB, message.Exception, new List<T>());
}
else //message.Kind == NotificationKind.OnCompleted
{
var output = state.qA.Concat(state.qB).ToList();
return (ImmutableQueue<T>.Empty, ImmutableQueue<T>.Empty, null, output);
}
})
.Publish(tuples => Observable.Merge(
tuples
.Where(t => t.outputMessages.Any() && (!t.qA.IsEmpty || !t.qB.IsEmpty))
.SelectMany(t => t.outputMessages
.Select(v => Notification.CreateOnNext<T>(v))
.ToObservable()
),
tuples
.Where(t => t.outputMessages.Any() && t.qA.IsEmpty && t.qB.IsEmpty)
.SelectMany(t => t.outputMessages
.Select(v => Notification.CreateOnNext<T>(v))
.ToObservable()
.Concat(Observable.Return(Notification.CreateOnCompleted<T>()))
),
tuples
.Where(t => t.exception != null)
.Select(t => Notification.CreateOnError<T>(t.exception))
))
.Dematerialize();
ImmutableQueue
来自System.Collections.Immutable
。需要Scan
来跟踪状态。由于OnCompleted
处理,需要实现化。诚然,这是一个复杂的解决方案,但我不确定是否有以Rx为中心的方式。
让我知道您是否需要更多澄清。