// 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. namespace System.Reactive.Linq.ObservableImpl { internal static class Select { internal sealed class Selector : Producer { private readonly IObservable _source; private readonly Func _selector; public Selector(IObservable source, Func selector) { _source = source; _selector = selector; } protected override _ CreateSink(IObserver observer) => new _(_selector, observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : Sink { private readonly Func _selector; public _(Func selector, IObserver observer) : base(observer) { _selector = selector; } public override void OnNext(TSource value) { TResult result; try { result = _selector(value); } catch (Exception exception) { ForwardOnError(exception); return; } ForwardOnNext(result); } } } internal sealed class SelectorIndexed : Producer { private readonly IObservable _source; private readonly Func _selector; public SelectorIndexed(IObservable source, Func selector) { _source = source; _selector = selector; } protected override _ CreateSink(IObserver observer) => new _(_selector, observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : Sink { private readonly Func _selector; private int _index; public _(Func selector, IObserver observer) : base(observer) { _selector = selector; } public override void OnNext(TSource value) { TResult result; try { result = _selector(value, checked(_index++)); } catch (Exception exception) { ForwardOnError(exception); return; } ForwardOnNext(result); } } } } }