2

How to plug a cancellation token into an existing IObservable pipeline before calling Publish on it (i.e., before it becomes an IConnectableObservable)?

This must be a part of a cold observable pipeline, before subscribing to it (otherwise, I could pass a CancellationToken token to IObservable's Subscribe, RunAsync, ToTask etc).

Is there a recommended pattern for this?

I can think of using TakeUntil to achieve that, as suggested by Theodor Zoulias here. For example:

using System.Reactive.Linq;
using System.Reactive.Threading.Tasks;

async Task Test(CancellationToken token)
{
    var publishedSequence = Observable
        .Interval(TimeSpan.FromMilliseconds(100))
        .Do(n => Console.WriteLine($"Emitting: {n}"))
        .Skip(3)
        .TakeUntil(
            Observable.Create<long>(
                observer => token.Register(
                    (_, token) => observer.OnError(new OperationCanceledException(token)),
                    null)))
        .Finally(() => Console.WriteLine($"Finally"))
        .Publish();

    using var subscription = publishedSequence.Subscribe(
            onNext: n => Console.WriteLine($"OnNext: {n}"),
            onError: e => Console.WriteLine($"OnError: {e}"),
            onCompleted: () => Console.WriteLine("OnCompleted"));

    using var connection = publishedSequence.Connect();
    await publishedSequence.ToTask();
}

var cts = new CancellationTokenSource(1000);
await Test(cts.Token);

Ouput:

Emitting: 0
Emitting: 1
Emitting: 2
Emitting: 3
OnNext: 3
Emitting: 4
OnNext: 4
Emitting: 5
OnNext: 5
Emitting: 6
OnNext: 6
Emitting: 7
OnNext: 7
Emitting: 8
OnNext: 8
OnError: System.OperationCanceledException: The operation was canceled.
Finally

I also have a prototype of a custom operator, WithCancellation, which is basically a pass-through IObservable also listening for a cancellation signal. I'd rather stick with a standard approach though.

Updated, I think I've found a race condition in how TakeUntil works (maybe a bug or just a behavior I can't explain), fiddle. I can't repro it with my homegrown WithCancellation implementation (commented out in the fiddle).

Updated, neither can I repro it if I use .TakeUntil(Task.Delay(Timeout.Infinite, token).ToObservable()).

1 Answer 1

1

Indeed, there is no library method for this. I would create an observable from the CancellationToken and then use the TakeUntil operator.

public static IObservable<Unit> ToObservable(this CancellationToken cancellationToken) {
  // if the token can't be cancelled, use Never which will not complete
  if (!cancellationToken.CanBeCanceled) {
    return Observable.Never<Unit>();
  }

  // if the token is already cancelled, use Return which publishes
  // Unit and completes immediately.
  // This may save you from ObjectDisposedException later
  if (cancellationToken.IsCancellationRequested) {
    return Observable.Return<Unit>();
  }

  // use Create so that each .Subscribe is handled independently
  return Observable.Create<Unit>(observer => {

    // Observable.Create does not handle errors on its own
    try {
      // return the registration because Dispose will unregister it
      return cancellationToken.Register(() => {
        // When the token is cancelled, publish and complete
        observer.OnNext(Unit.Default);
        observer.OnCompleted();
      });
    }
    catch (ObjectDisposedException e) {
      // todo: consider handling this as if it were cancellation
      observer.OnError(e);
    }
  });
}

public static IObservable<T> TakeUntil<T>(this IObservable<T> source, CancellationToken cancellationToken)
  => source.TakeUntil(cancellationToken.ToObservable());

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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