io.reactivex.Flowable.forEachWhile()方法的使用及代码示例

x33g5p2x  于2022-01-19 转载在 其他  
字(7.4k)|赞(0)|评价(0)|浏览(199)

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

Flowable.forEachWhile介绍

[英]Subscribes to the Publisher and receives notifications for each element until the onNext Predicate returns false.

If the Flowable emits an error, it is wrapped into an io.reactivex.exceptions.OnErrorNotImplementedExceptionand routed to the RxJavaPlugins.onError handler. Backpressure: The operator consumes the source Publisher in an unbounded manner (i.e., no backpressure is applied to it). Scheduler: forEachWhile does not operate by default on a particular Scheduler.
[中]订阅发布服务器并接收每个元素的通知,直到onNext谓词返回false。
如果Flowable发出错误,它将被包装到io中。reactivex。例外情况。OnErrorNotImplementedException并路由到RxJavaPlugins。一个错误处理器。背压:操作员以无限制的方式使用源发布服务器(即,不向其应用背压)。调度程序:forEachWhile默认情况下不会在特定调度程序上运行。

代码示例

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

  1. @Test(expected = NullPointerException.class)
  2. public void forEachWhileOnCompleteNull() {
  3. just1.forEachWhile(new Predicate<Integer>() {
  4. @Override
  5. public boolean test(Integer v) {
  6. return true;
  7. }
  8. }, new Consumer<Throwable>() {
  9. @Override
  10. public void accept(Throwable e) { }
  11. }, null);
  12. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void forEachWhileNull() {
  3. just1.forEachWhile(null);
  4. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void forEachWhileOnErrorNull() {
  3. just1.forEachWhile(new Predicate<Integer>() {
  4. @Override
  5. public boolean test(Integer v) {
  6. return true;
  7. }
  8. }, null);
  9. }

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

  1. /**
  2. * Subscribes to the {@link Publisher} and receives notifications for each element and error events until the
  3. * onNext Predicate returns false.
  4. * <dl>
  5. * <dt><b>Backpressure:</b></dt>
  6. * <dd>The operator consumes the source {@code Publisher} in an unbounded manner (i.e., no
  7. * backpressure is applied to it).</dd>
  8. * <dt><b>Scheduler:</b></dt>
  9. * <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
  10. * </dl>
  11. *
  12. * @param onNext
  13. * {@link Predicate} to execute for each item.
  14. * @param onError
  15. * {@link Consumer} to execute when an error is emitted.
  16. * @return
  17. * a {@link Disposable} that allows canceling an asynchronous sequence
  18. * @throws NullPointerException
  19. * if {@code onNext} is null, or
  20. * if {@code onError} is null
  21. * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
  22. */
  23. @CheckReturnValue
  24. @BackpressureSupport(BackpressureKind.NONE)
  25. @SchedulerSupport(SchedulerSupport.NONE)
  26. public final Disposable forEachWhile(Predicate<? super T> onNext, Consumer<? super Throwable> onError) {
  27. return forEachWhile(onNext, onError, Functions.EMPTY_ACTION);
  28. }

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

  1. /**
  2. * Subscribes to the {@link Publisher} and receives notifications for each element until the
  3. * onNext Predicate returns false.
  4. * <p>
  5. * If the Flowable emits an error, it is wrapped into an
  6. * {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
  7. * and routed to the RxJavaPlugins.onError handler.
  8. * <dl>
  9. * <dt><b>Backpressure:</b></dt>
  10. * <dd>The operator consumes the source {@code Publisher} in an unbounded manner (i.e., no
  11. * backpressure is applied to it).</dd>
  12. * <dt><b>Scheduler:</b></dt>
  13. * <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
  14. * </dl>
  15. *
  16. * @param onNext
  17. * {@link Predicate} to execute for each item.
  18. * @return
  19. * a {@link Disposable} that allows canceling an asynchronous sequence
  20. * @throws NullPointerException
  21. * if {@code onNext} is null
  22. * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
  23. */
  24. @CheckReturnValue
  25. @BackpressureSupport(BackpressureKind.NONE)
  26. @SchedulerSupport(SchedulerSupport.NONE)
  27. public final Disposable forEachWhile(Predicate<? super T> onNext) {
  28. return forEachWhile(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
  29. }

代码示例来源:origin: redisson/redisson

  1. /**
  2. * Subscribes to the {@link Publisher} and receives notifications for each element until the
  3. * onNext Predicate returns false.
  4. * <p>
  5. * If the Flowable emits an error, it is wrapped into an
  6. * {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
  7. * and routed to the RxJavaPlugins.onError handler.
  8. * <dl>
  9. * <dt><b>Backpressure:</b></dt>
  10. * <dd>The operator consumes the source {@code Publisher} in an unbounded manner (i.e., no
  11. * backpressure is applied to it).</dd>
  12. * <dt><b>Scheduler:</b></dt>
  13. * <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
  14. * </dl>
  15. *
  16. * @param onNext
  17. * {@link Predicate} to execute for each item.
  18. * @return
  19. * a {@link Disposable} that allows canceling an asynchronous sequence
  20. * @throws NullPointerException
  21. * if {@code onNext} is null
  22. * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
  23. */
  24. @CheckReturnValue
  25. @BackpressureSupport(BackpressureKind.NONE)
  26. @SchedulerSupport(SchedulerSupport.NONE)
  27. public final Disposable forEachWhile(Predicate<? super T> onNext) {
  28. return forEachWhile(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
  29. }

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

  1. @Test
  2. public void forEachWile() {
  3. final List<Object> list = new ArrayList<Object>();
  4. Flowable.range(1, 5)
  5. .doOnNext(new Consumer<Integer>() {
  6. @Override
  7. public void accept(Integer v) throws Exception {
  8. list.add(v);
  9. }
  10. })
  11. .forEachWhile(new Predicate<Integer>() {
  12. @Override
  13. public boolean test(Integer v) throws Exception {
  14. return v < 3;
  15. }
  16. });
  17. assertEquals(Arrays.asList(1, 2, 3), list);
  18. }

代码示例来源:origin: redisson/redisson

  1. /**
  2. * Subscribes to the {@link Publisher} and receives notifications for each element and error events until the
  3. * onNext Predicate returns false.
  4. * <dl>
  5. * <dt><b>Backpressure:</b></dt>
  6. * <dd>The operator consumes the source {@code Publisher} in an unbounded manner (i.e., no
  7. * backpressure is applied to it).</dd>
  8. * <dt><b>Scheduler:</b></dt>
  9. * <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
  10. * </dl>
  11. *
  12. * @param onNext
  13. * {@link Predicate} to execute for each item.
  14. * @param onError
  15. * {@link Consumer} to execute when an error is emitted.
  16. * @return
  17. * a {@link Disposable} that allows canceling an asynchronous sequence
  18. * @throws NullPointerException
  19. * if {@code onNext} is null, or
  20. * if {@code onError} is null
  21. * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
  22. */
  23. @CheckReturnValue
  24. @BackpressureSupport(BackpressureKind.NONE)
  25. @SchedulerSupport(SchedulerSupport.NONE)
  26. public final Disposable forEachWhile(Predicate<? super T> onNext, Consumer<? super Throwable> onError) {
  27. return forEachWhile(onNext, onError, Functions.EMPTY_ACTION);
  28. }

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

  1. @Test
  2. public void flowableForEachWhile() {
  3. Flowable.error(new TestException())
  4. .forEachWhile(Functions.alwaysTrue());
  5. }

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

  1. @Test
  2. public void forEachWileWithError() {
  3. final List<Object> list = new ArrayList<Object>();
  4. Flowable.range(1, 5).concatWith(Flowable.<Integer>error(new TestException()))
  5. .doOnNext(new Consumer<Integer>() {
  6. @Override
  7. public void accept(Integer v) throws Exception {
  8. list.add(v);
  9. }
  10. })
  11. .forEachWhile(new Predicate<Integer>() {
  12. @Override
  13. public boolean test(Integer v) throws Exception {
  14. return true;
  15. }
  16. }, new Consumer<Throwable>() {
  17. @Override
  18. public void accept(Throwable e) throws Exception {
  19. list.add(100);
  20. }
  21. });
  22. assertEquals(Arrays.asList(1, 2, 3, 4, 5, 100), list);
  23. }

相关文章

Flowable类方法