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

x33g5p2x  于2022-01-25 转载在 其他  
字(10.8k)|赞(0)|评价(0)|浏览(160)

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

Observable.takeUntil介绍

[英]Returns an Observable that emits the items emitted by the source Observable until a second ObservableSource emits an item.

Scheduler: takeUntil does not operate by default on a particular Scheduler.
[中]返回一个Observable，该Observable发出源Observable发出的项，直到第二个ObservableSource发出项为止。
调度程序：默认情况下，TakeTill不会在特定调度程序上运行。

代码示例

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

  1. @Override
  2.   public Observable<Integer> apply(Observable<Integer> o) throws Exception {
  3.     return o.takeUntil(Observable.never());
  4.   }
  5. });

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

  1. @Override
  2. public Observable<Integer> apply(Observable<Integer> xs) {
  3.   return xs.takeUntil(xs.skipWhile(new Predicate<Integer>() {
  4.     @Override
  5.     public boolean test(Integer i) {
  6.       return i <= 3;
  7.     }
  8.   }));
  9. }

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

  1. @Override
  2.   public ObservableSource<Object> apply(Observable<Object> o) throws Exception {
  3.     return o.takeUntil(Functions.alwaysFalse());
  4.   }
  5. });

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

  1. @Test(expected = NullPointerException.class)
  2. public void takeUntilPredicateNull() {
  3.   just1.takeUntil((Predicate<Integer>)null);
  4. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void takeUntilObservableNull() {
  3.   just1.takeUntil((Observable<Integer>)null);
  4. }

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

  1. /**
  2.  * Returns an Observable that emits those items emitted by source ObservableSource before a specified time runs
  3.  * out.
  4.  * <p>
  5.  * If time runs out before the {@code Observable} completes normally, the {@code onComplete} event will be
  6.  * signaled on the default {@code computation} {@link Scheduler}.
  7.  * <p>
  8.  * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/take.t.png" alt="">
  9.  * <dl>
  10.  *  <dt><b>Scheduler:</b></dt>
  11.  *  <dd>This version of {@code take} operates by default on the {@code computation} {@link Scheduler}.</dd>
  12.  * </dl>
  13.  *
  14.  * @param time
  15.  *            the length of the time window
  16.  * @param unit
  17.  *            the time unit of {@code time}
  18.  * @return an Observable that emits those items emitted by the source ObservableSource before the time runs out
  19.  * @see <a href="http://reactivex.io/documentation/operators/take.html">ReactiveX operators documentation: Take</a>
  20.  */
  21. @CheckReturnValue
  22. @SchedulerSupport(SchedulerSupport.NONE)
  23. public final Observable<T> take(long time, TimeUnit unit) {
  24.   return takeUntil(timer(time, unit));
  25. }

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

  1. @Test
  2. public void takeAll() {
  3.   Observer<Object> o = TestHelper.mockObserver();
  4.   Observable.just(1, 2).takeUntil(new Predicate<Integer>() {
  5.     @Override
  6.     public boolean test(Integer v) {
  7.       return false;
  8.     }
  9.   }).subscribe(o);
  10.   verify(o).onNext(1);
  11.   verify(o).onNext(2);
  12.   verify(o, never()).onError(any(Throwable.class));
  13.   verify(o).onComplete();
  14. }

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

  1. @Test
  2. public void takeEmpty() {
  3.   Observer<Object> o = TestHelper.mockObserver();
  4.   Observable.empty().takeUntil(new Predicate<Object>() {
  5.     @Override
  6.     public boolean test(Object v) {
  7.       return true;
  8.     }
  9.   }).subscribe(o);
  10.   verify(o, never()).onNext(any());
  11.   verify(o, never()).onError(any(Throwable.class));
  12.   verify(o).onComplete();
  13. }

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

  1. @Test
  2. public void takeFirst() {
  3.   Observer<Object> o = TestHelper.mockObserver();
  4.   Observable.just(1, 2).takeUntil(new Predicate<Integer>() {
  5.     @Override
  6.     public boolean test(Integer v) {
  7.       return true;
  8.     }
  9.   }).subscribe(o);
  10.   verify(o).onNext(1);
  11.   verify(o, never()).onNext(2);
  12.   verify(o, never()).onError(any(Throwable.class));
  13.   verify(o).onComplete();
  14. }

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

  1. @Test
  2. public void functionThrows() {
  3.   Observer<Object> o = TestHelper.mockObserver();
  4.   Predicate<Integer> predicate = (new Predicate<Integer>() {
  5.     @Override
  6.     public boolean test(Integer t1) {
  7.         throw new TestException("Forced failure");
  8.     }
  9.   });
  10.   Observable.just(1, 2, 3).takeUntil(predicate).subscribe(o);
  11.   verify(o).onNext(1);
  12.   verify(o, never()).onNext(2);
  13.   verify(o, never()).onNext(3);
  14.   verify(o).onError(any(TestException.class));
  15.   verify(o, never()).onComplete();
  16. }

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

  1. @Test
  2. public void takeSome() {
  3.   Observer<Object> o = TestHelper.mockObserver();
  4.   Observable.just(1, 2, 3).takeUntil(new Predicate<Integer>() {
  5.     @Override
  6.     public boolean test(Integer t1) {
  7.       return t1 == 2;
  8.     }
  9.   })
  10.   .subscribe(o);
  11.   verify(o).onNext(1);
  12.   verify(o).onNext(2);
  13.   verify(o, never()).onNext(3);
  14.   verify(o, never()).onError(any(Throwable.class));
  15.   verify(o).onComplete();
  16. }

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

  1. @Test
  2. public void testTakeUntilSourceCompleted() {
  3.   Disposable sSource = mock(Disposable.class);
  4.   Disposable sOther = mock(Disposable.class);
  5.   TestObservable source = new TestObservable(sSource);
  6.   TestObservable other = new TestObservable(sOther);
  7.   Observer<String> result = TestHelper.mockObserver();
  8.   Observable<String> stringObservable = Observable.unsafeCreate(source).takeUntil(Observable.unsafeCreate(other));
  9.   stringObservable.subscribe(result);
  10.   source.sendOnNext("one");
  11.   source.sendOnNext("two");
  12.   source.sendOnCompleted();
  13.   verify(result, times(1)).onNext("one");
  14.   verify(result, times(1)).onNext("two");
  15.   verify(sSource, never()).dispose(); // no longer disposing itself on terminal events
  16.   verify(sOther, times(1)).dispose();
  17. }

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

  1. @Test
  2. public void testTakeUntilSourceError() {
  3.   Disposable sSource = mock(Disposable.class);
  4.   Disposable sOther = mock(Disposable.class);
  5.   TestObservable source = new TestObservable(sSource);
  6.   TestObservable other = new TestObservable(sOther);
  7.   Throwable error = new Throwable();
  8.   Observer<String> result = TestHelper.mockObserver();
  9.   Observable<String> stringObservable = Observable.unsafeCreate(source).takeUntil(Observable.unsafeCreate(other));
  10.   stringObservable.subscribe(result);
  11.   source.sendOnNext("one");
  12.   source.sendOnNext("two");
  13.   source.sendOnError(error);
  14.   source.sendOnNext("three");
  15.   verify(result, times(1)).onNext("one");
  16.   verify(result, times(1)).onNext("two");
  17.   verify(result, times(0)).onNext("three");
  18.   verify(result, times(1)).onError(error);
  19.   verify(sSource, never()).dispose(); // no longer disposing itself on terminal events
  20.   verify(sOther, times(1)).dispose();
  21. }

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

  1. @Test
  2. public void sourceThrows() {
  3.   Observer<Object> o = TestHelper.mockObserver();
  4.   Observable.just(1)
  5.   .concatWith(Observable.<Integer>error(new TestException()))
  6.   .concatWith(Observable.just(2))
  7.   .takeUntil(new Predicate<Integer>() {
  8.     @Override
  9.     public boolean test(Integer v) {
  10.       return false;
  11.     }
  12.   }).subscribe(o);
  13.   verify(o).onNext(1);
  14.   verify(o, never()).onNext(2);
  15.   verify(o).onError(any(TestException.class));
  16.   verify(o, never()).onComplete();
  17. }

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

  1. @Test
  2. public void testTakeUntilOtherError() {
  3.   Disposable sSource = mock(Disposable.class);
  4.   Disposable sOther = mock(Disposable.class);
  5.   TestObservable source = new TestObservable(sSource);
  6.   TestObservable other = new TestObservable(sOther);
  7.   Throwable error = new Throwable();
  8.   Observer<String> result = TestHelper.mockObserver();
  9.   Observable<String> stringObservable = Observable.unsafeCreate(source).takeUntil(Observable.unsafeCreate(other));
  10.   stringObservable.subscribe(result);
  11.   source.sendOnNext("one");
  12.   source.sendOnNext("two");
  13.   other.sendOnError(error);
  14.   source.sendOnNext("three");
  15.   verify(result, times(1)).onNext("one");
  16.   verify(result, times(1)).onNext("two");
  17.   verify(result, times(0)).onNext("three");
  18.   verify(result, times(1)).onError(error);
  19.   verify(result, times(0)).onComplete();
  20.   verify(sSource, times(1)).dispose();
  21.   verify(sOther, never()).dispose(); // no longer disposing itself on termination
  22. }

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

  1. /**
  2.  * If the 'other' onCompletes then we unsubscribe from the source and onComplete.
  3.  */
  4. @Test
  5. public void testTakeUntilOtherCompleted() {
  6.   Disposable sSource = mock(Disposable.class);
  7.   Disposable sOther = mock(Disposable.class);
  8.   TestObservable source = new TestObservable(sSource);
  9.   TestObservable other = new TestObservable(sOther);
  10.   Observer<String> result = TestHelper.mockObserver();
  11.   Observable<String> stringObservable = Observable.unsafeCreate(source).takeUntil(Observable.unsafeCreate(other));
  12.   stringObservable.subscribe(result);
  13.   source.sendOnNext("one");
  14.   source.sendOnNext("two");
  15.   other.sendOnCompleted();
  16.   source.sendOnNext("three");
  17.   verify(result, times(1)).onNext("one");
  18.   verify(result, times(1)).onNext("two");
  19.   verify(result, times(0)).onNext("three");
  20.   verify(result, times(1)).onComplete();
  21.   verify(sSource, times(1)).dispose();
  22.   verify(sOther, never()).dispose(); // no longer disposing itself on terminal events
  23. }

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

  1. @Test
  2. public void testTakeUntil() {
  3.   Disposable sSource = mock(Disposable.class);
  4.   Disposable sOther = mock(Disposable.class);
  5.   TestObservable source = new TestObservable(sSource);
  6.   TestObservable other = new TestObservable(sOther);
  7.   Observer<String> result = TestHelper.mockObserver();
  8.   Observable<String> stringObservable = Observable.unsafeCreate(source)
  9.       .takeUntil(Observable.unsafeCreate(other));
  10.   stringObservable.subscribe(result);
  11.   source.sendOnNext("one");
  12.   source.sendOnNext("two");
  13.   other.sendOnNext("three");
  14.   source.sendOnNext("four");
  15.   source.sendOnCompleted();
  16.   other.sendOnCompleted();
  17.   verify(result, times(1)).onNext("one");
  18.   verify(result, times(1)).onNext("two");
  19.   verify(result, times(0)).onNext("three");
  20.   verify(result, times(0)).onNext("four");
  21.   verify(sSource, times(1)).dispose();
  22.   verify(sOther, times(1)).dispose();
  23. }

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

  1. @Test
  2.   public void testErrorIncludesLastValueAsCause() {
  3.     TestObserver<String> to = new TestObserver<String>();
  4.     final TestException e = new TestException("Forced failure");
  5.     Predicate<String> predicate = (new Predicate<String>() {
  6.       @Override
  7.       public boolean test(String t) {
  8.           throw e;
  9.       }
  10.     });
  11.     Observable.just("abc").takeUntil(predicate).subscribe(to);
  13.     to.assertTerminated();
  14.     to.assertNotComplete();
  15.     to.assertError(TestException.class);
  16.     // FIXME last cause value is not saved
  17. //        assertTrue(ts.errors().get(0).getCause().getMessage().contains("abc"));
  18.   }

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

  1. @Test /* (timeout = 8000) */
  2. public void testSingleSourceManyIterators() throws InterruptedException {
  3.   Observable<Long> o = Observable.interval(250, TimeUnit.MILLISECONDS);
  4.   PublishSubject<Integer> terminal = PublishSubject.create();
  5.   Observable<Long> source = o.takeUntil(terminal);
  6.   Iterable<Long> iter = source.blockingNext();
  7.   for (int j = 0; j < 3; j++) {
  8.     BlockingObservableNext.NextIterator<Long> it = (BlockingObservableNext.NextIterator<Long>)iter.iterator();
  9.     for (long i = 0; i < 10; i++) {
  10.       Assert.assertEquals(true, it.hasNext());
  11.       Assert.assertEquals(j + "th iteration next", Long.valueOf(i), it.next());
  12.     }
  13.     terminal.onNext(1);
  14.   }
  15. }

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

  1. @Test
  2.   public void disconnectBeforeConnect() {
  3.     BehaviorSubject<Integer> subject = BehaviorSubject.create();
  5.     Observable<Integer> observable = subject
  6.         .replay(1)
  7.         .refCount();
  9.     observable.takeUntil(Observable.just(1)).test();
  11.     subject.onNext(2);
  13.     observable.take(1).test().assertResult(2);
  14.   }
  15. }

内容来源于网络，如有侵权，请联系作者删除！

Observable

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Observable类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com