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

Flowable.repeat介绍

[英]Returns a Flowable that repeats the sequence of items emitted by the source Publisher indefinitely.

Backpressure: The operator honors downstream backpressure and expects the source Publisher to honor backpressure as well. If this expectation is violated, the operator may throw an IllegalStateException. Scheduler: repeat does not operate by default on a particular Scheduler.
[中]返回无限期重复源发布服务器发出的项序列的可流动项。
背压：操作员接受下游背压，并希望源发布者也接受背压。如果违反此期望，运算符可能抛出一个非法状态异常。计划程序：默认情况下，重复不会在特定计划程序上运行。

代码示例

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

@Override
  public Publisher<Integer> createPublisher(long elements) {
    return Flowable.just(1).repeat(elements);
  }
}

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

@Test(timeout = 2000)
public void testRepeatOne() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1).repeat(1).subscribe(subscriber);
  verify(subscriber).onComplete();
  verify(subscriber, times(1)).onNext(any());
  verify(subscriber, never()).onError(any(Throwable.class));
}

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

@Test(timeout = 2000)
public void testRepeatError() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.error(new TestException()).repeat(10).subscribe(subscriber);
  verify(subscriber).onError(any(TestException.class));
  verify(subscriber, never()).onNext(any());
  verify(subscriber, never()).onComplete();
}

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

@Test(timeout = 2000)
public void testRepeatZero() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1).repeat(0).subscribe(subscriber);
  verify(subscriber).onComplete();
  verify(subscriber, never()).onNext(any());
  verify(subscriber, never()).onError(any(Throwable.class));
}

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

@Test(timeout = 2000)
public void testRepeatLimited() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1).repeat(10).subscribe(subscriber);
  verify(subscriber, times(10)).onNext(1);
  verify(subscriber).onComplete();
  verify(subscriber, never()).onError(any(Throwable.class));
}

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

@Test(timeout = 2000)
public void testRepeatAndTake() {
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  Flowable.just(1).repeat().take(10).subscribe(subscriber);
  verify(subscriber, times(10)).onNext(1);
  verify(subscriber).onComplete();
  verify(subscriber, never()).onError(any(Throwable.class));
}

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

@Test
public void repeatLongPredicateInvalid() {
  try {
    Flowable.just(1).repeat(-99);
    fail("Should have thrown");
  } catch (IllegalArgumentException ex) {
    assertEquals("times >= 0 required but it was -99", ex.getMessage());
  }
}

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

@Test(timeout = 2000)
public void bufferWithSizeSkipTake1() {
  Flowable<Integer> source = Flowable.just(1).repeat();
  Flowable<List<Integer>> result = source.buffer(2, 3).take(1);
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  result.subscribe(subscriber);
  verify(subscriber).onNext(Arrays.asList(1, 1));
  verify(subscriber).onComplete();
  verify(subscriber, never()).onError(any(Throwable.class));
}

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

@Test(timeout = 2000)
public void bufferWithSizeTake1() {
  Flowable<Integer> source = Flowable.just(1).repeat();
  Flowable<List<Integer>> result = source.buffer(2).take(1);
  Subscriber<Object> subscriber = TestHelper.mockSubscriber();
  result.subscribe(subscriber);
  verify(subscriber).onNext(Arrays.asList(1, 1));
  verify(subscriber).onComplete();
  verify(subscriber, never()).onError(any(Throwable.class));
}

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

@Test
public void noCancelPreviousRepeat() {
  final AtomicInteger counter = new AtomicInteger();
  Flowable<Integer> source = Flowable.just(1).doOnCancel(new Action() {
    @Override
    public void run() throws Exception {
      counter.getAndIncrement();
    }
  });
  source.repeat(5)
  .test()
  .assertResult(1, 1, 1, 1, 1);
  assertEquals(0, counter.get());
}

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

@Test(timeout = 5000)
public void concatObservableMany() {
  Completable c = Completable.concat(Flowable.just(normal.completable).repeat(3));
  c.blockingAwait();
  normal.assertSubscriptions(3);
}

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

@Test(timeout = 5000)
public void mergeObservableMany() {
  Completable c = Completable.merge(Flowable.just(normal.completable).repeat(3));
  c.blockingAwait();
  normal.assertSubscriptions(3);
}

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

@Test(timeout = 5000)
public void mergeDelayErrorObservableMany() {
  Completable c = Completable.mergeDelayError(Flowable.just(normal.completable).repeat(3));
  c.blockingAwait();
  normal.assertSubscriptions(3);
}

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

@Test(timeout = 2000)
public void testRepeatTake() {
  Flowable<Integer> xs = Flowable.just(1, 2);
  Object[] ys = xs.repeat().subscribeOn(Schedulers.newThread()).take(4).toList().blockingGet().toArray();
  assertArrayEquals(new Object[] { 1, 2, 1, 2 }, ys);
}

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

@Test(timeout = 2000)
public void testRepetition() {
  int num = 10;
  final AtomicInteger count = new AtomicInteger();
  int value = Flowable.unsafeCreate(new Publisher<Integer>() {
    @Override
    public void subscribe(final Subscriber<? super Integer> subscriber) {
      subscriber.onNext(count.incrementAndGet());
      subscriber.onComplete();
    }
  }).repeat().subscribeOn(Schedulers.computation())
  .take(num).blockingLast();
  assertEquals(num, value);
}

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

@Test(timeout = 20000)
public void testNoStackOverFlow() {
  Flowable.just(1).repeat().subscribeOn(Schedulers.newThread()).take(100000).blockingLast();
}

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

@Test
public void testRepeatObservableThrowsError() {
  TestSubscriber<String> subscriber = TestSubscriber.create();
  Single<String> single = Flowable.just("First", "Second").repeat().single("");
  single.toFlowable().subscribe(subscriber);
  subscriber.assertError(IllegalArgumentException.class);
}

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

/** Issue #2587. */
@Test
public void testRepeatAndDistinctUnbounded() {
  Flowable<Integer> src = Flowable.fromIterable(Arrays.asList(1, 2, 3, 4, 5))
      .take(3)
      .repeat(3)
      .distinct();
  TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  src.subscribe(ts);
  ts.assertNoErrors();
  ts.assertTerminated();
  ts.assertValues(1, 2, 3);
}

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

@Test(timeout = 1000)
public void testRaceConditions() {
  Scheduler comp = Schedulers.computation();
  Scheduler limited = comp.when(new Function<Flowable<Flowable<Completable>>, Completable>() {
    @Override
    public Completable apply(Flowable<Flowable<Completable>> t) {
      return Completable.merge(Flowable.merge(t, 10));
    }
  });
  merge(just(just(1).subscribeOn(limited).observeOn(comp)).repeat(1000)).blockingSubscribe();
}

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

@Test
public void repeatScheduled() {
  TestSubscriber<Integer> ts = TestSubscriber.create();
  Flowable.just(1).subscribeOn(Schedulers.computation()).repeat(5).subscribe(ts);
  ts.awaitTerminalEvent(5, TimeUnit.SECONDS);
  ts.assertValues(1, 1, 1, 1, 1);
  ts.assertNoErrors();
  ts.assertComplete();
}