// 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; using System.Linq; namespace System.Reactive.Linq.ObservableImpl { internal sealed class ToLookup : Producer, ToLookup._> { private readonly IObservable _source; private readonly Func _keySelector; private readonly Func _elementSelector; private readonly IEqualityComparer _comparer; public ToLookup(IObservable source, Func keySelector, Func elementSelector, IEqualityComparer comparer) { _source = source; _keySelector = keySelector; _elementSelector = elementSelector; _comparer = comparer; } protected override _ CreateSink(IObserver> observer, IDisposable cancel) => new _(this, observer, cancel); protected override IDisposable Run(_ sink) => _source.SubscribeSafe(sink); internal sealed class _ : Sink> { private readonly Func _keySelector; private readonly Func _elementSelector; private readonly Lookup _lookup; public _(ToLookup parent, IObserver> observer, IDisposable cancel) : base(observer, cancel) { _keySelector = parent._keySelector; _elementSelector = parent._elementSelector; _lookup = new Lookup(parent._comparer); } public override void OnNext(TSource value) { try { _lookup.Add(_keySelector(value), _elementSelector(value)); } catch (Exception ex) { ForwardOnError(ex); } } public override void OnCompleted() { ForwardOnNext(_lookup); ForwardOnCompleted(); } } } }