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

Flowable.test介绍

[英]Creates a TestSubscriber that requests Long.MAX_VALUE and subscribes it to this Flowable. Backpressure: The returned TestSubscriber consumes this Flowable in an unbounded fashion. Scheduler: test does not operate by default on a particular Scheduler.
[中]创建一个请求长消息的TestSubscriber。最大_值，并将其订阅到此可流动项。背压：返回的TestSubscriber以一种无限制的方式使用此Flowable。调度程序：默认情况下，测试不会在特定调度程序上运行。

代码示例

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

@Test
public void timedNoOutdatedData() {
  TestScheduler scheduler = new TestScheduler();
  Flowable<Integer> source = Flowable.just(1)
      .replay(2, TimeUnit.SECONDS, scheduler)
      .autoConnect();
  source.test().assertResult(1);
  source.test().assertResult(1);
  scheduler.advanceTimeBy(3, TimeUnit.SECONDS);
  source.test().assertResult();
}

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

@Test
public void scalarMapToEmpty() {
  FlowableScalarXMap.scalarXMap(1, new Function<Integer, Publisher<Integer>>() {
    @Override
    public Publisher<Integer> apply(Integer v) throws Exception {
      return Flowable.empty();
    }
  })
  .test()
  .assertResult();
}

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

@Test
public void multicastSelectorCallableConnectableCrash() {
  FlowableReplay.multicastSelector(new Callable<ConnectableFlowable<Object>>() {
    @Override
    public ConnectableFlowable<Object> call() throws Exception {
      throw new TestException();
    }
  }, Functions.<Flowable<Object>>identity())
  .test()
  .assertFailure(TestException.class);
}

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

@Test
public void emitLastTimed() {
  Flowable.just(1)
  .sample(1, TimeUnit.DAYS, true)
  .test()
  .assertResult(1);
}

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

@Test
public void scanEmptyBackpressured() {
  Flowable.<Integer>empty()
  .scan(0, SUM)
  .test(1)
  .assertResult(0);
}

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

@Test
public void replaySelectorSizeScheduler() {
  Flowable.just(1).replay(Functions.<Flowable<Integer>>identity(), 1, Schedulers.io())
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

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

@Test
public void replaySelectorReturnsNull() {
  Flowable.just(1)
  .replay(new Function<Flowable<Integer>, Publisher<Object>>() {
    @Override
    public Publisher<Object> apply(Flowable<Integer> v) throws Exception {
      return null;
    }
  }, Schedulers.trampoline())
  .test()
  .assertFailureAndMessage(NullPointerException.class, "The selector returned a null Publisher");
}

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

@Test
public void emitLastOther() {
  Flowable.just(1)
  .sample(Flowable.timer(1, TimeUnit.DAYS), true)
  .test()
  .assertResult(1);
}

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

@Test
  public void singleOrError() {
    Flowable.empty()
    .singleOrError()
    .toFlowable()
    .test()
    .assertFailure(NoSuchElementException.class);
  }
}

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

@Test
public void replaySizeAndTime() {
  Flowable.just(1).replay(1, 1, TimeUnit.MILLISECONDS)
  .autoConnect()
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

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

@Test
public void boundaryDirectSecondMissingBackpressure() {
  BehaviorProcessor.createDefault(1)
  .window(Flowable.just(1))
  .test(1)
  .assertError(MissingBackpressureException.class)
  .assertNotComplete();
}

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

@Test
public void dontRetry() {
  Flowable.error(new TestException("Outer"))
  .retry(Functions.alwaysFalse())
  .test()
  .assertFailureAndMessage(TestException.class, "Outer");
}

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

@Test
public void sourceSupplierReturnsNull() {
  Flowable.using(Functions.justCallable(1),
      Functions.justFunction((Publisher<Object>)null),
      Functions.emptyConsumer())
  .test()
  .assertFailureAndMessage(NullPointerException.class, "The sourceSupplier returned a null Publisher")
  ;
}

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

@Test
public void prefetchFlowable() {
  Flowable.sequenceEqual(Flowable.range(1, 20), Flowable.range(1, 20), 2)
  .toFlowable()
  .test()
  .assertResult(true);
}

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

@Test
public void mainError() {
  Flowable.error(new TestException())
  .window(Functions.justCallable(Flowable.never()))
  .test()
  .assertError(TestException.class);
}

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

@Test
public void boundaryOnError() {
  TestSubscriber<Object> ts = Flowable.error(new TestException())
  .window(Flowable.never())
  .flatMap(Functions.<Flowable<Object>>identity(), true)
  .test()
  .assertFailure(CompositeException.class);
  List<Throwable> errors = TestHelper.compositeList(ts.errors().get(0));
  TestHelper.assertError(errors, 0, TestException.class);
}

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

@Test
public void skipLastTimedDefaultScheduler() {
  Flowable.just(1).concatWith(Flowable.just(2).delay(500, TimeUnit.MILLISECONDS))
  .skipLast(300, TimeUnit.MILLISECONDS)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}

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

@Test
public void boundaryError() {
  BehaviorProcessor.createDefault(1)
  .window(Functions.justCallable(Flowable.error(new TestException())))
  .test()
  .assertValueCount(1)
  .assertNotComplete()
  .assertError(TestException.class);
}

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

@Test
public void takeHalf() {
  int elements = 1024;
  Flowable.range(0, elements * 2).unsubscribeOn(Schedulers.single())
  .take(elements)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertValueCount(elements)
  .assertComplete()
  .assertNoErrors()
  .assertSubscribed();
}

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

@Test
public void cancelAsFlowable() {
  PublishProcessor<Integer> pp = PublishProcessor.create();
  TestSubscriber<Integer> ts = pp.singleOrError().toFlowable().test();
  assertTrue(pp.hasSubscribers());
  ts.assertEmpty();
  ts.cancel();
  assertFalse(pp.hasSubscribers());
}