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

Flowable.concatMapEager介绍

[英]Maps a sequence of values into Publishers and concatenates these Publishers eagerly into a single Publisher.

Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the source Publishers. The operator buffers the values emitted by these Publishers and then drains them in order, each one after the previous one completes. Backpressure: Backpressure is honored towards the downstream, however, due to the eagerness requirement, sources are subscribed to in unbounded mode and their values are queued up in an unbounded buffer. Scheduler: This method does not operate by default on a particular Scheduler.
[中]将一系列值映射到发布服务器，并将这些发布服务器连接到单个发布服务器。
即时连接意味着一旦订阅者订阅，该操作员就订阅所有源发布者。运算符缓冲这些发布服务器发出的值，然后依次将其排出，在前一个发布服务器完成后再排出。背压：背压向下游移动，但是，由于迫切需要，源以无界模式订阅，其值在无界缓冲区中排队。调度程序：默认情况下，此方法不会在特定调度程序上运行。

代码示例

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

@Override
  public Flowable<Object> apply(Flowable<Object> f) throws Exception {
    return f.concatMapEager(new Function<Object, Flowable<Object>>() {
      @Override
      public Flowable<Object> apply(Object v) throws Exception {
        return Flowable.just(v);
      }
    });
  }
});

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

@Test(expected = IllegalArgumentException.class)
public void testInvalidMaxConcurrent() {
  Flowable.just(1).concatMapEager(toJust, 0, Flowable.bufferSize());
}

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

@Test(expected = IllegalArgumentException.class)
public void testInvalidCapacityHint() {
  Flowable.just(1).concatMapEager(toJust, Flowable.bufferSize(), 0);
}

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

@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void mappingBadCapacityHint() throws Exception {
  Flowable<Integer> source = Flowable.just(1);
  try {
    Flowable.just(source, source, source).concatMapEager((Function)Functions.identity(), 10, -99);
  } catch (IllegalArgumentException ex) {
    assertEquals("prefetch > 0 required but it was -99", ex.getMessage());
  }
}

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

@Test
public void normal() {
  Flowable.range(1, 5)
  .concatMapEager(new Function<Integer, Publisher<Integer>>() {
    @Override
    public Publisher<Integer> apply(Integer t) {
      return Flowable.range(t, 2);
    }
  })
  .test()
  .assertResult(1, 2, 2, 3, 3, 4, 4, 5, 5, 6);
}

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

@Test
public void dispose() {
  TestHelper.checkDisposed(Flowable.just(1).hide().concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) throws Exception {
      return Flowable.range(1, 2);
    }
  }));
}

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

@Test
public void innerErrorMaxConcurrency() {
  Flowable.<Integer>just(1).hide().concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) throws Exception {
      return Flowable.error(new TestException());
    }
  }, 1, 128)
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void innerError() {
  Flowable.<Integer>just(1).hide().concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) throws Exception {
      return Flowable.error(new TestException());
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void empty() {
  Flowable.<Integer>empty().hide().concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) throws Exception {
      return Flowable.range(1, 2);
    }
  })
  .test()
  .assertResult();
}

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

@Test
public void innerErrorFused() {
  Flowable.<Integer>just(1).hide().concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) throws Exception {
      return Flowable.range(1, 2).map(new Function<Integer, Integer>() {
        @Override
        public Integer apply(Integer v) throws Exception {
          throw new TestException();
        }
      });
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void innerCallableThrows() {
  Flowable.<Integer>just(1).hide().concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) throws Exception {
      return Flowable.fromCallable(new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
          throw new TestException();
        }
      });
    }
  })
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void testSimple() {
  Flowable.range(1, 100).concatMapEager(toJust).subscribe(ts);
  ts.assertNoErrors();
  ts.assertValueCount(100);
  ts.assertComplete();
}

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

@Test
public void testSimple2() {
  Flowable.range(1, 100).concatMapEager(toRange).subscribe(ts);
  ts.assertNoErrors();
  ts.assertValueCount(200);
  ts.assertComplete();
}

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

@Test
public void testMapperThrows() {
  Flowable.just(1).concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer t) {
      throw new TestException();
    }
  }).subscribe(ts);
  ts.assertNoValues();
  ts.assertNotComplete();
  ts.assertError(TestException.class);
}

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

@Test
public void mapperCancels() {
  final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.just(1).hide()
  .concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer v) throws Exception {
      ts.cancel();
      return Flowable.never();
    }
  }, 1, 128)
  .subscribe(ts);
  ts.assertEmpty();
}

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

@Test
public void testMainError() {
  Flowable.<Integer>error(new TestException()).concatMapEager(toJust).subscribe(ts);
  ts.assertNoValues();
  ts.assertError(TestException.class);
  ts.assertNotComplete();
}

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

@Test
@Ignore("Null values are not allowed in RS")
public void testInnerNull() {
  Flowable.just(1).concatMapEager(new Function<Integer, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Integer t) {
      return Flowable.just(null);
    }
  }).subscribe(ts);
  ts.assertNoErrors();
  ts.assertComplete();
  ts.assertValue(null);
}

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

@Test
public void unboundedIn() {
  int n = Flowable.bufferSize() * 2;
  Flowable.range(1, n)
  .concatMapEager(new Function<Integer, Publisher<Integer>>() {
    @Override
    public Publisher<Integer> apply(Integer v) throws Exception {
      return Flowable.just(1);
    }
  }, Integer.MAX_VALUE, 16)
  .test()
  .assertValueCount(n)
  .assertComplete()
  .assertNoErrors();
}

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

@Test
public void longEager() {
  Flowable.range(1, 2 * Flowable.bufferSize())
  .concatMapEager(new Function<Integer, Publisher<Integer>>() {
    @Override
    public Publisher<Integer> apply(Integer v) {
      return Flowable.just(1);
    }
  })
  .test()
  .assertValueCount(2 * Flowable.bufferSize())
  .assertNoErrors()
  .assertComplete();
}

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

@Test
public void innerLong() {
  int n = Flowable.bufferSize() * 2;
  Flowable.just(1).hide()
  .concatMapEager(Functions.justFunction(Flowable.range(1, n).hide()))
  .rebatchRequests(1)
  .test()
  .assertValueCount(n)
  .assertComplete()
  .assertNoErrors();
}