使用一个函数从头开始创建一个Observable
你可以使用Create操作符从头开始创建一个Observable,给这个操作符传递一个接受观察者作为参数的函数,编写这个函数让它的行为表现为一个Observable--恰当的调用观察者的onNext,onError和onCompleted方法。
Create
一个形式正确的有限Observable必须尝试调用观察者的onCompleted正好一次或者它的onError正好一次,而且此后不能再调用观察者的任何其它方法。
RxJava将这个操作符实现为 create 方法。
create
建议你在传递给create方法的函数中检查观察者的isUnsubscribed状态,以便在没有观察者的时候,让你的Observable停止发射数据或者做昂贵的运算。
isUnsubscribed
示例代码:
Observable.create(new Observable.OnSubscribe<Integer>() { @Override public void call(Subscriber<? super Integer> observer) { try { if (!observer.isUnsubscribed()) { for (int i = 1; i < 5; i++) { observer.onNext(i); } observer.onCompleted(); } } catch (Exception e) { observer.onError(e); } } } ).subscribe(new Subscriber<Integer>() { @Override public void onNext(Integer item) { System.out.println("Next: " + item); } @Override public void onError(Throwable error) { System.err.println("Error: " + error.getMessage()); } @Override public void onCompleted() { System.out.println("Sequence complete."); } });
输出:
Next: 1 Next: 2 Next: 3 Next: 4 Sequence complete.
create方法默认不在任何特定的调度器上执行。
create(OnSubscribe)
Copyright© 2013-2020
All Rights Reserved 京ICP备2023019179号-8