// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information. using System.Collections.Generic; namespace System.Reactive.Linq.ObservableImpl { internal sealed class OnErrorResumeNext : Producer { private readonly IEnumerable> _sources; public OnErrorResumeNext(IEnumerable> sources) { _sources = sources; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return sink.Run(_sources); } private sealed class _ : TailRecursiveSink { public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { } protected override IEnumerable> Extract(IObservable source) { var oern = source as OnErrorResumeNext; if (oern != null) return oern._sources; return null; } public override void OnNext(TSource value) { base._observer.OnNext(value); } public override void OnError(Exception error) { _recurse(); } public override void OnCompleted() { _recurse(); } protected override bool Fail(Exception error) { // // Note that the invocation of _recurse in OnError will // cause the next MoveNext operation to be enqueued, so // we will still return to the caller immediately. // OnError(error); return true; } } } }