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

Flowable.takeUntil介绍

[英]Returns a Flowable that emits items emitted by the source Publisher, checks the specified predicate for each item, and then completes when the condition is satisfied.

The difference between this operator and #takeWhile(Predicate) is that here, the condition is evaluated after the item is emitted. Backpressure: The operator is a pass-through for backpressure; the backpressure behavior is determined by the upstream source and the downstream consumer. Scheduler: takeUntil does not operate by default on a particular Scheduler.
[中]返回一个可流动项，该可流动项发出源发布服务器发出的项，检查每个项的指定谓词，然后在满足条件时完成。
此运算符与#takeWhile（谓词）之间的区别在于，在这里，条件是在发出项之后计算的。背压：操作员是背压的传递者；背压行为由上游气源和下游耗电元件决定。调度程序：默认情况下，TakeTill不会在特定调度程序上运行。

代码示例

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

@Override
  public Flowable<Integer> apply(Flowable<Integer> c) throws Exception {
    return c.takeUntil(Flowable.never());
  }
});

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

@Override
public Flowable<Integer> apply(Flowable<Integer> xs) {
  return xs.takeUntil(xs.skipWhile(new Predicate<Integer>() {
    @Override
    public boolean test(Integer i) {
      return i <= 3;
    }
  }));
}

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

@Override
  public Flowable<Object> apply(Flowable<Object> f) throws Exception {
    return f.takeUntil(Functions.alwaysFalse());
  }
});

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

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

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

@Test(expected = NullPointerException.class)
public void takeUntilPublisherNull() {
  just1.takeUntil((Publisher<Integer>)null);
}

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

@Test(expected = NullPointerException.class)
public void takeUntilFlowableNull() {
  just1.takeUntil((Flowable<Integer>)null);
}

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

@Override
  public Publisher<Integer> createPublisher(long elements) {
    return
        Flowable.range(0, (int)elements).takeUntil(Flowable.never())
    ;
  }
}

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

@Test
public void takeEmpty() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.empty().takeUntil(new Predicate<Object>() {
    @Override
    public boolean test(Object v) {
      return true;
    }
  }).subscribe(subscriber);
  verify(subscriber, never()).onNext(any());
  verify(subscriber, never()).onError(any(Throwable.class));
  verify(subscriber).onComplete();
}

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

@Test
public void takeFirst() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1, 2).takeUntil(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) {
      return true;
    }
  }).subscribe(subscriber);
  verify(subscriber).onNext(1);
  verify(subscriber, never()).onNext(2);
  verify(subscriber, never()).onError(any(Throwable.class));
  verify(subscriber).onComplete();
}

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

@Test
public void functionThrows() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Predicate<Integer> predicate = new Predicate<Integer>() {
    @Override
    public boolean test(Integer t1) {
        throw new TestException("Forced failure");
    }
  };
  Flowable.just(1, 2, 3).takeUntil(predicate).subscribe(subscriber);
  verify(subscriber).onNext(1);
  verify(subscriber, never()).onNext(2);
  verify(subscriber, never()).onNext(3);
  verify(subscriber).onError(any(TestException.class));
  verify(subscriber, never()).onComplete();
}

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

@Test
public void takeAll() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1, 2).takeUntil(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) {
      return false;
    }
  }).subscribe(subscriber);
  verify(subscriber).onNext(1);
  verify(subscriber).onNext(2);
  verify(subscriber, never()).onError(any(Throwable.class));
  verify(subscriber).onComplete();
}

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

@Test
public void takeSome() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1, 2, 3).takeUntil(new Predicate<Integer>() {
    @Override
    public boolean test(Integer t1) {
      return t1 == 2;
    }
  })
  .subscribe(subscriber);
  verify(subscriber).onNext(1);
  verify(subscriber).onNext(2);
  verify(subscriber, never()).onNext(3);
  verify(subscriber, never()).onError(any(Throwable.class));
  verify(subscriber).onComplete();
}

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

