UniRx操作符_自Rx.Net

Time:时间
- Timer
- Interval
- Throttle
- Delay
- Sample
- Timestamp
- ThrottleFirst
- TimeInterval
- Timeout
Paging:分页
- TakeUntil
- SkipUntil
- Buffer
Creation:创建
- Return
- Defer
- Never
- Create
Aggregate:聚合
- Scan
Concatenate:联合
- Merge
- Switch
- StartWith
- CombineLatest
Common:常用
- Do
- Materialize/Dematerialize
- IgnoreElements
- DistinctUntilChanged
Concurrency:并发
- Amb
Events:事件
- FromEvent
Bindings:绑定
- Publish
- RefCount
- Replay
ErrorHandling
- Catch
- Finally
DelaySubscription
PairWise

Time:时间

Timer

它在⼀个给定的延迟后发射⼀个值/执行任务

        Observable.Timer(TimeSpan.FromSeconds(1.0), TimeSpan.FromSeconds(0.1f)).Subscribe(_ => Debug.Log("after 1 seconds"));

Interval

        Observable.Interval(TimeSpan.FromSeconds(0.5f)).Subscribe(times => Debug.Log(times));

Throttle

节流阀仅在过了⼀段指定的时间还没发射数据时才发射⼀个数据

        Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Throttle(TimeSpan.FromSeconds(1.0f)).Subscribe(_ => Debug.Log("after 1 seconds"));

Delay

延迟⼀段指定的时间再发射来⾃ Observable 的发射物

        Observable.Timer(TimeSpan.FromSeconds(1.0f)).Select(_ =>{Debug.Log("1 seconds");return _;}).Delay(TimeSpan.FromSeconds(1.0f)).Select(_ =>{Debug.Log("2 seconds");return _;}).Delay(TimeSpan.FromSeconds(1.0f)).Subscribe(_ => Debug.Log("3 seconds"));Observable.ReturnUnit().Delay(TimeSpan.FromSeconds(5.0f)).Subscribe(_ => Debug.Log("has return"));

Sample

定期发射 Observable 最近发射的数据项

定时查看⼀个Observable，然后发射⾃上次采样以来它最近发射的数据

        Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Select(_ => clickCount++).Sample(TimeSpan.FromSeconds(3)).Subscribe(_ => Debug.LogFormat("第 {0} 次点击", clickCount));

Timestamp

给发射的数据项附加⼀个时间戳

        Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Timestamp().Subscribe(timestamp => { Debug.LogFormat("timestamp:{0}", timestamp.Timestamp.LocalDateTime); });

ThrottleFirst

它总是发射原始的第⼀项数据，⽽不是最近的⼀项
它默认在 computation 调度器上执⾏，但是可以使⽤第三个参数指定其它的调度器

        Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).ThrottleFirst(TimeSpan.FromSeconds(2.0)).Subscribe(_ => Debug.Log("clicked!"));

⿏标点击之后，⽴即输出 “clicked”，输出之后的 2 秒内点击⽆效

TimeInterval

将⼀个发射数据的 Observable 转换为发射那些数据发射时间间隔的 Observable

        Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Select(_ => "clicked").TimeInterval().Subscribe(timeInterval => Debug.LogFormat("{0},{1}", timeInterval.Interval, timeInterval.Value));

表示了每次点击之间的时间间隔

Timeout

对原始 Observable 的⼀个镜像，如果过了⼀个指定的时⻓仍没有发射数据，它会发⼀个错误通知，以⼀个 onError 通知
终⽌这个 Observable

       Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Timeout(TimeSpan.FromSeconds(3.0f)).Subscribe(_ =>{Debug.Log("mouse clicked");});ObservableWWW.Get("http://baidu.com").Timeout(TimeSpan.FromSeconds(0.1f)).Subscribe(_ => { }, e =>{Debug.LogErrorFormat("发生了异常:{0}", e.Message);});

Paging:分页

TakeUntil

TakeUntil 订阅并开始发射原始 Observable，它还监视你提供的第⼆个 Observable。如果第⼆个 Observable 发射了⼀项数据或者发射了⼀个终⽌通知， TakeUntil 返回的Observable会停⽌发射原始 Observable 并终⽌

        Observable.EveryUpdate().TakeUntil(Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0))).Subscribe(_ => Debug.Log("123"));

SkipUntil

