This is expected. The main problem is such queries end up with a time gap between producing the inner observable (here the group) and subscribing to it. In this case, the time gap is caused by ObserveOnDispatcher which puts the group (before it's subscribed
to) in a scheduler queue. At that point, the OnNext call of the GroupBy operator to its observer returns and elements are produced in the group. However, the group is only subscribed to when the dispatcher gets around to drain the work item from the queue.
A similar problem would appear when you use Delay because of similar queuing. Other higher-order stream producers, such as Window, will exhibit the same behavior.
If possible, try to do away with the first ObserveOnDispatcher. If not possible, you'll have to pull off a trick that makes the group hot and backed by a subject before hitting an enqueuing operator such as ObserveOn[Dispatcher] or Delay. The use of Replay
comes to mind, but beware you need to make the replay subject hot. An alternative approach is to use a ReplaySubject directly.
The code below shows a Unit Test project's code using Microsoft.Reactive.Testing with minimal repros of the behavior you're seeing, and a possible fix at the end:
using System.Collections.Generic;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using Microsoft.Reactive.Testing;
using Microsoft.VisualStudio.TestTools.UnitTesting;
namespace Repro
{
[TestClass]
public class ObserveOnBehavior : ReactiveTest
{
[TestMethod]
public void GroupBy()
{
var s = new TestScheduler();
var xs = s.CreateHotObservable(
OnNext(210, 1),
OnNext(220, 2),
OnNext(230, 3),
OnNext(240, 4),
OnCompleted<int>(250)
);
var res = s.Start(() => xs.GroupBy(x => x % 2));
Assert.AreEqual(3, res.Messages.Count);
Assert.IsTrue(res.Messages[0].Time == 210 && res.Messages[0].Value.Value.Key == 1 /* odd */);
Assert.IsTrue(res.Messages[1].Time == 220 && res.Messages[1].Value.Value.Key == 0 /* even */);
Assert.IsTrue(res.Messages[2].Time == 250 && res.Messages[2].Value.Kind == NotificationKind.OnCompleted);
}
[TestMethod]
public void GroupBy_Count()
{
var s = new TestScheduler();
var xs = s.CreateHotObservable(
OnNext(210, 1),
OnNext(220, 2),
OnNext(230, 3),
OnNext(240, 4),
OnCompleted<int>(250)
);
var res = s.Start(() => xs.GroupBy(x => x % 2).SelectMany(g => g.Count()));
res.Messages.AssertEqual(
OnNext(250, 2),
OnNext(250, 2),
OnCompleted<int>(250)
);
}
[TestMethod]
public void GroupBy_ObserveOn()
{
var s = new TestScheduler();
var xs = s.CreateHotObservable(
OnNext(210, 1),
OnNext(220, 2),
OnNext(230, 3),
OnNext(240, 4),
OnCompleted<int>(250)
);
var res = s.Start(() => xs.GroupBy(x => x % 2).ObserveOn(s));
//
// Notice the additional tick in the timing assertions below. Each OnNext call with a new group is scheduled
// through the scheduler due to the use of ObserveOn. In this case, the scheduler queue is empty at that point,
// so the propagation of the OnNext message on the scheduler takes place one tick later.
//
Assert.AreEqual(3, res.Messages.Count);
Assert.IsTrue(res.Messages[0].Time == 210 + 1 && res.Messages[0].Value.Value.Key == 1 /* odd */);
Assert.IsTrue(res.Messages[1].Time == 220 + 1 && res.Messages[1].Value.Value.Key == 0 /* even */);
Assert.IsTrue(res.Messages[2].Time == 250 + 1 && res.Messages[2].Value.Kind == NotificationKind.OnCompleted);
}
[TestMethod]
public void GroupBy_ObserveOn_Count()
{
var s = new TestScheduler();
var xs = s.CreateHotObservable(
OnNext(210, 1),
OnNext(220, 2),
OnNext(230, 3),
OnNext(240, 4),
OnCompleted<int>(250)
);
//
// It should be clear why we're missing messages now. The groups are subscribed to by Count at the point
// they are received, which is one tick post the first value in the group. In realistic scenarios, the
// scheduler queue may have more work, postponing the subscription to the group even further.
//
var res = s.Start(() => xs.GroupBy(x => x % 2).ObserveOn(s).SelectMany(g => g.Count()));
res.Messages.AssertEqual(
OnNext(250, 1),
OnNext(250, 1),
OnCompleted<int>(251)
);
}
[TestMethod]
public void GroupBy_Replay_ObserveOn_Count()
{
var s = new TestScheduler();
var xs = s.CreateHotObservable(
OnNext(210, 1),
OnNext(220, 2),
OnNext(230, 3),
OnNext(240, 4),
OnCompleted<int>(250)
);
//
// Using an imperative island of code, we now cure all the groups by a ReplaySubject by subscribing to it
// before control is handed to ObserveOn which can cause time gaps due to the queueing behavior.
//
var res = s.Start(() => (from g in xs.GroupBy(x => x % 2)
let replay = new ReplaySubject<int>()
let _ = g.Subscribe(replay)
select new { g.Key, Items = replay.AsObservable() })
.ObserveOn(s)
.SelectMany(g => g.Items.Count()));
res.Messages.AssertEqual(
OnNext(250, 2),
OnNext(250, 2),
OnCompleted<int>(251)
);
}
}
static class Extensions
{
public static void AssertEqual<T>(this IEnumerable<T> actual, params T[] expected)
{
ReactiveAssert.AreElementsEqual(expected, actual);
}
}
}
Hope this helps!
using (Microsoft.Sql.Cloud.DataProgrammability.Rx) { Signature.Emit("Bart De Smet"); }
Higher-Order Hot Observables (e.g. Hot GroupedObservables)
