这个问题基于这里标题相似的问题，有两个区别：
- 我正在匹配多个键。没问题
- 按键可能重复。问题
我的测试代码如下。我需要以下行为：
- 一旦观察到至少一个
CoordMetrics和一个CoordData，就发布CoordBundle - 如果某个特定的X/Y键在任一可观察对象上重复出现，则会发布一个新的CoordBundle
我必须做些什么才能做到这一点？
public class CoordMetrics
{
internal CoordMetrics(int x, int y, IEnumerable<IMetric> metrics)
{
X = x;
Y = y;
Metrics = metrics;
}
internal int X { get; private set; }
internal int Y { get; private set; }
internal IEnumerable<IMetric> Metrics { get; private set; }
}
public class CoordData
{
internal CoordData(int x, int y, IEnumerable<IDatum> data)
{
X = x;
Y = y;
Data = data;
}
internal int X { get; private set; }
internal int Y { get; private set; }
internal IEnumerable<IDatum> Data { get; private set; }
}
public class CoordBundle
{
internal CoordBundle(int x, int y, IEnumerable<IMetric> metrics, IEnumerable<IDatum> data)
{
X = x;
Y = y;
Metrics = metrics;
Data = data;
}
internal int X { get; private set; }
internal int Y { get; private set; }
internal IEnumerable<IMetric> Metrics { get; private set; }
internal IEnumerable<IDatum> Data { get; private set; }
}
[TestClass]
public class PairingTest
{
[TestMethod, TestCategory("Temp")]
public void PairedObservableTest()
{
Trace.Listeners.Add(new TextWriterTraceListener(Console.Out));
var aSource = new Subject<CoordMetrics>();
var bSource = new Subject<CoordData>();
var paired = Observable.Merge(aSource.Select(a => new Pair(a, null)), bSource.Select(b => new Pair(null, b)))
.GroupBy(p => p.Item1 != null ? new { p.Item1.X, p.Item1.Y } : new { p.Item2.X, p.Item2.Y })
.SelectMany(g => g.Buffer(2).Take(1))
.Select(g => new Pair(
g.ElementAt(0).Item1 ?? g.ElementAt(1).Item1,
g.ElementAt(0).Item2 ?? g.ElementAt(1).Item2))
.Select(t => new CoordBundle(t.Item1.X, t.Item1.Y, t.Item1.Metrics, t.Item2.Data));
paired.Subscribe(g => Trace.WriteLine(String.Format("{0},{1}", g.X, g.Y)));
bSource.OnNext(new CoordData(2, 1, Enumerable.Empty<IDatum>()));
aSource.OnNext(new CoordMetrics(2, 2, Enumerable.Empty<IMetric>()));
aSource.OnNext(new CoordMetrics(1, 1, Enumerable.Empty<IMetric>()));
bSource.OnNext(new CoordData(1, 2, Enumerable.Empty<IDatum>()));
bSource.OnNext(new CoordData(2, 2, Enumerable.Empty<IDatum>()));
bSource.OnNext(new CoordData(1, 1, Enumerable.Empty<IDatum>()));
aSource.OnNext(new CoordMetrics(1, 2, Enumerable.Empty<IMetric>()));
aSource.OnNext(new CoordMetrics(2, 1, Enumerable.Empty<IMetric>()));
aSource.OnNext(new CoordMetrics(2, 2, Enumerable.Empty<IMetric>()));
bSource.OnNext(new CoordData(2,2,Enumerable.Empty<IDatum>()));
}
}
所需输出-以上代码仅输出前4行：
2,2
1,1
1,2
2,1
2,2
2,2
我想我有你想要的。公平地说，这不是一个容易的问题。一个序列是另一个序列的种子是很常见的，但这里的复杂情况是，任何一个序列都可以是另一序列的种子。
为了得到一个有效的解决方案，我做的第一件事就是将其分解为一个可验证的单元测试。我建议使用TestScheduler及其相关类型（而不是受试者等）。
我根据你的要求制作了大理石图。然后我可以将其映射到两个测试输入序列和预期输出序列中。
最后一部分是实际创建查询。
我最终使用的方法是创建两个序列，尝试从主序列和子序列进行匹配->SelectMany+Where。然而，由于两个输入都可以扮演主序列的角色，我需要做两次。由于我会订阅两次，所以我需要共享序列->Publish()。此外，由于每个序列都可能产生多个值，所以当重复项到达->TakeUntil时，我需要取消之前匹配的匹配。最后，我只是将两个结果集合并在一起->Merge。
*我考虑了CCD_ 10&CombineLatest，但它们似乎对我不起作用。
[TestClass]
public class PairingTest
{
[TestMethod, TestCategory("Temp")]
public void PairedObservableTest()
{
var scheduer = new TestScheduler();
/*
Legend
a = aSource (CoordMetrics)
b = bSource (CoordData)
r = expected result
a ----2--1-----------1--2--2-----
2 1 2 1 2
b -2--------1--2--1-----------2--
1 2 2 1 2
r -------------2--1--1--2--2--2--
2 1 2 1 2 2
*/
var aSource = scheduer.CreateColdObservable<CoordMetrics>(
ReactiveTest.OnNext(5, new CoordMetrics(2, 2)),
ReactiveTest.OnNext(8, new CoordMetrics(1, 1)),
ReactiveTest.OnNext(20, new CoordMetrics(1, 2)),
ReactiveTest.OnNext(23, new CoordMetrics(2, 1)),
ReactiveTest.OnNext(26, new CoordMetrics(2, 2))
);
var bSource = scheduer.CreateColdObservable<CoordData>(
ReactiveTest.OnNext(2, new CoordData(2, 1)),
ReactiveTest.OnNext(11, new CoordData(1, 2)),
ReactiveTest.OnNext(14, new CoordData(2, 2)),
ReactiveTest.OnNext(17, new CoordData(1, 1)),
ReactiveTest.OnNext(29, new CoordData(2, 2))
);
var testObserver = scheduer.CreateObserver<string>();
Implementation(aSource, bSource)
.Subscribe(testObserver);
scheduer.Start();
ReactiveAssert.AreElementsEqual(
new[] {
ReactiveTest.OnNext(14, "2,2"),
ReactiveTest.OnNext(17, "1,1"),
ReactiveTest.OnNext(20, "1,2"),
ReactiveTest.OnNext(23, "2,1"),
ReactiveTest.OnNext(26, "2,2"),
ReactiveTest.OnNext(29, "2,2")
},
testObserver.Messages
);
}
private static IObservable<string> Implementation(IObservable<CoordMetrics> aSource, IObservable<CoordData> bSource)
{
return Observable.Create<string>(observer =>
{
var aShared = aSource.Publish();
var bShared = bSource.Publish();
var fromA = aShared.SelectMany(a => bShared
//Find matching values from B's
.Where(b => a.X == b.X && a.Y == b.Y)
//Only run until another matching A is produced
.TakeUntil(aShared.Where(a2 => a2.X == a.X && a2.Y == a.Y))
//Project/Map to required type.
.Select(b => new CoordBundle(a.X, a.Y /*, a.Metrics, b.Data*/ ))
);
var fromB = bShared.SelectMany(b => aShared
//Find matching values from A's
.Where(a => a.X == b.X && a.Y == b.Y)
//Only run until another matching B is produced
.TakeUntil(bShared.Where(b2 => b2.X == b.X && b2.Y == b.Y))
//Project/Map to required type.
.Select(a => new CoordBundle(a.X, a.Y /*, a.Metrics, b.Data*/ ))
);
var paired = Observable.Merge(fromA, fromB);
paired
.Select(g => String.Format("{0},{1}", g.X, g.Y))
.Subscribe(observer);
return new CompositeDisposable(aShared.Connect(), bShared.Connect());
});
}
}
// Define other methods and classes here
public class CoordMetrics
{
internal CoordMetrics(int x, int y)
{
X = x;
Y = y;
}
internal int X { get; private set; }
internal int Y { get; private set; }
}
public class CoordData
{
internal CoordData(int x, int y)
{
X = x;
Y = y;
}
internal int X { get; private set; }
internal int Y { get; private set; }
}
public class CoordBundle
{
internal CoordBundle(int x, int y)
{
X = x;
Y = y;
}
internal int X { get; private set; }
internal int Y { get; private set; }
}
public class Pair
{
public Pair(CoordMetrics x, CoordData y)
{
Item1 = x;
Item2 = y;
}
public CoordMetrics Item1 { get; set; }
public CoordData Item2 { get; set; }
}