Joining Rx Streams

I'm trying to model an Rx query that is not trivial (to me):

• In a room there are Men and Women.
• They enter and leave the room, and while in the room sometimes they change their location.
• Each man can look at one (or zero) woman at a given time.

• Each man has the following properties:

  class Man
  {
    public const int LookingAtNobody = 0;
    public int Id { get; set; }
    public double Location { get; set; }
    public int LookingAt { get; set; }
  }
  

• Each woman has the following properties:

  class Woman
  {
    public int Id { get; set; }
    public double Location { get; set; }
  }
  

• To represent the Men we have IObservable<IObservable<Man>>, and to represent the Women we have IObservable<IObservable<Woman>>.

How would you use Rx to generate vectors from men to the women they're looking at: IObservable<IObservable<Tuple<double,double>>> ?

To help, here are a few unit-tests for some simple cases:

public class Tests : ReactiveTest
{
    [Test]
    public void Puzzle1()
    {
        var scheduler = new TestScheduler();

        var m1 = scheduler.CreateHotObservable(
            OnNext(100, new Man { Id = 1, Location = 1.0, LookingAt = Man.LookingAtNobody }),
            OnNext(200, new Man { Id = 1, Location = 2.0, LookingAt = 10 }),
            OnCompleted<Man>(300));

        var w1 = scheduler.CreateHotObservable(
            OnNext(150, new Woman { Id = 10, Location = 10.0 }),
            OnNext(250, new Woman { Id = 10, Location = 20.0 }),
            OnCompleted<Woman>(350));

        var men = scheduler.CreateHotObservable(OnNext(50, m1));
        var women = scheduler.CreateHotObservable(OnNext(50, w1));

        var results = runQuery(scheduler, women, men);

        var innerResults = (from msg in results