Observable.Generate 在 RxJava 中?
Observable.Generate in RxJava?
.NET Reactive Extensions 有一种使用 corecursion 生成序列的简洁方法,称为 Observable.Generate。
RxJava 中是否有允许通过 corecursion 生成数据的类似方法?如果不是,是否可以根据现有的方法实现?
这不是完全匹配,但我们有 SyncOnSubscribe
(和 AsyncOnSubscriber
)可以生成值,例如:
@Test
public void testRange() {
final int start = 1;
final int count = 4000;
OnSubscribe<Integer> os = SyncOnSubscribe.createStateful(new Func0<Integer>(){
@Override
public Integer call() {
return start;
}},
new Func2<Integer, Observer<? super Integer>, Integer>() {
@Override
public Integer call(Integer state, Observer<? super Integer> subscriber) {
subscriber.onNext(state);
if (state == count) {
subscriber.onCompleted();
}
return state + 1;
}
});
@SuppressWarnings("unchecked")
Observer<Object> o = mock(Observer.class);
InOrder inOrder = inOrder(o);
Observable.create(os).subscribe(o);
verify(o, never()).onError(any(TestException.class));
inOrder.verify(o, times(count)).onNext(any(Integer.class));
inOrder.verify(o).onCompleted();
inOrder.verifyNoMoreInteractions();
}
.NET Reactive Extensions 有一种使用 corecursion 生成序列的简洁方法,称为 Observable.Generate。
RxJava 中是否有允许通过 corecursion 生成数据的类似方法?如果不是,是否可以根据现有的方法实现?
这不是完全匹配,但我们有 SyncOnSubscribe
(和 AsyncOnSubscriber
)可以生成值,例如:
@Test
public void testRange() {
final int start = 1;
final int count = 4000;
OnSubscribe<Integer> os = SyncOnSubscribe.createStateful(new Func0<Integer>(){
@Override
public Integer call() {
return start;
}},
new Func2<Integer, Observer<? super Integer>, Integer>() {
@Override
public Integer call(Integer state, Observer<? super Integer> subscriber) {
subscriber.onNext(state);
if (state == count) {
subscriber.onCompleted();
}
return state + 1;
}
});
@SuppressWarnings("unchecked")
Observer<Object> o = mock(Observer.class);
InOrder inOrder = inOrder(o);
Observable.create(os).subscribe(o);
verify(o, never()).onError(any(TestException.class));
inOrder.verify(o, times(count)).onNext(any(Integer.class));
inOrder.verify(o).onCompleted();
inOrder.verifyNoMoreInteractions();
}