using System.Collections; using System.Collections.Generic; using System.Reactive.Disposables; namespace System.Reactive.Linq.ObservableImpl { internal abstract class PushToPullAdapter<TSource, TResult> : IEnumerable<TResult>, IEnumerable { private readonly IObservable<TSource> _source; public PushToPullAdapter(IObservable<TSource> source) { _source = source; } IEnumerator IEnumerable.GetEnumerator() { return GetEnumerator(); } public IEnumerator<TResult> GetEnumerator() { SingleAssignmentDisposable singleAssignmentDisposable = new SingleAssignmentDisposable(); PushToPullSink<TSource, TResult> pushToPullSink = Run(singleAssignmentDisposable); singleAssignmentDisposable.Disposable = ObservableExtensions.SubscribeSafe<TSource>(_source, (IObserver<TSource>)pushToPullSink); return pushToPullSink; } protected abstract PushToPullSink<TSource, TResult> Run(IDisposable subscription); } }