using System.Collections.Generic; namespace System.Reactive.Linq.ObservableImpl { internal sealed class ZipObserver<T> : IObserver<T> { private readonly object _gate; private readonly IZip _parent; private readonly int _index; private readonly IDisposable _self; private readonly Queue<T> _values; public Queue<T> Values => _values; public ZipObserver(object gate, IZip parent, int index, IDisposable self) { _gate = gate; _parent = parent; _index = index; _self = self; _values = new Queue<T>(); } public void OnNext(T value) { lock (_gate) { _values.Enqueue(value); _parent.Next(_index); } } public void OnError(Exception error) { _self.Dispose(); lock (_gate) { _parent.Fail(error); } } public void OnCompleted() { _self.Dispose(); lock (_gate) { _parent.Done(_index); } } } }