123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657 |
- using System;
- namespace UniRx.Operators
- {
- internal class CastObservable<TSource, TResult> : OperatorObservableBase<TResult>
- {
- readonly IObservable<TSource> source;
- public CastObservable(IObservable<TSource> source)
- : base(source.IsRequiredSubscribeOnCurrentThread())
- {
- this.source = source;
- }
- protected override IDisposable SubscribeCore(IObserver<TResult> observer, IDisposable cancel)
- {
- return source.Subscribe(new Cast(observer, cancel));
- }
- class Cast : OperatorObserverBase<TSource, TResult>
- {
- public Cast(IObserver<TResult> observer, IDisposable cancel)
- : base(observer, cancel)
- {
- }
- public override void OnNext(TSource value)
- {
- var castValue = default(TResult);
- try
- {
- castValue = (TResult)(object)value;
- }
- catch (Exception ex)
- {
- try { observer.OnError(ex); }
- finally { Dispose(); }
- return;
- }
- observer.OnNext(castValue);
- }
- public override void OnError(Exception error)
- {
- try { observer.OnError(error); }
- finally { Dispose(); }
- }
- public override void OnCompleted()
- {
- try { observer.OnCompleted(); }
- finally { Dispose(); }
- }
- }
- }
- }
|