io.reactivex.Flowable.flatMapMaybe()方法的使用及代码示例

x33g5p2x  于2022-01-19 转载在 其他  
字(7.4k)|赞(0)|评价(0)|浏览(207)

本文整理了Java中io.reactivex.Flowable.flatMapMaybe()方法的一些代码示例,展示了Flowable.flatMapMaybe()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台,是从一些精选项目中提取出来的代码,具有较强的参考意义,能在一定程度帮忙到你。Flowable.flatMapMaybe()方法的具体详情如下:
包路径:io.reactivex.Flowable
类名称:Flowable
方法名:flatMapMaybe

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

  1. @Override
  2. public Flowable<Integer> apply(Flowable<Object> f) throws Exception {
  3. return f.flatMapMaybe(Functions.justFunction(Maybe.just(2)));
  4. }
  5. });

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normal() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v);
  8. }
  9. })
  10. .test()
  11. .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  12. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalEmpty() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.empty();
  8. }
  9. })
  10. .test()
  11. .assertResult();
  12. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalDelayError() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v);
  8. }
  9. }, true, Integer.MAX_VALUE)
  10. .test()
  11. .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  12. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalBackpressured() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v);
  8. }
  9. })
  10. .rebatchRequests(1)
  11. .test()
  12. .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  13. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalMaxConcurrent1Backpressured() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v);
  8. }
  9. }, false, 1)
  10. .rebatchRequests(1)
  11. .test()
  12. .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  13. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalAsyncMaxConcurrency1() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v).subscribeOn(Schedulers.computation());
  8. }
  9. }, false, 1)
  10. .test()
  11. .awaitDone(5, TimeUnit.SECONDS)
  12. .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  13. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalMaxConcurrent2Backpressured() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v);
  8. }
  9. }, false, 2)
  10. .rebatchRequests(1)
  11. .test()
  12. .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
  13. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void take() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v);
  8. }
  9. })
  10. .take(2)
  11. .test()
  12. .assertResult(1, 2);
  13. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void asyncFlattenNone() {
  3. Flowable.range(1, 1000)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.<Integer>empty().subscribeOn(Schedulers.computation());
  8. }
  9. })
  10. .take(500)
  11. .test()
  12. .awaitDone(5, TimeUnit.SECONDS)
  13. .assertResult();
  14. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void asyncFlattenNoneMaxConcurrency() {
  3. Flowable.range(1, 1000)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.<Integer>empty().subscribeOn(Schedulers.computation());
  8. }
  9. }, false, 128)
  10. .take(500)
  11. .test()
  12. .awaitDone(5, TimeUnit.SECONDS)
  13. .assertResult();
  14. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void asyncFlattenErrorMaxConcurrency() {
  3. Flowable.range(1, 1000)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.<Integer>error(new TestException()).subscribeOn(Schedulers.computation());
  8. }
  9. }, true, 128)
  10. .take(500)
  11. .test()
  12. .awaitDone(5, TimeUnit.SECONDS)
  13. .assertFailure(CompositeException.class);
  14. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalAsync() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v).subscribeOn(Schedulers.computation());
  8. }
  9. })
  10. .test()
  11. .awaitDone(5, TimeUnit.SECONDS)
  12. .assertSubscribed()
  13. .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  14. .assertNoErrors()
  15. .assertComplete();
  16. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void asyncFlattenErrorMaxConcurrency() {
  3. Flowable.range(1, 1000)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.<Integer>error(new TestException()).subscribeOn(Schedulers.computation());
  8. }
  9. }, true, 128)
  10. .take(500)
  11. .test()
  12. .awaitDone(5, TimeUnit.SECONDS)
  13. .assertFailure(CompositeException.class);
  14. }
  15. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalAsyncMaxConcurrency() {
  3. Flowable.range(1, 10)
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.just(v).subscribeOn(Schedulers.computation());
  8. }
  9. }, false, 3)
  10. .test()
  11. .awaitDone(5, TimeUnit.SECONDS)
  12. .assertSubscribed()
  13. .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  14. .assertNoErrors()
  15. .assertComplete();
  16. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void error() {
  3. Flowable.just(1)
  4. .flatMapMaybe(Functions.justFunction(Maybe.<Integer>error(new TestException())))
  5. .test(0L)
  6. .assertFailure(TestException.class);
  7. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void errorDelayed() {
  3. Flowable.just(1)
  4. .flatMapMaybe(Functions.justFunction(Maybe.<Integer>error(new TestException())), true, 16)
  5. .test(0L)
  6. .assertFailure(TestException.class);
  7. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void backpressure() {
  3. TestSubscriber<Integer> ts = Flowable.just(1)
  4. .flatMapMaybe(Functions.justFunction(Maybe.just(2)))
  5. .test(0L)
  6. .assertEmpty();
  7. ts.request(1);
  8. ts.assertResult(2);
  9. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void normalDelayErrorAll() {
  3. TestSubscriber<Integer> ts = Flowable.range(1, 10).concatWith(Flowable.<Integer>error(new TestException()))
  4. .flatMapMaybe(new Function<Integer, MaybeSource<Integer>>() {
  5. @Override
  6. public MaybeSource<Integer> apply(Integer v) throws Exception {
  7. return Maybe.error(new TestException());
  8. }
  9. }, true, Integer.MAX_VALUE)
  10. .test()
  11. .assertFailure(CompositeException.class);
  12. List<Throwable> errors = TestHelper.compositeList(ts.errors().get(0));
  13. for (int i = 0; i < 11; i++) {
  14. TestHelper.assertError(errors, i, TestException.class);
  15. }
  16. }

代码示例来源:origin: ReactiveX/RxJava

  1. @Test
  2. public void innerSuccessCompletesAfterMain() {
  3. PublishProcessor<Integer> pp = PublishProcessor.create();
  4. TestSubscriber<Integer> ts = Flowable.just(1).flatMapMaybe(Functions.justFunction(pp.singleElement()))
  5. .test();
  6. pp.onNext(2);
  7. pp.onComplete();
  8. ts
  9. .assertResult(2);
  10. }

相关文章

Flowable类方法