介绍
这个问题的关键是排序。不管你怎么看，都需要某种形式的缓冲。虽然毫无疑问，一些精心设计的运算符组合可能会实现这一点，但我认为这是一个很好的例子，Observable.Create它是一个很好的选择。
推广解决方案
我已经做出了一些努力来概括我接受任何类型的排序键的方法。为此，我希望得到：
- 用于获取事件键的键选择器函数，类型为
Func<TSource,TKey>
- 类型的初始键
TKey
- 一个按顺序获取下一个键的函数，类型
Func<TKey,TKey>
- 结果选择器，用于从源流中的配对事件生成结果，类型为
Func<TSource,TSource,TSource>
由于我只是使用基于 1 的整数序列进行测试，因此满足以下条件：
- 键选择器：
i => i
- 第一个键：
1
- 下一个键函数：
k => k+1
- 结果选择器：
(left,right) => left
种类
这是我的Sort尝试。它将事件缓冲到字典中并尽快将它们刷新给订阅者：
public static IObservable<TSource> Sort<TSource, TKey>
(this IObservable<TSource> source,
Func<TSource, TKey> keySelector,
TKey firstKey,
Func<TKey, TKey> nextKeyFunc)
{
return Observable.Create<TSource>(o =>
{
var nextKey = firstKey;
var buffer = new Dictionary<TKey, TSource>();
return source.Subscribe(i =>
{
if (keySelector(i).Equals(nextKey))
{
nextKey = nextKeyFunc(nextKey);
o.OnNext(i);
TSource nextValue;
while (buffer.TryGetValue(nextKey, out nextValue))
{
buffer.Remove(nextKey);
o.OnNext(nextValue);
nextKey = nextKeyFunc(nextKey);
}
}
else buffer.Add(keySelector(i), i);
});
});
}
我不得不说这是一个非常幼稚的实现。在过去的生产代码中，我已经详细说明了这一点，包括特定的错误处理、固定大小的缓冲区和超时，以防止资源泄漏。但是，它适用于这个例子。:)
有了这个排序（对不起！），我们现在可以处理多个流。
结合结果
第一次尝试
我对此的第一次尝试是产生一个无序的事件流，这些事件已经被看到了所需的次数。然后可以对其进行排序。我通过按键对元素进行分组来做到这一点，GroupByUntil用于保持每个组直到捕获两个元素。然后，每个组都是相同键的结果流。对于整数事件的简单示例，我可以只取每组的最后一个元素。但是，我不喜欢这样，因为对于更真实的场景来说，这很尴尬，因为每个结果流都可能会贡献一些有用的东西。为了感兴趣，我包含了代码。请注意，为了可以在本次尝试和我的第二次尝试之间共享测试，我接受了一个未使用的 resultSelector 参数：
public static IObservable<TSource> OrderedCollect<TSource, TKey>
(this IObservable<TSource> left,
IObservable<TSource> right,
Func<TSource, TKey> keySelector,
TKey firstKey,
Func<TKey, TKey> nextKeyFunc
Func<TSource,TSource,TSource> resultSelector)
{
return left.Merge(right)
.GroupByUntil(keySelector, x => x.Take(2).LastAsync())
.SelectMany(x => x.LastAsync())
.Sort(keySelector, firstKey, nextKeyFunc);
}
旁白：您可以修改该SelectMany子句来决定如何选择结果。与第二次尝试相比，该解决方案的一个优势是，在具有许多结果流的场景中，更容易看到如何扩展它以选择三个结果元组中的前两个到达。
第二次尝试
对于这种方法，我独立地对每个流进行排序，然后Zip将结果放在一起。这不仅是一个看起来简单得多的操作，而且以有趣的方式组合每个流的结果也容易得多。为了使测试与我的第一种方法兼容，我选择了 resultSelector 函数来使用第一个流的事件作为结果，但显然你可以灵活地在你的场景中做一些有用的事情：
public static IObservable<TSource> OrderedCollect<TSource, TKey>
(this IObservable<TSource> left,
IObservable<TSource> right,
Func<TSource, TKey> keySelector,
TKey firstKey,
Func<TKey, TKey> nextKeyFunc,
Func<TSource, TSource, TSource> resultSelector)
{
return Observable.Zip(
left.Sort(keySelector, firstKey, nextKeyFunc),
right.Sort(keySelector, firstKey, nextKeyFunc),
resultSelector);
}
另外：不难看出如何将此代码扩展到接受任意数量的输入流的更一般的情况，但正如前面提到的，使用Zip使得在给定键处阻塞非常不灵活，直到所有流的结果在。
测试用例
最后，这是我的测试，与您的示例场景相呼应。要运行这些，请导入 nuget 包rx-testing并将nunit上面的实现放入静态类中：
public class ReorderingEventsTests : ReactiveTest
{
[Test]
public void ReorderingTest1()
{
var scheduler = new TestScheduler();
var s1 = scheduler.CreateColdObservable(
OnNext(100, 1),
OnNext(200, 2),
OnNext(400, 3),
OnNext(500, 4));
var s2 = scheduler.CreateColdObservable(
OnNext(100, 1),
OnNext(200, 3),
OnNext(300, 2),
OnNext(500, 4));
var results = scheduler.CreateObserver<int>();
s1.OrderedCollect(
right: s2,
keySelector: i => i,
firstKey: 1,
nextKeyFunc: i => i + 1,
resultSelector: (left,right) => left).Subscribe(results);
scheduler.Start();
results.Messages.AssertEqual(
OnNext(100, 1),
OnNext(300, 2),
OnNext(400, 3),
OnNext(500, 4));
}
[Test]
public void ReorderingTest2()
{
var scheduler = new TestScheduler();
var s1 = scheduler.CreateColdObservable(
OnNext(100, 1),
OnNext(200, 2),
OnNext(300, 3),
OnNext(400, 4));
var s2 = scheduler.CreateColdObservable(
OnNext(100, 4),
OnNext(200, 3),
OnNext(300, 2),
OnNext(400, 1));
var results = scheduler.CreateObserver<int>();
s1.OrderedCollect(
right: s2,
keySelector: i => i,
firstKey: 1,
nextKeyFunc: i => i + 1,
resultSelector: (left, right) => left).Subscribe(results);
scheduler.Start();
results.Messages.AssertEqual(
OnNext(400, 1),
OnNext(400, 2),
OnNext(400, 3),
OnNext(400, 4));
}
}
柯里化以避免重复
最后的评论，因为我讨厌在代码中重复自己，这里有一个调整，可以避免我Sort在第二种方法中调用的重复方式。我没有将它包含在正文中，以免使不熟悉柯里化的读者感到困惑：
public static IObservable<TSource> OrderedCollect<TSource, TKey>
(this IObservable<TSource> left,
IObservable<TSource> right,
Func<TSource, TKey> keySelector,
TKey firstKey,
Func<TKey, TKey> nextKeyFunc,
Func<TSource, TSource, TSource> resultSelector)
{
Func<IObservable<TSource>, IObservable<TSource>> curriedSort =
events => events.Sort(keySelector, firstKey, nextKeyFunc);
return Observable.Zip(
curriedSort(left),
curriedSort(right),
resultSelector);
}