namespace System.Reactive.Linq.ObservableImpl { internal sealed class SumDecimal : Producer<decimal, SumDecimal._> { internal sealed class _ : Sink<decimal>, IObserver<decimal> { private decimal _sum; public _(IObserver<decimal> observer, IDisposable cancel) : base(observer, cancel) { _sum = default(decimal); } public void OnNext(decimal value) { _sum += value; } public void OnError(Exception error) { _observer.OnError(error); base.Dispose(); } public void OnCompleted() { _observer.OnNext(_sum); _observer.OnCompleted(); base.Dispose(); } } private readonly IObservable<decimal> _source; public SumDecimal(IObservable<decimal> source) { _source = source; } protected override _ CreateSink(IObserver<decimal> observer, IDisposable cancel) { return new _(observer, cancel); } protected override IDisposable Run(_ sink) { return _source.SubscribeSafe(sink); } } }