Outer join streams

Question

Have the query below that does not produce a value until both streams have generated a value. The behaviour I need is that a default value (new Order()) should be generated if the inner stream (orders) does not have a value

public static void Main()
  {   

   Subject<Quote> quotes = new Subject<Quote>();
   Subject<Order> orders = new Subject<Order>();
     
           
   var query = quotes.Join(orders,_=>quotes,_=>Observable.Empty<Unit>(), (q,e)=>Tuple.Create(q, e));
   query.Dump();

   quotes.OnNext(new Quote {InstrumentId = "A", Px = 10.10m, Exchange = "O", Volume = 100L});
   quotes.OnNext(new Quote {InstrumentId = "A", Px = 50.10m, Exchange = "N", Volume = 100L});
   Thread.Sleep(1000);
   orders.OnNext(new Order {InstrumentId = "A", Status = "O", OrderQty = 100});
   orders.OnNext(new Order {InstrumentId = "A", Status = "R", OrderQty = 100});
   orders.OnNext(new Order {InstrumentId = "A", Status = "O", OrderQty = 100});
   
   quotes.OnNext(new Quote {InstrumentId = "A", Px = 20.11m, Exchange = "R", Volume = 100L});

  }
     public class Quote
  {
   public string InstrumentId { get; set; }
   public decimal Px { get; set; }
   public string Exchange { get; set; }
   public long Volume { get; set; }

   public override string ToString()
   {
    return String.Format("InstrumentId: {0}, Price: {1} Volume: {2}", InstrumentId, Px, Volume);
   }

  }

  public class Order
  {
   public string InstrumentId { get; set; }
   public int OrderQty { get; set; }
      public string Status { get; set; }
  }

Answer 1

You could simply prefix the Orders sequence with a default Order instance.

I drew out a marble diagram of what I think your query would produce and I am not sure what you are hoping to get when a subsequent quote is produced. i.e. If an initial quote is produced you want it to create a tuple with a default instance of an Order. When an Order is produced then a new tuple of that new Order and the Quote is produced. What are you expecting when the next Quote is produced? Should it get the last order? Should it wait for the next order? Should it use a default instance of an order?

Is it also your intention that once a quote is produced it will pair with every subsequent order? This will mean each time an Order is pushed it will be joined with every quote that has been pushed, which may scale poorly.

---

Lee Campbell http://LeeCampbell.blogspot.com

Answer 2

When the subsequent quote is produced it should get the last order. (if the last order were the default order then that should be in the tuple produced).

Also, I don't need the orders to be combined with every quote produced - just the last quote produced.

So my desired result is as follows

Order {}, Quote {InstrumentId = "A", Px = 10.10m, Exchange = "O", Volume = 100L}

Order {}, Quote {InstrumentId = "A", Px = 50.10m, Exchange = "N", Volume = 100L}

Order {InstrumentId = "A", Status = "O", OrderQty = 100}, Quote {InstrumentId = "A", Px = 50.10m, Exchange = "N", Volume = 100L}

Order {InstrumentId = "A", Status = "R", OrderQty = 100},Quote {InstrumentId = "A", Px = 50.10m, Exchange = "N", Volume = 100L}

Order {InstrumentId = "A", Status = "O", OrderQty = 100}, Quote {InstrumentId = "A", Px = 50.10m, Exchange = "N", Volume = 100L}

Order {InstrumentId = "A", Status = "O", OrderQty = 100}, Quote {InstrumentId = "A", Px = 20.11m, Exchange = "R", Volume = 100L}

And I need the values to be produced when either stream publishes but before either stream issues an OnComplete.

Answer 3

If they are your requriements then this could be easy. It seems you could just

• prefix each sequence with a default value,
• use combinelatest
• and ignore the first value(because it will be the 2 default values).

quotes.StartWith(new Quote())
		.CombineLatest(
			orders.StartWith(new Order()),
			(q,o)=>Tuple.Create(q, o))
		.Select(t=>string.Format("{0} {1}", t.Item2, t.Item1))
		.Skip(1)
		.Dump();

Output

InstrumentId: , OrderQty: 0 Status:  InstrumentId: A, Price: 10.10 Volume: 100
InstrumentId: , OrderQty: 0 Status:  InstrumentId: A, Price: 50.10 Volume: 100
InstrumentId: A, OrderQty: 100 Status: O InstrumentId: A, Price: 50.10 Volume: 100
InstrumentId: A, OrderQty: 100 Status: R InstrumentId: A, Price: 50.10 Volume: 100
InstrumentId: A, OrderQty: 100 Status: O InstrumentId: A, Price: 50.10 Volume: 100
InstrumentId: A, OrderQty: 100 Status: O InstrumentId: A, Price: 20.11 Volume: 100

HTH

Lee

---

Lee Campbell http://LeeCampbell.blogspot.com

Answer 4

If I had multiple Quote InstrumentID's and Order InstrumentID's that neeed to be matched and combined - can I use an elegant solution like above  - pls refer to my earlier post to see how I am tackling it currently:

http://social.msdn.microsoft.com/Forums/vi-VN/rx/thread/4c4bab36-0818-4945-a5cc-0c74eb265e3b