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

Observable.concatMapCompletable介绍

[英]Maps each element of the upstream Observable into CompletableSources, subscribes to them one at a time in order and waits until the upstream and all CompletableSources complete.

Scheduler: concatMapCompletable does not operate by default on a particular Scheduler.
[中]将上游可观测的每个元素映射到CompletableSources中，按顺序一次订阅一个元素，并等待上游和所有CompletableSources完成。
调度程序：默认情况下，concatMapCompletable不会在特定调度程序上运行。

代码示例

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

/**
 * Maps each element of the upstream Observable into CompletableSources, subscribes to them one at a time in
 * order and waits until the upstream and all CompletableSources complete.
 * <p>
 * <img width="640" height="505" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/concatMapCompletable.o.png" alt="">
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code concatMapCompletable} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 * <p>History: 2.1.6 - experimental
 * @param mapper
 *            a function that, when applied to an item emitted by the source ObservableSource, returns a CompletableSource
 * @return a Completable that signals {@code onComplete} when the upstream and all CompletableSources complete
 * @since 2.2
 */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Completable concatMapCompletable(Function<? super T, ? extends CompletableSource> mapper) {
  return concatMapCompletable(mapper, 2);
}

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

@Override
  public Completable apply(Observable<Object> f)
      throws Exception {
    return f.concatMapCompletable(
        Functions.justFunction(Completable.complete()));
  }
}

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

@Test
public void mapperCrash() {
  Observable.just(1)
  .concatMapCompletable(new Function<Integer, CompletableSource>() {
    @Override
    public CompletableSource apply(Integer v) throws Exception {
      throw new TestException();
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void fusedPollThrows() {
  Observable.just(1)
  .map(new Function<Integer, Integer>() {
    @Override
    public Integer apply(Integer v) throws Exception {
      throw new TestException();
    }
  })
  .concatMapCompletable(completableComplete())
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void dispose() {
  TestHelper.checkDisposed(Observable.<Integer>just(1).hide()
  .concatMapCompletable(completableError()));
}

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

@Test
public void mapperCrashHidden() {
  Observable.just(1).hide()
  .concatMapCompletable(new Function<Integer, CompletableSource>() {
    @Override
    public CompletableSource apply(Integer v) throws Exception {
      throw new TestException();
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void innerError() {
  Observable.<Integer>just(1).hide()
  .concatMapCompletable(completableError())
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void mapperThrows() {
  Observable.just(1).hide()
  .concatMapCompletable(completableThrows())
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void innerError() {
  Observable.just(1)
  .concatMapCompletable(Functions.justFunction(Completable.error(new TestException())))
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void simple2() {
  final AtomicInteger counter = new AtomicInteger();
  Observable.range(1, 5)
  .concatMapCompletable(Functions.justFunction(Completable.fromAction(new Action() {
    @Override
    public void run() throws Exception {
      counter.incrementAndGet();
    }
  })))
  .test()
  .assertResult();
  assertEquals(5, counter.get());
}

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

@Test
public void disposed() {
  TestHelper.checkDisposed(
      Observable.never()
      .concatMapCompletable(
          Functions.justFunction(Completable.complete()))
  );
}

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

@Test
public void simple() {
  Observable.range(1, 5)
  .concatMapCompletable(Functions.justFunction(Completable.complete()))
  .test()
  .assertResult();
}

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

@Test
public void mainError() {
  Observable.<Integer>error(new TestException())
  .concatMapCompletable(completableComplete())
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void fusionRejected() {
  final CompletableSubject cs = CompletableSubject.create();
  TestHelper.rejectObservableFusion()
  .concatMapCompletable(
      Functions.justFunction(cs)
  )
  .test()
  .assertEmpty();
}

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

@Test
public void simpleLongPrefetch() {
  Observable.range(1, 1024)
  .concatMapCompletable(Functions.justFunction(Completable.complete()), 32)
  .test()
  .assertResult();
}

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

@Test
public void emptyScalarSource() {
  final CompletableSubject cs = CompletableSubject.create();
  Observable.empty()
  .concatMapCompletable(Functions.justFunction(cs))
  .test()
  .assertResult();
}

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

@Test
  public void justScalarSource() {
    final CompletableSubject cs = CompletableSubject.create();

    TestObserver<Void> to = Observable.just(1)
    .concatMapCompletable(Functions.justFunction(cs))
    .test();

    to.assertEmpty();

    assertTrue(cs.hasObservers());

    cs.onComplete();

    to.assertResult();
  }
}

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

@Test
public void mainError() {
  Observable.<Integer>error(new TestException())
  .concatMapCompletable(Functions.justFunction(Completable.complete()))
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void asyncFused() {
  final PublishSubject<Integer> ps = PublishSubject.create();
  final CompletableSubject cs = CompletableSubject.create();
  final TestObserver<Void> to = ps.observeOn(ImmediateThinScheduler.INSTANCE)
  .concatMapCompletable(
      Functions.justFunction(cs)
  )
  .test();
  ps.onNext(1);
  ps.onComplete();
  cs.onComplete();
  to.assertResult();
}

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

@Test
public void notFused() throws Exception {
  UnicastSubject<Integer> us = UnicastSubject.create();
  TestObserver<Void> to = us.hide().concatMapCompletable(completableComplete(), 2).test();
  us.onNext(1);
  us.onNext(2);
  us.onComplete();
  to.assertComplete();
  to.assertValueCount(0);
  to.assertNoErrors();
}