@Test
public void testTakeUntilSourceCompleted() {
  Subscription sSource = mock(Subscription.class);
  Subscription sOther = mock(Subscription.class);
  TestObservable source = new TestObservable(sSource);
  TestObservable other = new TestObservable(sOther);
  Subscriber<String> result = TestHelper.mockSubscriber();
  Flowable<String> stringObservable = Flowable.unsafeCreate(source).takeUntil(Flowable.unsafeCreate(other));
  stringObservable.subscribe(result);
  source.sendOnNext("one");
  source.sendOnNext("two");
  source.sendOnCompleted();
  verify(result, times(1)).onNext("one");
  verify(result, times(1)).onNext("two");
  verify(sSource, never()).cancel();
  verify(sOther, times(1)).cancel();
}

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

@Test
public void sourceThrows() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1)
  .concatWith(Flowable.<Integer>error(new TestException()))
  .concatWith(Flowable.just(2))
  .takeUntil(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) {
      return false;
    }
  }).subscribe(subscriber);
  verify(subscriber).onNext(1);
  verify(subscriber, never()).onNext(2);
  verify(subscriber).onError(any(TestException.class));
  verify(subscriber, never()).onComplete();
}

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

@Test /* (timeout = 8000) */
public void testSingleSourceManyIterators() throws InterruptedException {
  Flowable<Long> f = Flowable.interval(250, TimeUnit.MILLISECONDS);
  PublishProcessor<Integer> terminal = PublishProcessor.create();
  Flowable<Long> source = f.takeUntil(terminal);
  Iterable<Long> iter = source.blockingNext();
  for (int j = 0; j < 3; j++) {
    BlockingFlowableNext.NextIterator<Long> it = (BlockingFlowableNext.NextIterator<Long>)iter.iterator();
    for (long i = 0; i < 10; i++) {
      Assert.assertEquals(true, it.hasNext());
      Assert.assertEquals(j + "th iteration next", Long.valueOf(i), it.next());
    }
    terminal.onNext(1);
  }
}

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

@Test
public void delayAndTakeUntilNeverSubscribeToSource() {
  PublishProcessor<Integer> delayUntil = PublishProcessor.create();
  PublishProcessor<Integer> interrupt = PublishProcessor.create();
  final AtomicBoolean subscribed = new AtomicBoolean(false);
  Flowable.just(1)
  .doOnSubscribe(new Consumer<Subscription>() {
    @Override
    public void accept(Subscription s) {
      subscribed.set(true);
    }
  })
  .delaySubscription(delayUntil)
  .takeUntil(interrupt)
  .subscribe();
  interrupt.onNext(9000);
  delayUntil.onNext(1);
  Assert.assertFalse(subscribed.get());
}

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

@Test
public void backpressure() {
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>(5L);
  Flowable.range(1, 1000).takeUntil(new Predicate<Integer>() {
    @Override
    public boolean test(Integer v) {
      return false;
    }
  }).subscribe(ts);
  ts.assertNoErrors();
  ts.assertValues(1, 2, 3, 4, 5);
  ts.assertNotComplete();
}

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

@Test
  public void testErrorIncludesLastValueAsCause() {
    TestSubscriber<String> ts = new TestSubscriber<String>();
    final TestException e = new TestException("Forced failure");
    Predicate<String> predicate = new Predicate<String>() {
      @Override
      public boolean test(String t) {
          throw e;
      }
    };
    Flowable.just("abc").takeUntil(predicate).subscribe(ts);
    ts.assertTerminated();
    ts.assertNotComplete();
    ts.assertError(TestException.class);
    // FIXME last cause value is not saved
//        assertTrue(ts.errors().get(0).getCause().getMessage().contains("abc"));
  }

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

@Test
  public void disconnectBeforeConnect() {
    BehaviorProcessor<Integer> processor = BehaviorProcessor.create();
    Flowable<Integer> flowable = processor
        .replay(1)
        .refCount();
    flowable.takeUntil(Flowable.just(1)).test();
    processor.onNext(2);
    flowable.take(1).test().assertResult(2);
  }
}

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

@Test
public void testBackpressure() {
  PublishProcessor<Integer> until = PublishProcessor.create();
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L);
  Flowable.range(1, 10).takeUntil(until).subscribe(ts);
  assertTrue(until.hasSubscribers());
  ts.request(1);
  ts.assertValue(1);
  ts.assertNoErrors();
  ts.assertNotComplete();
  until.onNext(5);
  ts.assertComplete();
  ts.assertNoErrors();
  assertFalse("Until still has observers", until.hasSubscribers());
  assertFalse("TestSubscriber is unsubscribed", ts.isCancelled());
}