using System.Collections.Generic; namespace System.Reactive.Linq.ObservableImpl { internal sealed class MaxBy<TSource, TKey> : Producer<IList<TSource>, MaxBy<TSource, TKey>._> { internal sealed class _ : Sink<IList<TSource>>, IObserver<TSource> { private readonly MaxBy<TSource, TKey> _parent; private bool _hasValue; private TKey _lastKey; private List<TSource> _list; public _(MaxBy<TSource, TKey> parent, IObserver<IList<TSource>> observer, IDisposable cancel) : base(observer, cancel) { _parent = parent; _hasValue = false; _lastKey = default(TKey); _list = new List<TSource>(); } public void OnNext(TSource value) { TKey val = default(TKey); try { val = _parent._keySelector(value); } catch (Exception error) { _observer.OnError(error); base.Dispose(); return; } int num = 0; if (_hasValue) try { num = _parent._comparer.Compare(val, _lastKey); } catch (Exception error2) { _observer.OnError(error2); base.Dispose(); return; } else { _hasValue = true; _lastKey = val; } if (num > 0) { _lastKey = val; _list.Clear(); } if (num >= 0) _list.Add(value); } public void OnError(Exception error) { _observer.OnError(error); base.Dispose(); } public void OnCompleted() { _observer.OnNext(_list); _observer.OnCompleted(); base.Dispose(); } } private readonly IObservable<TSource> _source; private readonly Func<TSource, TKey> _keySelector; private readonly IComparer<TKey> _comparer; public MaxBy(IObservable<TSource> source, Func<TSource, TKey> keySelector, IComparer<TKey> comparer) { _source = source; _keySelector = keySelector; _comparer = comparer; } protected override _ CreateSink(IObserver<IList<TSource>> observer, IDisposable cancel) { return new _(this, observer, cancel); } protected override IDisposable Run(_ sink) { return ObservableExtensions.SubscribeSafe<TSource>(_source, (IObserver<TSource>)sink); } } }