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

Flowable.concatWith介绍

[英]Returns a Flowable that emits items from this Flowable and when it completes normally, the other CompletableSource is subscribed to and the returned Flowable emits its terminal events.

Backpressure: The operator does not interfere with backpressure between the current Flowable and the downstream consumer (i.e., acts as pass-through). When the operator switches to the Completable, backpressure is no longer present because Completable doesn't have items to apply backpressure to. Scheduler: concatWith does not operate by default on a particular Scheduler.
[中]返回一个从该流表中发出项的流表，当它正常完成时，订阅另一个CompletableSource，返回的流表发出其终端事件。
背压：操作员不会干扰可流动电流和下游耗电元件之间的背压（即，作为直通）。当操作员切换到Completable时，背压不再存在，因为Completable没有可应用背压的项目。调度程序：默认情况下，concatWith不会在特定调度程序上运行。

代码示例

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

@Override
  public Publisher<Integer> createPublisher(long elements) {
    return
        Flowable.range(1, Math.max(0, (int)elements - 1))
        .concatWith(Single.just((int)elements))
      ;
  }
}

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

@Override
  public Flowable<String> apply(Resource resource) {
    return Flowable.fromArray(resource.getTextFromWeb().split(" "))
        .concatWith(Flowable.<String>error(new RuntimeException()));
  }
};

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

@Test
public void skipLastTimedCustomSchedulerDelayError() {
  Flowable.just(1).concatWith(Flowable.just(2).delay(500, TimeUnit.MILLISECONDS))
  .skipLast(300, TimeUnit.MILLISECONDS, Schedulers.io(), true)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

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

@Test
public void normal() {
  final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.range(1, 5)
  .concatWith(Single.just(100))
  .subscribe(ts);
  ts.assertResult(1, 2, 3, 4, 5, 100);
}

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

@Test
public void normalNonDelayErrorOuter() {
  Flowable.range(1, 10).concatWith(Flowable.<Integer>error(new TestException()))
  .flatMapCompletable(new Function<Integer, CompletableSource>() {
    @Override
    public CompletableSource apply(Integer v) throws Exception {
      return Completable.complete();
    }
  }, false, Integer.MAX_VALUE)
  .test()
  .assertFailure(TestException.class);
}

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

@SuppressWarnings("unchecked")
@Test
public void concatIterableDelayErrorWithError() {
  Flowable.concatDelayError(
      Arrays.asList(Flowable.just(1), Flowable.just(2),
      Flowable.just(3).concatWith(Flowable.<Integer>error(new TestException())),
      Flowable.just(4)))
  .test()
  .assertFailure(TestException.class, 1, 2, 3, 4);
}

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

@Test
public void skipLastTimedDefaultScheduler() {
  Flowable.just(1).concatWith(Flowable.just(2).delay(500, TimeUnit.MILLISECONDS))
  .skipLast(300, TimeUnit.MILLISECONDS)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

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

@Test
public void delayWithTimeDelayError() throws Exception {
  Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
  .delay(100, TimeUnit.MILLISECONDS, true)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertFailure(TestException.class, 1);
}

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

@Test
public void mergeDelayError3WithError() {
  Flowable.mergeDelayError(
      Flowable.just(1),
      Flowable.just(2).concatWith(Flowable.<Integer>error(new TestException())),
      Flowable.just(3)
  )
  .test()
  .assertFailure(TestException.class, 1, 2, 3);
}

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

@Test
public void delayError() {
  TestSubscriber<Integer> ts = Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
  .onBackpressureBuffer(true)
  .test(0L)
  .assertEmpty();
  ts.request(1);
  ts.assertFailure(TestException.class, 1);
}

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

@Test
public void error() {
  new FlowablePublishMulticast<Integer, Integer>(Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException())),
      Functions.<Flowable<Integer>>identity(), 16, true)
  .test()
  .assertFailure(TestException.class, 1);
}

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

@Test
public void cancelOther() {
  CompletableSubject other = CompletableSubject.create();
  TestSubscriber<Object> ts = Flowable.empty()
      .concatWith(other)
      .test();
  assertTrue(other.hasObservers());
  ts.cancel();
  assertFalse(other.hasObservers());
}

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

@Test
public void scalarAndRangeBackpressured() {
  TestSubscriber<Integer> ts = TestSubscriber.create(0);
  Flowable.just(1).concatWith(Flowable.range(2, 3)).subscribe(ts);
  ts.assertNoValues();
  ts.request(5);
  ts.assertValues(1, 2, 3, 4);
  ts.assertComplete();
  ts.assertNoErrors();
}

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

@Test
public void rangeAndEmptyBackpressured() {
  TestSubscriber<Integer> ts = TestSubscriber.create(0);
  Flowable.range(1, 2).concatWith(Flowable.<Integer>empty()).subscribe(ts);
  ts.assertNoValues();
  ts.request(5);
  ts.assertValues(1, 2);
  ts.assertComplete();
  ts.assertNoErrors();
}

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

@Test
public void emptyAndScalarBackpressured() {
  TestSubscriber<Integer> ts = TestSubscriber.create(0);
  Flowable.<Integer>empty().concatWith(Flowable.just(1)).subscribe(ts);
  ts.assertNoValues();
  ts.request(5);
  ts.assertValue(1);
  ts.assertComplete();
  ts.assertNoErrors();
}

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

@Test
public void scalarAndEmptyBackpressured() {
  TestSubscriber<Integer> ts = TestSubscriber.create(0);
  Flowable.just(1).concatWith(Flowable.<Integer>empty()).subscribe(ts);
  ts.assertNoValues();
  ts.request(5);
  ts.assertValue(1);
  ts.assertComplete();
  ts.assertNoErrors();
}

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

@Test
public void concatObservableDelayErrorWithError() {
  Flowable.concatDelayError(
      Flowable.just(Flowable.just(1), Flowable.just(2),
      Flowable.just(3).concatWith(Flowable.<Integer>error(new TestException())),
      Flowable.just(4)))
  .test()
  .assertFailure(TestException.class, 1, 2, 3, 4);
}

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

@Test
public void concatObservableDelayErrorTillEnd() {
  Flowable.concatDelayError(
      Flowable.just(Flowable.just(1), Flowable.just(2),
      Flowable.just(3).concatWith(Flowable.<Integer>error(new TestException())),
      Flowable.just(4)), 2, true)
  .test()
  .assertFailure(TestException.class, 1, 2, 3, 4);
}

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

@Test
public void testInnerBackpressureWithAlignedBoundaries() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.range(0, Flowable.bufferSize() * 2)
      .concatWith(Flowable.range(0, Flowable.bufferSize() * 2))
      .observeOn(Schedulers.computation()) // observeOn has a backpressured RxRingBuffer
      .subscribe(ts);
  ts.awaitTerminalEvent();
  ts.assertNoErrors();
  assertEquals(Flowable.bufferSize() * 4, ts.valueCount());
}

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

@Test
public void observers() {
  PublishProcessor<Integer> pp = PublishProcessor.create();
  FlowableCache<Integer> cache = (FlowableCache<Integer>)Flowable.range(1, 5).concatWith(pp).cache();
  assertFalse(cache.hasSubscribers());
  assertEquals(0, cache.cachedEventCount());
  TestSubscriber<Integer> ts = cache.test();
  assertTrue(cache.hasSubscribers());
  assertEquals(5, cache.cachedEventCount());
  pp.onComplete();
  ts.assertResult(1, 2, 3, 4, 5);
}