io.reactivex.Observable.concatMapSingle()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(7.9k)|赞(0)|评价(0)|浏览(167)

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

Observable.concatMapSingle介绍

[英]Maps the upstream items into SingleSources and subscribes to them one after the other succeeds, emits their success values or terminates immediately if either this Observable or the current inner SingleSource fail.

Scheduler: concatMapSingle does not operate by default on a particular Scheduler.
[中]将上游项映射到单一源中，并一个接一个地订阅它们，发出它们的成功值，或者在该可观察项或当前内部单一源失败时立即终止。
调度程序：默认情况下，concatMapSingle不会在特定调度程序上运行。

代码示例

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

  1. /**
  2.  * Maps the upstream items into {@link SingleSource}s and subscribes to them one after the
  3.  * other succeeds, emits their success values or terminates immediately if
  4.  * either this {@code Observable} or the current inner {@code SingleSource} fail.
  5.  * <p>
  6.  * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/concatMap.png" alt="">
  7.  * <dl>
  8.  *  <dt><b>Scheduler:</b></dt>
  9.  *  <dd>{@code concatMapSingle} does not operate by default on a particular {@link Scheduler}.</dd>
  10.  * </dl>
  11.  * <p>History: 2.1.11 - experimental
  12.  * @param <R> the result type of the inner {@code SingleSource}s
  13.  * @param mapper the function called with the upstream item and should return
  14.  *               a {@code SingleSource} to become the next source to
  15.  *               be subscribed to
  16.  * @return a new Observable instance
  17.  * @see #concatMapSingleDelayError(Function)
  18.  * @see #concatMapSingle(Function, int)
  19.  * @since 2.2
  20.  */
  21. @CheckReturnValue
  22. @SchedulerSupport(SchedulerSupport.NONE)
  23. public final <R> Observable<R> concatMapSingle(Function<? super T, ? extends SingleSource<? extends R>> mapper) {
  24.   return concatMapSingle(mapper, 2);
  25. }

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

  1. @Test
  2. public void mapperCrashScalar() {
  3.   TestObserver<Object> to = Observable.just(1)
  4.   .concatMapSingle(new Function<Integer, SingleSource<? extends Object>>() {
  5.     @Override
  6.     public SingleSource<? extends Object> apply(Integer v)
  7.         throws Exception {
  8.           throw new TestException();
  9.         }
  10.   })
  11.   .test();
  12.   to.assertFailure(TestException.class);
  13. }

代码示例来源：origin: redisson/redisson

  1. /**
  2.  * Maps the upstream items into {@link SingleSource}s and subscribes to them one after the
  3.  * other succeeds, emits their success values or terminates immediately if
  4.  * either this {@code Observable} or the current inner {@code SingleSource} fail.
  5.  * <p>
  6.  * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/concatMap.png" alt="">
  7.  * <dl>
  8.  *  <dt><b>Scheduler:</b></dt>
  9.  *  <dd>{@code concatMapSingle} does not operate by default on a particular {@link Scheduler}.</dd>
  10.  * </dl>
  11.  * @param <R> the result type of the inner {@code SingleSource}s
  12.  * @param mapper the function called with the upstream item and should return
  13.  *               a {@code SingleSource} to become the next source to
  14.  *               be subscribed to
  15.  * @return a new Observable instance
  16.  * @since 2.1.11 - experimental
  17.  * @see #concatMapSingleDelayError(Function)
  18.  * @see #concatMapSingle(Function, int)
  19.  */
  20. @CheckReturnValue
  21. @SchedulerSupport(SchedulerSupport.NONE)
  22. @Experimental
  23. public final <R> Observable<R> concatMapSingle(Function<? super T, ? extends SingleSource<? extends R>> mapper) {
  24.   return concatMapSingle(mapper, 2);
  25. }

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

  1. @Test
  2. public void disposed() {
  3.   TestHelper.checkDisposed(Observable.just(1).hide()
  4.       .concatMapSingle(Functions.justFunction(Single.never()))
  5.   );
  6. }

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

  1. @Test
  2. public void innerError() {
  3.   Observable.just(1)
  4.   .concatMapSingle(Functions.justFunction(Single.error(new TestException())))
  5.   .test()
  6.   .assertFailure(TestException.class);
  7. }

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

  1. @Test
  2. public void simple() {
  3.   Observable.range(1, 5)
  4.   .concatMapSingle(new Function<Integer, SingleSource<Integer>>() {
  5.     @Override
  6.     public SingleSource<Integer> apply(Integer v)
  7.         throws Exception {
  8.       return Single.just(v);
  9.     }
  10.   })
  11.   .test()
  12.   .assertResult(1, 2, 3, 4, 5);
  13. }

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

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

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

  1. @Test
  2. public void simpleLong() {
  3.   Observable.range(1, 1024)
  4.   .concatMapSingle(new Function<Integer, SingleSource<Integer>>() {
  5.     @Override
  6.     public SingleSource<Integer> apply(Integer v)
  7.         throws Exception {
  8.       return Single.just(v);
  9.     }
  10.   }, 32)
  11.   .test()
  12.   .assertValueCount(1024)
  13.   .assertNoErrors()
  14.   .assertComplete();
  15. }

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

  1. @Test
  2. public void mainError() {
  3.   Observable.error(new TestException())
  4.   .concatMapSingle(Functions.justFunction(Single.just(1)))
  5.   .test()
  6.   .assertFailure(TestException.class);
  7. }

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

  1. @Test
  2. public void innerSuccessDisposeRace() {
  3.   for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
  4.     final SingleSubject<Integer> ss = SingleSubject.create();
  5.     final TestObserver<Integer> to = Observable.just(1)
  6.         .hide()
  7.         .concatMapSingle(Functions.justFunction(ss))
  8.         .test();
  9.     Runnable r1 = new Runnable() {
  10.       @Override
  11.       public void run() {
  12.         ss.onSuccess(1);
  13.       }
  14.     };
  15.     Runnable r2 = new Runnable() {
  16.       @Override
  17.       public void run() {
  18.         to.dispose();
  19.       }
  20.     };
  21.     TestHelper.race(r1, r2);
  22.     to.assertNoErrors();
  23.   }
  24. }

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

  1. @Test
  2. public void scalarEmptySource() {
  3.   SingleSubject<Integer> ss = SingleSubject.create();
  4.   Observable.empty()
  5.   .concatMapSingle(Functions.justFunction(ss))
  6.   .test()
  7.   .assertResult();
  8.   assertFalse(ss.hasObservers());
  9. }

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

  1. @Test
  2. public void cancel() {
  3.   Observable.range(1, 5).concatWith(Observable.<Integer>never())
  4.   .concatMapSingle(new Function<Integer, SingleSource<Integer>>() {
  5.     @Override
  6.     public SingleSource<Integer> apply(Integer v)
  7.         throws Exception {
  8.       return Single.just(v);
  9.     }
  10.   })
  11.   .test()
  12.   .assertValues(1, 2, 3, 4, 5)
  13.   .assertNoErrors()
  14.   .assertNotComplete()
  15.   .cancel();
  16. }

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

  1. @Test
  2. public void checkUnboundedInnerQueue() {
  3.   SingleSubject<Integer> ss = SingleSubject.create();
  4.   @SuppressWarnings("unchecked")
  5.   TestObserver<Integer> to = Observable
  6.       .fromArray(ss, Single.just(2), Single.just(3), Single.just(4))
  7.       .concatMapSingle(Functions.<Single<Integer>>identity(), 2)
  8.       .test();
  9.   to.assertEmpty();
  10.   ss.onSuccess(1);
  11.   to.assertResult(1, 2, 3, 4);
  12. }