丢弃原始 Observable 发射的数据，直到第⼆个 Observable 发射了⼀项数据

       var clickStream = this.UpdateAsObservable().Where(_ => Input.GetMouseButtonDown(0));// 监听this.UpdateAsObservable().SkipUntil(clickStream).Take(100).Repeat().Subscribe(_ => Debug.Log("mouse clicked"));

Buffer

缓冲

        Observable.Interval(TimeSpan.FromSeconds(1.0f)).Buffer(TimeSpan.FromSeconds(4.0f)).Subscribe(countList =>{Debug.Log("countList");countList.ToList().ForEach(count => Debug.Log(count));});

Creation:创建

Return

        Observable.Return(Unit.Default).Delay(TimeSpan.FromSeconds(1.0f)).Repeat().Subscribe(_ => Debug.Log("after 1 seconds"));Observable.Return("hello").Subscribe(Debug.Log);

Defer

直到有观察者订阅时才创建 Observable，并且为每个观察者创建⼀个新的 Observable

        Observable.Defer(() => Observable.Start(() => random.Next())).Delay(TimeSpan.FromMilliseconds(1000)).Repeat().Subscribe(randomNumber => Debug.Log(randomNumber));

Never

创建⼀个不发射数据也不终⽌的Observable

        var never = Observable.Never<string>();never.Subscribe(_ => Debug.Log(_), () => { Debug.Log("completed"); });

Create

使⽤⼀个函数从头开始创建⼀个Observable

给这个操作符传递⼀个接受观察者作为参数的函数，编写这个函数让它的⾏为表现为⼀个 Observable ，恰当的调⽤观察者的 onNext， onError 和 onCompleted ⽅法

        Observable.Create<int>(observer =>{observer.OnNext(1);observer.OnNext(2);Observable.Timer(TimeSpan.FromSeconds(1.0f)).Subscribe(_ => observer.OnCompleted());return Disposable.Create(() => Debug.Log("观察者已销毁"));}).Subscribe(number => Debug.LogFormat("number is:{0}", number));

Aggregate:聚合

Scan

连续地对数据序列的每⼀项应⽤⼀个函数，然后连续发射结果

对原始Observable发射的第⼀项数据应⽤⼀个函数，然后将那个函数的结果作为⾃⼰的第⼀项数据发射。它将函数的结果同第二项数据⼀起填充给这个函数来产⽣它⾃⼰的第⼆项数据。它持续进⾏这个过程来产⽣剩余的数据序列

        Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Scan(0, (times, nextValue) => ++times).Subscribe(times => Debug.LogFormat("clicked {0} times", times));

Concatenate:联合

Merge

结合事件流

        var leftClickStream = Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Select(_ => "A");var rightClickStream = Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(1)).Select(_ => "B");leftClickStream.Merge(rightClickStream, rightClickStream).Subscribe(clickEventName => Debug.Log(clickEventName));

Switch

订阅⼀个发射多个 Observables 的 Observable。它每次观察那些 Observables 中的⼀个，Switch 返回的这个 Observable 取消订阅前⼀个发射数据的 Observable，开始发射最近的Observable 发射的数据。注意：当原始 Observable 发射了⼀个新的 Observable 时（不是这个新的Observable 发射了⼀条数据时），它将取消订阅之前的那个 Observable。这意味着，在后来那个
Observable 产⽣之后到它开始发射数据之前的这段时间⾥，前⼀个 Observable 发射的数据将被丢弃

        var stateJump = Observable.Return("jump state");Observable.EveryUpdate().Where(_ => Input.GetKeyDown(KeyCode.Space)).Select(_ => stateJump).Switch().Subscribe(stateName => Debug.Log(stateName));var wObservable = Observable.EveryUpdate().Where(_ => Input.GetKeyDown(KeyCode.W));var aObservable = Observable.EveryUpdate().Where(_ => Input.GetKeyDown(KeyCode.A));var sObservable = Observable.EveryUpdate().Where(_ => Input.GetKeyDown(KeyCode.S));var dObservable = Observable.EveryUpdate().Where(_ => Input.GetKeyDown(KeyCode.D));wObservable.Select(_ => aObservable).Switch().Select(_ => sObservable).Switch().Select(_ => dObservable).Switch().Repeat().Subscribe(_ => Debug.Log("触发彩蛋"));

StartWith

