Observable.Generate 在 RxJava 中?

问题描述

.NET Reactive Extensions 有一个使用 corecursion 生成序列的简洁方法，称为 Observable.Generate.

.NET Reactive Extensions has a neat method to generate sequences using corecursion which is called Observable.Generate.

RxJava 中是否有允许通过 corecursion 生成数据的类似方法?如果不是，是否可以基于现有方法实现?

Is there analogues method in RxJava that allows data generation via corecursion? If not, could it be implemented based on existing methods?

推荐答案

这不是完全匹配，但我们有 SyncOnSubscribe(和 AsyncOnSubscriber)可以生成值，例如例子:

It's not an exact match but we have SyncOnSubscribe (and AsyncOnSubscriber) that can generate values, for example:

@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();
}

这篇关于Observable.Generate 在 RxJava 中?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！