// https://stackoverflow.com/q/72471152/1768303

// https://dotnetfiddle.net/bLYDgw

​

/*

I don't think I should be seeing "Emitting: 19" after "Cancelling"

​

Emitting: 18

TakeUntil passed: 18

OnNext: 18

Cancelling

OnError: System.OperationCanceledException: The operation was canceled.

Emitting: 19

Finally

Unhandled exception. System.OperationCanceledException: The operation was canceled.

​

I'm not seeing it when I use a custom WithCancellation operator (below)

*/

​

#nullable enable

using System;

using System.Reactive;

using System.Reactive.Disposables;

using System.Reactive.Linq;

using System.Reactive.Threading.Tasks;

using System.Threading;

using System.Threading.Tasks;

using System.Diagnostics;

​

async Task Test(CancellationToken token)

{

    var publishedSequence = Observable

        .Interval(TimeSpan.FromMilliseconds(50))

        .Do(n => Console.WriteLine($"Emitting: {n}"))

        .Skip(3)

        //.TakeUntil(Task.Delay(Timeout.Infinite, token).ToObservable())

        //.WithCancellation(token)

        .TakeUntil(

            Observable.Create<long>(

                observer => token.Register(

                    (_, token) =>

                    {

                        Console.WriteLine($"Cancelling");

                        observer.OnError(new OperationCanceledException(token));

                    },

                    null)))

        .Do(n => Console.WriteLine($"TakeUntil passed: {n}"))

        .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();

    try

    {

        await publishedSequence.ToTask();

    }

    finally

    {

        // don't dispose anything just yet

        await Task.Delay(1000);

    }

}

​

ThreadPool.SetMaxThreads(workerThreads: 100, completionPortThreads: 100);

var cts = new CancellationTokenSource(1000);

await Test(cts.Token);

Environment.Exit(0);

​

public static class RxExt

{

    public static IObservable<T> WithCancellation<T>(this IObservable<T> @this, CancellationToken cancelToken)

    {

        if (cancelToken.IsCancellationRequested)

        {

            return Observable.Throw<T>(new OperationCanceledException(cancelToken));

        }

​

        if (!cancelToken.CanBeCanceled)

        {

            return Observable.Never<T>();

        }

​

        return Observable.Create<T>(observer =>

        {

            object syncRoot = new();

            IDisposable subscription = Disposable.Empty;

            IDisposable rego = Disposable.Empty;

            bool subscribed = true;

            bool cancelled = false;

​

            subscription = @this.Subscribe(OnNext, OnError, OnCompleted);

            rego = cancelToken.Register(OnCancel, null);

            return Disposable.Create(Dispose);

​

            void Dispose()

            {

                lock (syncRoot)

                {

                    Unsubscribe();

                }

            }

​

            void OnCancel(object? _, CancellationToken token)

            {

                lock (syncRoot)

                {

                    Cancel(token);

                }

            }

​

            void Unsubscribe()

            {

                Debug.Assert(Monitor.IsEntered(syncRoot));

                if (!subscribed) { return; }

                subscribed = false;

                subscription.Dispose();

                if (!cancelToken.IsCancellationRequested)

                {

                    // a workaround for a rare deadlock when

                    // this is called from the cancellation callback

                    rego.Dispose();

                }

            }

​

            void Cancel(CancellationToken token)

            {

                Debug.Assert(Monitor.IsEntered(syncRoot));

                if (!subscribed || cancelled) { return; }

                cancelled = true;

​

                Unsubscribe();

                observer.OnError(new OperationCanceledException(token));

            }

​

            void OnNext(T item)

            {

                lock (syncRoot)

                {

                    if (!subscribed)

                    {

                        return;

                    }

                    if (cancelToken.IsCancellationRequested)

                    {

                        Cancel(cancelToken);

                    }

                    else

                    {

                        observer.OnNext(item);

                    }

                }

            }

​

            void OnError(Exception error)

            {

                lock (syncRoot)

                {

                    if (!subscribed)

                    {

                        return;

                    }

                    if (cancelToken.IsCancellationRequested)

                    {

                        Cancel(cancelToken);

                    }

                    else

                    {

                        Unsubscribe();

                        observer.OnError(error);

                    }

                }

            }

​

            void OnCompleted()

            {

                lock (syncRoot)

                {

                    if (!subscribed)

                    {

                        return;

                    }

                    if (cancelToken.IsCancellationRequested)

                    {

                        Cancel(cancelToken);

                    }

                    else

                    {

                        Unsubscribe();

                        observer.OnCompleted();

                    }

                }

            }

        });

    }

}

​