diff options
Diffstat (limited to 'Rx.NET/System.Reactive.Linq/Reactive/Linq/Observαble/For.cs')
-rw-r--r-- | Rx.NET/System.Reactive.Linq/Reactive/Linq/Observαble/For.cs | 53 |
1 files changed, 53 insertions, 0 deletions
diff --git a/Rx.NET/System.Reactive.Linq/Reactive/Linq/Observαble/For.cs b/Rx.NET/System.Reactive.Linq/Reactive/Linq/Observαble/For.cs new file mode 100644 index 0000000..85b9e03 --- /dev/null +++ b/Rx.NET/System.Reactive.Linq/Reactive/Linq/Observαble/For.cs @@ -0,0 +1,53 @@ +// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information. + +#if !NO_PERF +using System; +using System.Collections.Generic; + +namespace System.Reactive.Linq.Observαble +{ + class For<TSource, TResult> : Producer<TResult>, IConcatenatable<TResult> + { + private readonly IEnumerable<TSource> _source; + private readonly Func<TSource, IObservable<TResult>> _resultSelector; + + public For(IEnumerable<TSource> source, Func<TSource, IObservable<TResult>> resultSelector) + { + _source = source; + _resultSelector = resultSelector; + } + + protected override IDisposable Run(IObserver<TResult> observer, IDisposable cancel, Action<IDisposable> setSink) + { + var sink = new _(observer, cancel); + setSink(sink); + return sink.Run(GetSources()); + } + + public IEnumerable<IObservable<TResult>> GetSources() + { + foreach (var item in _source) + yield return _resultSelector(item); + } + + class _ : ConcatSink<TResult> + { + public _(IObserver<TResult> observer, IDisposable cancel) + : base(observer, cancel) + { + } + + public override void OnNext(TResult value) + { + base._observer.OnNext(value); + } + + public override void OnError(Exception error) + { + base._observer.OnError(error); + base.Dispose(); + } + } + } +} +#endif |