在发射数据之前先发射⼀个指定的数据序列

        Observable.Return("baidu.com").StartWith("http", "://").Aggregate((current, next) => current + next).Subscribe(Debug.Log);

CombineLatest

当两个 Observables 中的任何⼀个发射了数据时，使⽤⼀个函数结合每个 Observable 发射的最近数据项，并且基于这个函数的结果发射数据

        var a = 0;var i = 0;var leftStream = this.UpdateAsObservable().Where(_ =>Input.GetMouseButtonDown(0)).Select(_ => (++a).ToString());var rightStream = this.UpdateAsObservable().Where(_ =>Input.GetMouseButtonDown(1)).Select(_ => (++i).ToString());leftStream.CombineLatest(rightStream, (left, right) => left + right).Subscribe(Debug.Log);

Common:常用

Do

        Observable.ReturnUnit().Delay(TimeSpan.FromSeconds(1.0f)).Do(_ => Debug.Log("after 1 seconds")).Delay(TimeSpan.FromSeconds(1.0f)).Do(_ => Debug.Log("after 2 seconds")).Delay(TimeSpan.FromSeconds(1.0f)).Do(_ => Debug.Log("after 3 seconds")).Delay(TimeSpan.FromSeconds(1.0f)).Do(_ => Debug.Log("after 4 seconds")).Subscribe();

Materialize/Dematerialize

Materialize 将数据项和事件通知都当做数据项发射

⼀个合法的有限的 Obversable 将调⽤它的观察者的 onNext ⽅法零次或多次，然后调⽤观察者的 onCompleted 或 onError 正好⼀次。 Materialize 操作符将这⼀系列调⽤，包括原来的 onNext 通知和终⽌通知 onCompleted 或 onError 都转换为⼀个Observable 发射的数据序列

        var subject = new Subject<int>();var onlyExceptions = subject.Materialize().Where(notification => notification.Exception != null).Dematerialize();subject.Subscribe(number => Debug.LogFormat("subcribe 1:{0}", number), e => Debug.LogFormat("subcribe excpetion 1:{0}", e));onlyExceptions.Subscribe(number => Debug.LogFormat("subcribe 2:{0}", number), e => Debug.LogFormat("subcribe excpetion 2:{0}", e));subject.OnNext(123);subject.OnError(new Exception("no this vlaue"));

IgnoreElements

不发射任何数据，只发射Observable的终⽌通知

操作符抑制原始 Observable 发射的所有数据，只允许它的终⽌通知（onError 或 onCompleted）通过

        var subject = new Subject<int>();var noElements = subject.IgnoreElements();subject.Subscribe(number => Debug.LogFormat("subject.OnNext({0})", number), () => Debug.Log("subject.OnCompleted()"));noElements.Subscribe(number => Debug.LogFormat("noElements.OnNext({0})", number), () => Debug.Log("noElements.OnCompleted()"));subject.OnNext(1);subject.OnNext(2);subject.OnNext(3);subject.OnCompleted();

DistinctUntilChanged

过掉和之前为相同的

        var state = "runState";Observable.EveryUpdate().DistinctUntilChanged(_ => state).Subscribe(stateName => Debug.LogFormat("on state changed:{0}", state));Observable.ReturnUnit().Delay(TimeSpan.FromSeconds(1.0f)).Do(_ => state = "jumpState").Delay(TimeSpan.FromSeconds(1.0f)).Do(_ => state = "idleState").Subscribe();var subject = new Subject<int>();var distinct = subject.DistinctUntilChanged();subject.Subscribe(i => Debug.LogFormat("{0}", i),() => Debug.LogFormat("subject.OnCompleted()"));distinct.Subscribe(i => Debug.LogFormat("distinct.OnNext({0})", i),() => Debug.LogFormat("distinct.OnCompleted()"));subject.OnNext(1);subject.OnNext(2);subject.OnNext(3);subject.OnNext(1);subject.OnNext(1);subject.OnNext(4);subject.OnCompleted();

Concurrency:并发

Amb

