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

Observable.forEachWhile介绍

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

If the Observable emits an error, it is wrapped into an io.reactivex.exceptions.OnErrorNotImplementedExceptionand routed to the RxJavaPlugins.onError handler. Scheduler: forEachWhile does not operate by default on a particular Scheduler.
[中]订阅ObservableSource并接收每个元素的通知，直到onNext谓词返回false。
如果可观测对象发出错误，它将被包装到io中。reactivex。例外。OnErrorNotImplementedException并路由到RxJavaPlugins。一个错误处理程序。调度器：forEachWhile默认情况下不会在特定的调度器上运行。

代码示例

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

/**
 * Subscribes to the {@link ObservableSource} and receives notifications for each element and error events until the
 * onNext Predicate returns false.
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 *
 * @param onNext
 *            {@link Predicate} to execute for each item.
 * @param onError
 *            {@link Consumer} to execute when an error is emitted.
 * @return
 *            a Disposable that allows disposing of an asynchronous sequence
 * @throws NullPointerException
 *             if {@code onNext} is null, or
 *             if {@code onError} is null
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable forEachWhile(Predicate<? super T> onNext, Consumer<? super Throwable> onError) {
  return forEachWhile(onNext, onError, Functions.EMPTY_ACTION);
}

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

@Override
  public Object apply(Observable<Integer> f) throws Exception {
    return f.forEachWhile(Functions.alwaysTrue());
  }
}, false, 1, 1, (Object[])null);

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

@Test(expected = NullPointerException.class)
public void forEachWhileOnErrorNull() {
  just1.forEachWhile(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) {
      return true;
    }
  }, null);
}

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

@Test(expected = NullPointerException.class)
public void forEachWhileOnCompleteNull() {
  just1.forEachWhile(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) {
      return true;
    }
  }, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable e) { }
  }, null);
}

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

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

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

/**
 * Subscribes to the {@link ObservableSource} and receives notifications for each element until the
 * onNext Predicate returns false.
 * <p>
 * <img width="640" height="272" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/forEachWhile.o.png" alt="">
 * <p>
 * If the Observable emits an error, it is wrapped into an
 * {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
 * and routed to the RxJavaPlugins.onError handler.
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 *
 * @param onNext
 *            {@link Predicate} to execute for each item.
 * @return
 *            a Disposable that allows disposing of an asynchronous sequence
 * @throws NullPointerException
 *             if {@code onNext} is null
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable forEachWhile(Predicate<? super T> onNext) {
  return forEachWhile(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
}

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

/**
 * Subscribes to the {@link ObservableSource} and receives notifications for each element and error events until the
 * onNext Predicate returns false.
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 *
 * @param onNext
 *            {@link Predicate} to execute for each item.
 * @param onError
 *            {@link Consumer} to execute when an error is emitted.
 * @return
 *            a Disposable that allows cancelling an asynchronous sequence
 * @throws NullPointerException
 *             if {@code onNext} is null, or
 *             if {@code onError} is null
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable forEachWhile(Predicate<? super T> onNext, Consumer<? super Throwable> onError) {
  return forEachWhile(onNext, onError, Functions.EMPTY_ACTION);
}

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

/**
 * Subscribes to the {@link ObservableSource} and receives notifications for each element until the
 * onNext Predicate returns false.
 * <p>
 * <img width="640" height="272" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/forEachWhile.o.png" alt="">
 * <p>
 * If the Observable emits an error, it is wrapped into an
 * {@link io.reactivex.exceptions.OnErrorNotImplementedException OnErrorNotImplementedException}
 * and routed to the RxJavaPlugins.onError handler.
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code forEachWhile} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 *
 * @param onNext
 *            {@link Predicate} to execute for each item.
 * @return
 *            a Disposable that allows cancelling an asynchronous sequence
 * @throws NullPointerException
 *             if {@code onNext} is null
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable forEachWhile(Predicate<? super T> onNext) {
  return forEachWhile(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
}

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

@Test
public void forEachWile() {
  final List<Object> list = new ArrayList<Object>();
  Observable.range(1, 5)
  .doOnNext(new Consumer<Integer>() {
    @Override
    public void accept(Integer v) throws Exception {
      list.add(v);
    }
  })
  .forEachWhile(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) throws Exception {
      return v < 3;
    }
  });
  assertEquals(Arrays.asList(1, 2, 3), list);
}

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

@Test
public void whilePredicateThrows() {
  List<Throwable> errors = TestHelper.trackPluginErrors();
  try {
    Observable.just(1).forEachWhile(new Predicate<Integer>() {
      @Override
      public boolean test(Integer v) throws Exception {
        throw new TestException();
      }
    });
    TestHelper.assertError(errors, 0, OnErrorNotImplementedException.class);
    Throwable c = errors.get(0).getCause();
    assertTrue("" + c, c instanceof TestException);
  } finally {
    RxJavaPlugins.reset();
  }
}

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

@Test
public void observableForEachWhile() {
  Observable.error(new TestException())
  .forEachWhile(Functions.alwaysTrue());
}

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

@Test
public void forEachWileWithError() {
  final List<Object> list = new ArrayList<Object>();
  Observable.range(1, 5).concatWith(Observable.<Integer>error(new TestException()))
  .doOnNext(new Consumer<Integer>() {
    @Override
    public void accept(Integer v) throws Exception {
      list.add(v);
    }
  })
  .forEachWhile(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) throws Exception {
      return true;
    }
  }, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable e) throws Exception {
      list.add(100);
    }
  });
  assertEquals(Arrays.asList(1, 2, 3, 4, 5, 100), list);
}

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

@Test
public void whileCompleteThrows() {
  List<Throwable> errors = TestHelper.trackPluginErrors();
  try {
    Observable.just(1).forEachWhile(Functions.alwaysTrue(), Functions.emptyConsumer(),
        new Action() {
          @Override
          public void run() throws Exception {
            throw new TestException();
          }
        });
    TestHelper.assertUndeliverable(errors, 0, TestException.class);
  } finally {
    RxJavaPlugins.reset();
  }
}

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

@Test
public void whileErrorThrows() {
  List<Throwable> errors = TestHelper.trackPluginErrors();
  try {
    Observable.<Integer>error(new TestException("Outer"))
    .forEachWhile(Functions.alwaysTrue(), new Consumer<Throwable>() {
      @Override
      public void accept(Throwable v) throws Exception {
        throw new TestException("Inner");
      }
    });
    TestHelper.assertError(errors, 0, CompositeException.class);
    List<Throwable> ce = TestHelper.compositeList(errors.get(0));
    TestHelper.assertError(ce, 0, TestException.class, "Outer");
    TestHelper.assertError(ce, 1, TestException.class, "Inner");
  } finally {
    RxJavaPlugins.reset();
  }
}

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

Observable.forEachWhile介绍

代码示例

相关文章

热门标签

最新文章

Observable类方法