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

Flowable.concatMap介绍

[英]Returns a new Flowable that emits items resulting from applying a function that you supply to each item emitted by the source Publisher, where that function returns a Publisher, and then emitting the items that result from concatenating those resulting Publishers.

Backpressure: The operator honors backpressure from downstream. Both this and the inner Publishers are expected to honor backpressure as well. If the source Publisher violates the rule, the operator will signal a MissingBackpressureException. If any of the inner Publishers doesn't honor backpressure, that may throw an IllegalStateException when that Publisher completes. Scheduler: concatMap does not operate by default on a particular Scheduler.
[中]返回一个新的可流动项，该可流动项通过将提供的函数应用于源发布服务器发出的每个项（其中该函数返回发布服务器），然后发送通过连接这些结果发布服务器而产生的项。
背压：操作员接受来自下游的背压。这两个和内部出版商预计将尊重背压以及。如果源发布服务器违反规则，操作员将发出MissingBackpressureException信号。如果任何内部发布服务器不支持backpressure，则该发布服务器完成后可能抛出非法状态异常。调度程序：默认情况下，concatMap不会在特定调度程序上运行。

代码示例

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

@Override
  public Publisher<Integer> apply(final Flowable<Integer> shared)
      throws Exception {
    return shared.take(1).concatMap(new Function<Integer, Publisher<? extends Integer>>() {
      @Override
      public Publisher<? extends Integer> apply(Integer first)
          throws Exception {
        calls.incrementAndGet();
        return transformer.apply(Flowable.just(first).concatWith(shared));
      }
    });
  }
})

代码示例来源：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 dispose() {
  TestHelper.checkDisposed(Flowable.range(1, 2)
  .concatMap(Functions.justFunction(Flowable.just(1))));
  TestHelper.checkDisposed(Flowable.range(1, 2)
  .concatMapDelayError(Functions.justFunction(Flowable.just(1))));
}

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

@Override
  public Publisher<Integer> apply(final Flowable<Integer> shared)
      throws Exception {
    return shared.take(1).concatMap(new Function<Integer, Publisher<? extends Integer>>() {
      @Override
      public Publisher<? extends Integer> apply(Integer first)
          throws Exception {
        calls.incrementAndGet();
        return transformer.apply(Flowable.just(first).concatWith(shared));
      }
    });
  }
})

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

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

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

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

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

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

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

@Override
  public Publisher<Integer> apply(Flowable<Object> f) throws Exception {
    return f.concatMap(Functions.justFunction(Flowable.just(2)));
  }
});

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

@Override
  public Object apply(Flowable<Integer> f) throws Exception {
    return f.concatMap(Functions.justFunction(Flowable.just(1).hide()));
  }
}, true, 1, 1, 1);

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

@Override
  public Object apply(Flowable<Integer> f) throws Exception {
    return f.concatMap(Functions.justFunction(Flowable.just(1).hide()));
  }
}, true, 1, 1, 1);

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

@Test
public void mapperThrows() {
  Flowable.range(1, 2)
  .concatMap(new Function<Integer, Publisher<Object>>() {
    @Override
    public Publisher<Object> apply(Integer v) throws Exception {
      throw new TestException();
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void concatMapErrorEmptySource() {
  assertSame(Flowable.empty(), Flowable.<Object>empty()
      .concatMap(new Function<Object, Flowable<Integer>>() {
        @Override
        public Flowable<Integer> apply(Object v) throws Exception {
          return Flowable.just(1);
        }
      }, 16));
}

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

@Test
public void concatMapJustSource() {
  Flowable.just(0).hide()
  .concatMap(new Function<Object, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Object v) throws Exception {
      return Flowable.just(1);
    }
  }, 16)
  .test()
  .assertResult(1);
}

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

@Test
public void concatMapScalarBackpressured() {
  Flowable.just(1).hide()
  .concatMap(Functions.justFunction(Flowable.just(2)))
  .test(1L)
  .assertResult(2);
}

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

@Test
public void fusedCrash() {
  Flowable.range(1, 2)
  .map(new Function<Integer, Object>() {
    @Override
    public Object apply(Integer v) throws Exception { throw new TestException(); }
  })
  .concatMap(Functions.justFunction(Flowable.just(1)))
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void callableCrash() {
  Flowable.just(1).hide()
  .concatMap(Functions.justFunction(Flowable.fromCallable(new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      throw new TestException();
    }
  })))
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void concatMapEmpty() {
  Flowable.just(1).hide()
  .concatMap(Functions.justFunction(Flowable.empty()))
  .test()
  .assertResult();
}

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

@Test
public void concatMapInnerError() {
  Flowable.just(1).hide()
  .concatMap(Functions.justFunction(Flowable.error(new TestException())))
  .test()
  .assertFailure(TestException.class);
}

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

@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void concatMapJustJust() {
  TestSubscriber<Integer> ts = TestSubscriber.create();
  Flowable.just(Flowable.just(1)).concatMap((Function)Functions.identity()).subscribe(ts);
  ts.assertValue(1);
  ts.assertNoErrors();
  ts.assertComplete();
}

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

@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void concatMapJustRange() {
  TestSubscriber<Integer> ts = TestSubscriber.create();
  Flowable.just(Flowable.range(1, 5)).concatMap((Function)Functions.identity()).subscribe(ts);
  ts.assertValues(1, 2, 3, 4, 5);
  ts.assertNoErrors();
  ts.assertComplete();
}