Can I Pair Two Sequences Together by a Matching Key?

Question

Let's say sequence one is going out to the web to retrieve the contents of sites 1, 2, 3, 4, 5 (but will return in unpredictable order).

Sequence two is going to a database to retrieve context about these same records 1, 2, 3, 4, 5 (but for the purposes of this example will return in unpredictable order).

Is there an Rx extension method that will combine these into one sequence when each matching pair is ready in both sequences? Ie, if the first sequence returns in the order 4,2,3,5,1 and the second sequence returns in the order 1,4,3,2,5, the merged sequence would be (4,4), (3,3), (2,2), (1,1), (5,5) - as soon as each pair is ready. I've looked at Merge and Zip but they don't seem to be exactly what I'm looking for.

I wouldn't want to discard pairs that don't match, which I think rules out a simple .Where.Select combination.

Answer 1

var paired = Observable
    .Merge(aSource, bSource)
    .GroupBy(i => i)
    .SelectMany(g => g.Buffer(2).Take(1));

The test below gives the correct results. It's just taking ints at the moment, if you're using data with keys and values, then you'll need to group by i.Key instead of i.

var aSource = new Subject<int>();
var bSource = new Subject<int>();

paired.Subscribe(g => Console.WriteLine("{0}:{1}", g.ElementAt(0), g.ElementAt(1)));

aSource.OnNext(4);
bSource.OnNext(1);
aSource.OnNext(2);
bSource.OnNext(4);
aSource.OnNext(3);
bSource.OnNext(3);
aSource.OnNext(5);
bSource.OnNext(2);
aSource.OnNext(1);
bSource.OnNext(5);

yields:

4:4
3:3
2:2
1:1
5:5

Edit in response to Brandon:

For the situation where the items are different classes (AClass and BClass), the following adjustment can be made.

using Pair = Tuple<AClass, BClass>;

var paired = Observable
    .Merge(aSource.Select(a => new Pair(a, null)), bSource.Select(b => new Pair(null, b)))
    .GroupBy(p => p.Item1 != null ? p.Item1.Key : p.Item2.Key)
    .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));