diff options
author | Atsushi Eno <atsushieno@veritas-vos-liberabit.com> | 2013-01-22 12:25:22 +0400 |
---|---|---|
committer | Atsushi Eno <atsushieno@veritas-vos-liberabit.com> | 2013-01-22 12:25:22 +0400 |
commit | cde9fc6a8fe569203cb991121a35c2a9c7f4c420 (patch) | |
tree | 8633a637be4973b221d9c7e9378af5e0a08b5670 /Rx/NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Select.cs | |
parent | 8911e1d3f169a0e378b4e237926269d9218c8fd3 (diff) |
import 2b5dbddd740b, new directory structure in the original rx.
Diffstat (limited to 'Rx/NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Select.cs')
-rw-r--r-- | Rx/NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Select.cs | 138 |
1 files changed, 138 insertions, 0 deletions
diff --git a/Rx/NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Select.cs b/Rx/NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Select.cs new file mode 100644 index 0000000..e853210 --- /dev/null +++ b/Rx/NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Select.cs @@ -0,0 +1,138 @@ +// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information. + +#if !NO_PERF +using System; + +namespace System.Reactive.Linq.Observαble +{ + abstract class Select<TResult> : Producer<TResult> + { + public abstract IObservable<TResult2> Ω<TResult2>(Func<TResult, TResult2> selector); + } + + class Select<TSource, TResult> : Select<TResult> + { + private readonly IObservable<TSource> _source; + private readonly Func<TSource, TResult> _selector; + private readonly Func<TSource, int, TResult> _selectorI; + + public Select(IObservable<TSource> source, Func<TSource, TResult> selector) + { + _source = source; + _selector = selector; + } + + public Select(IObservable<TSource> source, Func<TSource, int, TResult> selector) + { + _source = source; + _selectorI = selector; + } + + public override IObservable<TResult2> Ω<TResult2>(Func<TResult, TResult2> selector) + { + if (_selector != null) + return new Select<TSource, TResult2>(_source, x => selector(_selector(x))); + else + return new Select<TResult, TResult2>(this, selector); + } + + protected override IDisposable Run(IObserver<TResult> observer, IDisposable cancel, Action<IDisposable> setSink) + { + if (_selector != null) + { + var sink = new _(this, observer, cancel); + setSink(sink); + return _source.SubscribeSafe(sink); + } + else + { + var sink = new τ(this, observer, cancel); + setSink(sink); + return _source.SubscribeSafe(sink); + } + } + + class _ : Sink<TResult>, IObserver<TSource> + { + private readonly Select<TSource, TResult> _parent; + + public _(Select<TSource, TResult> parent, IObserver<TResult> observer, IDisposable cancel) + : base(observer, cancel) + { + _parent = parent; + } + + public void OnNext(TSource value) + { + var result = default(TResult); + try + { + result = _parent._selector(value); + } + catch (Exception exception) + { + base._observer.OnError(exception); + base.Dispose(); + return; + } + + base._observer.OnNext(result); + } + + public void OnError(Exception error) + { + base._observer.OnError(error); + base.Dispose(); + } + + public void OnCompleted() + { + base._observer.OnCompleted(); + base.Dispose(); + } + } + + class τ : Sink<TResult>, IObserver<TSource> + { + private readonly Select<TSource, TResult> _parent; + private int _index; + + public τ(Select<TSource, TResult> parent, IObserver<TResult> observer, IDisposable cancel) + : base(observer, cancel) + { + _parent = parent; + _index = 0; + } + + public void OnNext(TSource value) + { + var result = default(TResult); + try + { + result = _parent._selectorI(value, checked(_index++)); + } + catch (Exception exception) + { + base._observer.OnError(exception); + base.Dispose(); + return; + } + + base._observer.OnNext(result); + } + + public void OnError(Exception error) + { + base._observer.OnError(error); + base.Dispose(); + } + + public void OnCompleted() + { + base._observer.OnCompleted(); + base.Dispose(); + } + } + } +} +#endif
\ No newline at end of file |