本文整理了Java中io.reactivex.Flowable.flatMapMaybe()
方法的一些代码示例,展示了Flowable.flatMapMaybe()
的具体用法。这些代码示例主要来源于Github
/Stackoverflow
/Maven
等平台,是从一些精选项目中提取出来的代码,具有较强的参考意义,能在一定程度帮忙到你。Flowable.flatMapMaybe()
方法的具体详情如下:
包路径:io.reactivex.Flowable
类名称:Flowable
方法名:flatMapMaybe
[英]Maps each element of the upstream Flowable into MaybeSources, subscribes to all of them and merges their onSuccess values, in no particular order, into a single Flowable sequence. Backpressure: The operator consumes the upstream in an unbounded manner. Scheduler: flatMapMaybe does not operate by default on a particular Scheduler.
[中]将上游可流动的每个元素映射到MaybeSources,订阅所有这些元素,并将其onSuccess值(无特定顺序)合并到单个可流动序列中。背压:操作员以无限制的方式消耗上游压力。调度程序:默认情况下,FlatMap可能不会在特定调度程序上运行。
代码示例来源:origin: ReactiveX/RxJava
@Override
public Flowable<Integer> apply(Flowable<Object> f) throws Exception {
return f.flatMapMaybe(Functions.justFunction(Maybe.just(2)));
}
});
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normal() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v);
}
})
.test()
.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalEmpty() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.empty();
}
})
.test()
.assertResult();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalDelayError() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v);
}
}, true, Integer.MAX_VALUE)
.test()
.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalBackpressured() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v);
}
})
.rebatchRequests(1)
.test()
.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalMaxConcurrent1Backpressured() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v);
}
}, false, 1)
.rebatchRequests(1)
.test()
.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalAsyncMaxConcurrency1() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v).subscribeOn(Schedulers.computation());
}
}, false, 1)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalMaxConcurrent2Backpressured() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v);
}
}, false, 2)
.rebatchRequests(1)
.test()
.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void take() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v);
}
})
.take(2)
.test()
.assertResult(1, 2);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void asyncFlattenNone() {
Flowable.range(1, 1000)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.<Integer>empty().subscribeOn(Schedulers.computation());
}
})
.take(500)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void asyncFlattenNoneMaxConcurrency() {
Flowable.range(1, 1000)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.<Integer>empty().subscribeOn(Schedulers.computation());
}
}, false, 128)
.take(500)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void asyncFlattenErrorMaxConcurrency() {
Flowable.range(1, 1000)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.<Integer>error(new TestException()).subscribeOn(Schedulers.computation());
}
}, true, 128)
.take(500)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertFailure(CompositeException.class);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalAsync() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v).subscribeOn(Schedulers.computation());
}
})
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertSubscribed()
.assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
.assertNoErrors()
.assertComplete();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void asyncFlattenErrorMaxConcurrency() {
Flowable.range(1, 1000)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.<Integer>error(new TestException()).subscribeOn(Schedulers.computation());
}
}, true, 128)
.take(500)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertFailure(CompositeException.class);
}
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalAsyncMaxConcurrency() {
Flowable.range(1, 10)
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.just(v).subscribeOn(Schedulers.computation());
}
}, false, 3)
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertSubscribed()
.assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
.assertNoErrors()
.assertComplete();
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void error() {
Flowable.just(1)
.flatMapMaybe(Functions.justFunction(Maybe.<Integer>error(new TestException())))
.test(0L)
.assertFailure(TestException.class);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void errorDelayed() {
Flowable.just(1)
.flatMapMaybe(Functions.justFunction(Maybe.<Integer>error(new TestException())), true, 16)
.test(0L)
.assertFailure(TestException.class);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void backpressure() {
TestSubscriber<Integer> ts = Flowable.just(1)
.flatMapMaybe(Functions.justFunction(Maybe.just(2)))
.test(0L)
.assertEmpty();
ts.request(1);
ts.assertResult(2);
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void normalDelayErrorAll() {
TestSubscriber<Integer> ts = Flowable.range(1, 10).concatWith(Flowable.<Integer>error(new TestException()))
.flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
@Override
public MaybeSource<Integer> apply(Integer v) throws Exception {
return Maybe.error(new TestException());
}
}, true, Integer.MAX_VALUE)
.test()
.assertFailure(CompositeException.class);
List<Throwable> errors = TestHelper.compositeList(ts.errors().get(0));
for (int i = 0; i < 11; i++) {
TestHelper.assertError(errors, i, TestException.class);
}
}
代码示例来源:origin: ReactiveX/RxJava
@Test
public void innerSuccessCompletesAfterMain() {
PublishProcessor<Integer> pp = PublishProcessor.create();
TestSubscriber<Integer> ts = Flowable.just(1).flatMapMaybe(Functions.justFunction(pp.singleElement()))
.test();
pp.onNext(2);
pp.onComplete();
ts
.assertResult(2);
}
内容来源于网络,如有侵权,请联系作者删除!