I'm new to Rx, and have found the book quite useful. However, I have a problem combining SubscribeOn and cancellation support.
In an UI application, I have a request for data that may take a long time. Obviously, I do not want to block the UI thread, and I have successfully accomplished this with a method like this:
public IObservable<string> GetData()
{
return Observable.Create<string>(observer =>
{
var cancel = new CancellationDisposable();
NewThreadScheduler.Default.Schedule(() => GetDataLongRunning(observer, cancel.Token));
return cancel;
}
}
private void GetDataLongRunning(IObserver<string> observer, CancellationToken cancellationToken)
{
for (int i = 0; i < 100; i++)
{
Thread.Sleep(100);
if (cancellationToken.IsCancellationRequested)
{
observer.OnCompleted();
return;
}
observer.OnNext(i.ToString());
}
observer.OnCompleted();
}
Notice the NewThreadScheduler, which ensures the UI is not blocked. This means the UI layer can get the observable and use it like this (using Buffer to avoid updating the UI for every single record, and ObserveOn to ensure it happens on the UI thread):
_subscription = GetData().Buffer(TimeSpan.FromSeconds(1), 10).ObserveOn(SynchronizationContext.Current.Subscribe(
items => AddRecordsToControl(items),
ex => HandleError(ex),
() => IndicateCompleted());
And if the UI layer disposes _subscription, the operation is cancelled. This all works fine.
Then I read the following in part 4 of the book:
When writing UI applications, the SubscribeOn/ObserveOn pair is very useful for two reasons:
- you do not want to block the UI thread
- but you do need to update UI objects on the UI thread.
Since this is exactly what I'm doing (and I like the idea of the caller being able to decide what thread it should execute on), I got rid of the NewThreadScheduler in GetData (and just called GetDataLongRunning instead), and then added a SubscribeOn(NewThreadScheduler.Default)
in the UI's request.
However, although it basically works as before, it does change the behaviour so cancellation no longer works. Disposing the subscription means the UI no longer received data, but the GetDataLongRunning continues to run until it is finished. The reason
is, that it is no longer just GetDataLongRunning that is running in another thread, but the entire subscribe delagate provided to Observable.Create. I.e. the CancellationDisposable object is not returned before the entire operation is finished, and therefore
not tied to the disposable returned by Subscribe call.
Now, I can of course just go back to my original code (optionally supplying the IScheduler to use from the caller as a parameter to GetData), but then I don't understand the usefulness of SubscribeOn in this scenario. Is there another way to support
cancellation when using SubscribeOn?
