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

Flowable.concatMapIterable介绍

[英]Returns a Flowable that concatenate each item emitted by the source Publisher with the values in an Iterable corresponding to that item that is generated by a selector. Backpressure: The operator honors backpressure from downstream. The source Publishers is expected to honor backpressure as well. If the source Publisher violates the rule, the operator will signal a MissingBackpressureException. Scheduler: concatMapIterable does not operate by default on a particular Scheduler.
[中]返回一个Flowable，它将源发布服务器发出的每个项与选择器生成的该项对应的Iterable中的值连接起来。背压：操作员接受来自下游的背压。预计出版者也会接受背压。如果源发布服务器违反规则，操作员将发出MissingBackpressureException信号。调度程序：默认情况下，ConcatMapiteTable不会在特定调度程序上运行。

代码示例

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

@Override
  public Publisher<Integer> createPublisher(long elements) {
    return
        Flowable.range(0, (int)elements)
        .concatMapIterable(new Function<Integer, Iterable<Integer>>() {
          @Override
          public Iterable<Integer> apply(Integer v) throws Exception {
            return Collections.singletonList(v);
          }
        })
      ;
  }
}

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

@Test(expected = NullPointerException.class)
public void concatMapIterableNull() {
  just1.concatMapIterable(null);
}

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

/**
 * Returns a Flowable that concatenate each item emitted by the source Publisher with the values in an
 * Iterable corresponding to that item that is generated by a selector.
 *
 * <dl>
 *  <dt><b>Backpressure:</b></dt>
 *  <dd>The operator honors backpressure from downstream. The source {@code Publisher}s is
 *  expected to honor backpressure as well. If the source {@code Publisher} violates the rule, the operator will
 *  signal a {@code MissingBackpressureException}.</dd>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code concatMapIterable} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 *
 * @param <U>
 *            the type of item emitted by the resulting Publisher
 * @param mapper
 *            a function that returns an Iterable sequence of values for when given an item emitted by the
 *            source Publisher
 * @return a Flowable that emits the results of concatenating the items emitted by the source Publisher with
 *         the values in the Iterables corresponding to those items, as generated by {@code collectionSelector}
 * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public final <U> Flowable<U> concatMapIterable(Function<? super T, ? extends Iterable<? extends U>> mapper) {
  return concatMapIterable(mapper, 2);
}

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

@Test(expected = NullPointerException.class)
public void concatMapIterableReturnNull() {
  just1.concatMapIterable(new Function<Integer, Iterable<Object>>() {
    @Override
    public Iterable<Object> apply(Integer v) {
      return null;
    }
  }).blockingSubscribe();
}

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

@Test(expected = NullPointerException.class)
public void concatMapIterableIteratorNull() {
  just1.concatMapIterable(new Function<Integer, Iterable<Object>>() {
    @Override
    public Iterable<Object> apply(Integer v) {
      return new Iterable<Object>() {
        @Override
        public Iterator<Object> iterator() {
          return null;
        }
      };
    }
  }).blockingSubscribe();
}

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

@Test
public void concatMapIterableBufferSize() {
  Flowable.just(1, 2).concatMapIterable(new Function<Integer, Iterable<Integer>>() {
    @Override
    public Iterable<Integer> apply(Integer v) throws Exception {
      return Arrays.asList(1, 2, 3, 4, 5);
    }
  }, 1)
  .test()
  .assertResult(1, 2, 3, 4, 5, 1, 2, 3, 4, 5);
}

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

@Test
public void longRunning() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  int n = 1000 * 1000;
  Flowable.range(1, n).concatMapIterable(mapper)
  .subscribe(ts);
  ts.assertValueCount(n * 2);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void normal() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.range(1, 5).concatMapIterable(mapper)
  .subscribe(ts);
  ts.assertValues(1, 2, 2, 3, 3, 4, 4, 5, 5, 6);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void just() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.just(1).concatMapIterable(mapper)
  .subscribe(ts);
  ts.assertValues(1, 2);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void empty() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.<Integer>empty().concatMapIterable(mapper)
  .subscribe(ts);
  ts.assertNoValues();
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void mixture() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.range(0, 1000)
  .concatMapIterable(new Function<Integer, Iterable<Integer>>() {
    @Override
    public Iterable<Integer> apply(Integer v) {
      return (v % 2) == 0 ? Collections.singleton(1) : Collections.<Integer>emptySet();
    }
  })
  .subscribe(ts);
  ts.assertValueCount(500);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void justHidden() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.just(1).hide().concatMapIterable(mapper)
  .subscribe(ts);
  ts.assertValues(1, 2);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void emptyInnerThenSingleBackpressured() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1);
  Flowable.range(1, 2)
  .concatMapIterable(new Function<Integer, Iterable<Integer>>() {
    @Override
    public Iterable<Integer> apply(Integer v) {
      return v == 2 ? Collections.singleton(1) : Collections.<Integer>emptySet();
    }
  })
  .subscribe(ts);
  ts.assertValue(1);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void manyEmptyInnerThenSingleBackpressured() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1);
  Flowable.range(1, 1000)
  .concatMapIterable(new Function<Integer, Iterable<Integer>>() {
    @Override
    public Iterable<Integer> apply(Integer v) {
      return v == 1000 ? Collections.singleton(1) : Collections.<Integer>emptySet();
    }
  })
  .subscribe(ts);
  ts.assertValue(1);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void asIntermediate() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  int n = 1000 * 1000;
  Flowable.range(1, n).concatMapIterable(mapper).concatMap(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) {
      return Flowable.just(v);
    }
  })
  .subscribe(ts);
  ts.assertValueCount(n * 2);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@Test
public void error() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.<Integer>just(1).concatWith(Flowable.<Integer>error(new TestException()))
  .concatMapIterable(mapper)
  .subscribe(ts);
  ts.assertValues(1, 2);
  ts.assertError(TestException.class);
  ts.assertNotComplete();
}

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

.concatMapIterable(new Function<Integer, Iterable<Integer>>() {
  @Override
  public Iterable<Integer> apply(Integer v) {

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

.concatMapIterable(new Function<Integer, Iterable<Integer>>() {
  @Override
  public Iterable<Integer> apply(Integer v) {

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

.concatMapIterable(new Function<Integer, Iterable<Integer>>() {
  @Override
  public Iterable<Integer> apply(Integer v) {

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

@Test
public void normalBackpressured() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0);
  Flowable.range(1, 5).concatMapIterable(mapper)
  .subscribe(ts);
  ts.assertNoValues();
  ts.assertNoErrors();
  ts.assertNotComplete();
  ts.request(1);
  ts.assertValue(1);
  ts.assertNoErrors();
  ts.assertNotComplete();
  ts.request(2);
  ts.assertValues(1, 2, 2);
  ts.assertNoErrors();
  ts.assertNotComplete();
  ts.request(7);
  ts.assertValues(1, 2, 2, 3, 3, 4, 4, 5, 5, 6);
  ts.assertNoErrors();
  ts.assertComplete();
}