给定两个或多个 Observable 时，它只发射最先发射数据或通知的那个 Observable 的所有数据

        Observable.Amb(Observable.Timer(TimeSpan.FromSeconds(4.0f)).Select(_ => "4 sec"),Observable.Timer(TimeSpan.FromSeconds(3.0f)).Select(_ => "3 sec"),Observable.Timer(TimeSpan.FromSeconds(1.0f)).Select(_ => "1 sec"),Observable.Timer(TimeSpan.FromSeconds(7.0f)).Select(_ => "7 sec"),Observable.Timer(TimeSpan.FromSeconds(9.0f)).Select(_ => "9 sec")).Subscribe(observableName => { Debug.Log(observableName); });

Events:事件

FromEvent

将其它种类的对象和数据类型转换为Observable

        Observable.EveryUpdate().Where(_ => Input.GetMouseButtonDown(0)).Subscribe(_ => mOnMouseDownEvent.Invoke());Observable.FromEvent(mouseDownEvent => mOnMouseDownEvent += mouseDownEvent,mouseDownEvent => mOnMouseDownEvent -= mOnMouseDownEvent).Subscribe(_ => Debug.Log("mouse clicked"));

Bindings:绑定

Publish

将普通的 Observable 转换为可连接的 Observable

可连接的 Observable (connectable Observable)与普通的 Observable 差不多，不过它并不会在被订阅时开始发射数据，⽽是直到使⽤了Connect操作符时才会开始。⽤这种⽅法，你可以在任何时候让⼀个Observable 开始发射数据

        var unShared = Observable.Range(1, 3);unShared.Subscribe(number => Debug.LogFormat("unShared Observable #1:{0}", number));unShared.Subscribe(number => Debug.LogFormat("unShared Observable #2:{0}", number));var shared = unShared.Publish();shared.Subscribe(number => Debug.LogFormat("shared Observable #1:{0}", number));shared.Subscribe(number => Debug.LogFormat("shared Observable #2:{0}", number));Observable.Timer(TimeSpan.FromSeconds(3.0f)).Subscribe(_ => shared.Connect());

RefCount

让⼀个可连接的 Observable ⾏为像普通的Observable

        var oneSecondPeriod = TimeSpan.FromSeconds(1.0f);var refCountObservable = Observable.Interval(oneSecondPeriod).Do(l => Debug.LogFormat("Publishing {0}", l)).Publish().RefCount();var subscription1 = refCountObservable.Subscribe(l => Debug.LogFormat("subscription #1 : {0}", l));yield return new WaitForSeconds(5.0f);var subscription2 = refCountObservable.Subscribe(l => Debug.LogFormat("subscription #2  : {0}", l));yield return new WaitForSeconds(5.0f);subscription1.Dispose();yield return new WaitForSeconds(1.0f);subscription2.Dispose();

Replay

保证所有的观察者收到相同的数据序列，即使它们在Observable开始发射数据之后才订阅

        var replayObservable = Observable.Interval(TimeSpan.FromSeconds(1.0f)).Do(l => Debug.LogFormat("Observable:{0}", l)).Replay();replayObservable.Subscribe(l => Debug.LogFormat("Subscription #1:{0}", l));replayObservable.Connect();Observable.Timer(TimeSpan.FromSeconds(5.0f)).Subscribe(_ => { replayObservable.Subscribe(l => Debug.LogFormat("Subscription #2:{0}", l)); });

ErrorHandling

Catch

从onError通知中恢复发射数据

拦截原始 Observable 的 onError 通知，将它替换为其它的数据项或数据序列，让产⽣的 Observable 能够正常终⽌或者根本不终⽌

        Observable.Throw<string>(new Exception("error")).Catch<string, Exception>(e =>{Debug.LogFormat("catched excpetion:{0}", e.Message);return Observable.Timer(TimeSpan.FromSeconds(1.0f)).Select(_ => "timer called");}).Subscribe(result => Debug.Log(result));

Finally

        var subject = new Subject<int>();var result = subject.Finally(() => Debug.Log("Finally action run"));result.Subscribe(number => Debug.LogFormat("OnNext({0});", number), () => Debug.Log("OnCompleted"));subject.OnNext(1);subject.OnNext(2);subject.OnNext(3);subject.OnCompleted();

DelaySubscription

        Debug.Log(Time.time);Observable.ReturnUnit().DelaySubscription(TimeSpan.FromSeconds(1.0f)).Subscribe(_ => Debug.Log(Time.time));

PairWise

        Observable.Range(0, 3).Pairwise().Subscribe(pair => Debug.LogFormat("{0}:{1}", pair.Previous, pair.Current));