12
\$\begingroup\$

This question over on SO about how to call a function "later" using Rx led me to post the following ill-advised piece of code to solve the problem:

void Main() {
  getScheduleTimes()
    .Where(t => t >= DateTimeOffset.Now)
    .SelectMany(t => Observable.Timer(t).ToEnumerable())
    .ToObservable()
    .Subscribe(_ => {
      Console.WriteLine("Ding! It's now {0}", DateTimeOffset.Now);
    });
}


IEnumerable<DateTimeOffset> getScheduleTimes() {
  DateTimeOffset day = DateTimeOffset.Now.Date;
  while (true) {
    yield return day + new TimeSpan(09, 00, 00);
    yield return day + new TimeSpan(13, 00, 00);
    day = day.AddDays(1);
  }
}

This doesn't work. Or rather, it appears to work -- if you run this, you'll notice that the call to .Subscribe() never returns. I was lambasted (rightly or otherwise) for involving .ToEnumerable() and .ToObservable(), as using these can cause "subscription side effects, collapsed times and dead-locks", as well as the code being confusing overall. The accepted answer builds an Observable directly.

Have I learned my lesson? Hardly. I still like the idea of using an (infinite) IEnumerable for producing the schedule. My question to you is: is anything still wrong with this code (assuming we keep the same getScheduleTimes()):

getScheduleTimes()
  .Where(t => t >= DateTimeOffset.Now)
  .Select(t => Observable.Timer(t))
  .Concat()
  .Subscribe(_ => {
    Console.WriteLine("Ding! It's now {0}", DateTimeOffset.Now);
  });

I was originally fixated on .SelectMany(), but instead .Concat() seems to be what I'm looking for. The code works, in that the subscribe no longer blocks and, as far as I can tell, no unacceptable overhead or confusion is produced, but my experience with Rx is limited. Are there any problems with this I'm not seeing, perhaps when you take it out of the toy box (cancellation, multiple subscriptions, thread overhead, etc.)?

The accepted answer also mentions using .Generate():

Observable.Generate(
  getScheduleTimes().GetEnumerator(), 
  e => e.MoveNext(), 
  e => e, 
  e => e.Current, 
  e => e.Current);

Would you say this is better than .Select(...).Concat(), based on whatever people experienced in Rx use to judge such things?

\$\endgroup\$
  • \$\begingroup\$ I fear this question is too specific to get a useful answer here. \$\endgroup\$ – svick Dec 15 '14 at 15:24
  • \$\begingroup\$ @svick: do you mean the question is inappropriate for the site (in which case I'm open to suggestions for improvement) or do you mean you think there's nobody with the specific knowledge around to answer it? I don't mind the latter, I could always answer the question myself... way, way later. \$\endgroup\$ – Jeroen Mostert Dec 15 '14 at 16:24
  • \$\begingroup\$ The latter, it's much more focused than the usual Code Review question, but I think that's okay. But there are only 16 Rx questions on CR, so there might not be anyone who can answer the question here. \$\endgroup\$ – svick Dec 15 '14 at 18:23
3
\$\begingroup\$

Although you didn't explicitly ask this (and I should now, as I am you) it's worthwhile to first look at why the first piece of code doesn't work, and why the approach is ill-advised.

Observable.ToEnumerable() has a fundamental problem in that it doesn't (cannot) know when the consumer of the resulting IEnumerable is going to enumerate. As a result, it is forced to buffer all the values that the observable is producing, until the consumer is kind enough to fetch them. Essentially, this produces a hidden producer/consumer model involving two threads. If the consumer is "sufficiently fast", the overhead of this may not be noticeable, but it's always there. Whenever you find yourself using Observable.ToEnumerable(), you're probably doing something wrong.

The second approach, where we build an enumerable sequence of observables and then .Concat() works (and I can't see any potential problems with it in terms of multiple subscriptions or cancellation), but has more overhead than necessary in that it produces a separate Observable for every single point in time, with each of these producing a single value. This wouldn't be necessary if we had a single observable to produce the values in order, and that's essentially what the third piece of code achieves.

The third piece of code is a tad cryptic if you don't see what overload of .Generate() it's calling. We can make this clearer by adding the parameter names:

Observable.Generate(
  initialState: getScheduleTimes().GetEnumerator(), 
  condition: e => e.MoveNext(), 
  iterate: e => e, 
  resultSelector: e => e.Current, 
  timeSelector: e => e.Current
);

The timeSelector parameter is documented rather sparsely as "the time selector function to control the speed of values being produced each iteration". Paraphrasing the actual implementation and assuming an implicit scheduler, .Generate() produces an observable that does this:

var dueTime = default(DateTimeOffset);
bool first = true;
var state = initialState();
if (hasResult) {
    observer.OnNext(result);
}
if (first) {
    first = false;
} else {
    state = iterate(state);
}
hasResult = condition(state);
if (hasResult) {
    result = resultSelector(state);
    dueTime = timeSelector(state);
    scheduler.Schedule(dueTime, Generate(...));
} else {
    observer.OnCompleted();
}

In the actual implementation, scheduling the tail-recursive call is more involved than this. If we substitute in our actual arguments:

var enumerator = getScheduleTimes().GetEnumerator();
if (hasResult) {
    observer.OnNext(result);
}
if (first) {
    first = false;
} else {
    enumerator = enumerator;
}
hasResult = enumerator.MoveNext();
if (hasResult) {
    result = enumerator.Current;
    dueTime = enumerator.Current;
    scheduler.Schedule(dueTime, generate(...));
} else {
    observer.OnCompleted();
}

So this is almost like a regular enumeration, but with the twist that each value we produce is also used as the time at which it should be produced.

There's a slight niggle in that, unlike the foreach pattern, this application of .Generate() will not call IEnumerator.Dispose() like an enumeration is supposed to. In this case, that's not a problem since we're using an iterator and those always have do-nothing Dispose() implementations, but if you were getting the times from another source, you would need to dispose them. That's exactly what Observable.Using() is for. If we put this together in a little extension method that works like Observable.Timer(), but for more than one point in time:

public static class ObservableExtensions {
  public static IObservable<DateTimeOffset> Timers(
    IEnumerable<DateTimeOffset> sequence
  ) {
    return Timers(sequence, Scheduler.Default);
  }

  public static IObservable<DateTimeOffset> Timers(
    IEnumerable<DateTimeOffset> sequence,
    IScheduler scheduler
  ) {
    return Observable.Using(
      () => sequence.GetEnumerator(),
      e => Observable.Generate(
        initialState: e,
        condition: ee => ee.MoveNext(),
        iterate: ee => ee,
        resultSelector: ee => ee.Current,
        timeSelector: ee => ee.Current,
        scheduler: scheduler
      )
    );
  }
}

And this would be my preferred way of doing it: using an IEnumerable allows us to test the result of code that produces a schedule in a simple way, while providing an IScheduler overload still allows us to test observable sequences as a whole using the TestScheduler if we are so inclined.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.