操作符(Operators)
Rx 的操作符能够操作(创建/转换/组合) Observable。
Creating Observables
本文主题为创建/生成 Observable 的操作符。
这里的 Observable 实质上是可观察的数据流。
RxJava操作符(一)Creating Observables
Creation Operators
公共代码
- RxJava
fun <T> Observable<T>.dump() =
this.subscribe({ println("onNext: $it") },
{ println("onError: $it: ${it.message}") },
{ println("onComplete") })
fun <T> Observable<T>.dump(name: String) =
this.subscribe({ println("$name: onNext: $it") },
{ println("$name: onError: $it: ${it.message}") },
{ println("$name: onComplete") })
Create / Generate
ReactiveX - Create operator
Reactive Extensions再入門 その3「IObservableのファクトリメソッド」
Reactive Extensions再入門 その4「Timer系のファクトリメソッド」
Create 通过调用 Observer 的 3 个方法 OnNext, OnError, OnComplete 来创建数据流。
Generate 通过模拟 for 循环来生成数据流。
Generate 带有如下参数
- 初始状态
- 终止条件
- 一个能根据当前状态产生下一个状态的函数
- 一个能把状态转换为输出结果的函数
- (可选)一个指定状态产生时间的函数
- RxNET
var ob = Observable.Create<string>(observer =>
{
var timer = new Timer();
timer.Interval = 1000;
timer.Elapsed += (s, e) => observer.OnNext("tick");
timer.Elapsed += (s, e) => Console.WriteLine(e.SignalTime);
timer.Start();
return timer;
});
using (ob.Subscribe(Console.WriteLine))
Console.ReadLine();
/*
tick
2018/07/18 10:32:09
tick
2018/07/18 10:32:10
...
*/
var source = Observable.Generate(0, i => i < 10, i => ++i, i => i * i);
var subscription = source.Subscribe(
i => Console.WriteLine("OnNext({0})", i),
ex => Console.WriteLine("OnError({0})", ex.Message),
() => Console.WriteLine("Completed()"));
/*
OnNext(0)
OnNext(1)
OnNext(4)
OnNext(9)
OnNext(16)
OnNext(25)
OnNext(36)
OnNext(49)
OnNext(64)
OnNext(81)
Completed()
*/
var source = Observable.Generate(
0,
i => i < 10,
i => ++i,
i => i * i,
i => TimeSpan.FromMilliseconds(i * 100));
var subscription = source.Subscribe(
i => Console.WriteLine("OnNext({0})", i),
ex => Console.WriteLine("OnError({0})", ex.Message),
() => Console.WriteLine("Completed()"));
Console.ReadLine();
/*
OnNext(0)
OnNext(1)
OnNext(4)
OnNext(9)
OnNext(16)
OnNext(25)
OnNext(36)
OnNext(49)
OnNext(64)
OnNext(81)
Completed()
*/
- RxJava
val values = Observable.create<String> { o ->
o.onNext("Hello")
o.onComplete()
}
values.dump()
/*
onNext: Hello
onComplete
*/
val values = Observable.generate<Int, Int>( Callable { 0 }, BiFunction { i, o ->
if (i < 10) {
o.onNext(i * i); i + 1
} else {
o.onComplete(); i
}
}, Consumer { i -> println(i) } )
values.dump()
/*
onNext: 0
onNext: 1
onNext: 4
onNext: 9
...
onNext: 81
onComplete
10
*/
- RxSwift
let disposeBag = DisposeBag()
let myJust = { (element: String) -> Observable<String> in
return Observable.create { observer in
observer.on(.next(element))
observer.on(.completed)
return Disposables.create()
}
}
myJust("