代码示例来源：origin: forkachild/reel-search-android

  1. @Override
  2. protected void onStart() {
  3.   super.onStart();
  4.   mDisposable.add(mDictionaryManager.loadDictionary()
  5.       .doOnSubscribe(d -> {
  6.         mBinding.txtQuery.setEnabled(false);
  7.         mBinding.txtQuery.setHint("Loading dictionary");
  8.       })
  9.       .doOnComplete(() -> {
  10.         mBinding.txtQuery.setEnabled(true);
  11.         mBinding.txtQuery.setHint("Start typing");
  12.       })
  13.       .subscribe(() -> {
  14.       }, Throwable::printStackTrace));
  15.   mDisposable.add(RxUtils.onTextChange(mBinding.txtQuery)
  16.       .filter(in -> mDictionaryManager.isLoaded())
  17.       .concatMapSingle(in -> mDictionaryManager.query(in))
  18.       .doOnNext(in -> mBinding.btnSelect.setEnabled(!in.isEmpty()))
  19.       .subscribe(mAdapter::setData, Throwable::printStackTrace));
  20. }

代码示例来源：origin: akarnokd/akarnokd-misc

  1. @Test
  2.   public void test() {
  3.    TestScheduler testScheduler = new TestScheduler();
  5.    final Single<List<Integer>> first = Single.timer(2, TimeUnit.SECONDS, testScheduler)
  6.        .map(u -> Arrays.asList(1, 2, 3));
  7.    final Single<List<Integer>> second = Single.just(Collections.emptyList());
  8.    final Single<List<Integer>> third = Single.just(Collections.singletonList(4));
  9.    final Single<List<Integer>> fourth = Single.just(Collections.singletonList(5));
  11.    Single<List<Integer>> subject = Observable
  12.     .fromIterable(Arrays.asList(first, second, third, fourth))
  13.     .concatMapSingle(single -> single)
  14.     .reduce(new ArrayList<>(), (seed, items) -> {
  15.      seed.addAll(items);
  16.      return seed;
  17.     });
  19.     TestObserver<List<Integer>> testObserver = subject.test();
  20.     testScheduler.advanceTimeBy(3, TimeUnit.SECONDS);
  22.     System.out.println(testObserver.values());
  23.     testObserver.assertValue(list -> list.equals(Arrays.asList(1, 2, 3, 4, 5))); 
  24.     // 5 is currently missing ; fourth was never subscribed in the first place
  25.   }
  26. }

内容来源于网络，如有侵权，请联系作者删除！

Observable

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Observable类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com