<commit_before>﻿using System;
using System.Collections.Generic;

namespace UniRx.Operators
{
    internal class CatchObservable<T, TException> : OperatorObservableBase<T>
        where TException : Exception
    {
        readonly IObservable<T> source;
        readonly Func<TException, IObservable<T>> errorHandler;

        public CatchObservable(IObservable<T> source, Func<TException, IObservable<T>> errorHandler)
            : base(source.IsRequiredSubscribeOnCurrentThread())
        {
            this.source = source;
            this.errorHandler = errorHandler;
        }

        protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
        {
            return new Catch(this, observer, cancel).Run();
        }

        class Catch : OperatorObserverBase<T, T>
        {
            readonly CatchObservable<T, TException> parent;
            SerialDisposable serialDisposable;

            public Catch(CatchObservable<T, TException> parent, IObserver<T> observer, IDisposable cancel)
                : base(observer, cancel)
            {
                this.parent = parent;
            }

            public IDisposable Run()
            {
                this.serialDisposable = new SerialDisposable();
                this.serialDisposable.Disposable = parent.source.Subscribe(this);

                return serialDisposable;
            }

            public override void OnNext(T value)
            {
                base.observer.OnNext(value);
            }

            public override void OnError(Exception error)
            {
                var e = error as TException;
                if (e != null)
                {
                    IObservable<T> next;
                    try
                    {
                        if (parent.errorHandler == Stubs.CatchIgnore<T>)
                        {
                            next = Observable.Empty<T>(); // for avoid iOS AOT
                        }
                        else
                        {
                            next = parent.errorHandler(e);
                        }
                    }
                    catch (Exception ex)
                    {
                        try { observer.OnError(ex); } finally { Dispose(); };
                        return;
                    }

                    var d = new SingleAssignmentDisposable();
                    serialDisposable.Disposable = d;
                    d.Disposable = next.Subscribe(observer);
                }
                else
                {
                    try { observer.OnError(error); } finally { Dispose(); };
                    return;
                }
            }

            public override void OnCompleted()
            {<commit_msg>Fix Catch which has returned time-related observable is cancelled unexpectedly. #186<commit_after>﻿using System;
using System.Collections.Generic;

namespace UniRx.Operators
{
    internal class CatchObservable<T, TException> : OperatorObservableBase<T>
        where TException : Exception
    {
        readonly IObservable<T> source;
        readonly Func<TException, IObservable<T>> errorHandler;

        public CatchObservable(IObservable<T> source, Func<TException, IObservable<T>> errorHandler)
            : base(source.IsRequiredSubscribeOnCurrentThread())
        {
            this.source = source;
            this.errorHandler = errorHandler;
        }

        protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
        {
            return new Catch(this, observer, cancel).Run();
        }

        class Catch : OperatorObserverBase<T, T>
        {
            readonly CatchObservable<T, TException> parent;
            SingleAssignmentDisposable sourceSubscription;
            SingleAssignmentDisposable exceptionSubscription;

            public Catch(CatchObservable<T, TException> parent, IObserver<T> observer, IDisposable cancel)
                : base(observer, cancel)
            {
                this.parent = parent;
            }

            public IDisposable Run()
            {
                this.sourceSubscription = new SingleAssignmentDisposable();
                this.exceptionSubscription = new SingleAssignmentDisposable();

                this.sourceSubscription.Disposable = parent.source.Subscribe(this);
                return StableCompositeDisposable.Create(sourceSubscription, exceptionSubscription);
            }

            public override void OnNext(T value)
            {
                base.observer.OnNext(value);
            }

            public override void OnError(Exception error)
            {
                var e = error as TException;
                if (e != null)
                {
                    IObservable<T> next;
                    try
                    {
                        if (parent.errorHandler == Stubs.CatchIgnore<T>)
                        {
                            next = Observable.Empty<T>(); // for avoid iOS AOT
                        }
                        else
                        {
                            next = parent.errorHandler(e);
                        }
                    }
                    catch (Exception ex)
                    {
                        try { observer.OnError(ex); } finally { Dispose(); };
                        return;
                    }

                    exceptionSubscription.Disposable = next.Subscribe(observer);
                }
                else
                {
                    try { observer.OnError(error); } finally { Dispose(); };
                    return;
                }
            }

            public override void OnCompleted()
            {
{'fraction_non_alphanumeric': 0.07465277777777778, 'fraction_numerical': 0.0005208333333333333, 'mean_word_length': 1.2870186581976975, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4547019', 'n_tokens_mistral': 1388, 'n_tokens_neox': 1269, 'n_words': 282}