rx.Observable.test()方法的使用及代码示例

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

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

Observable.test介绍

暂无

代码示例

代码示例来源：origin: akarnokd/RxJava2Interop

  1. @Test
  2. public void v1RequestZero() {
  3.   toV1Observable(Flowable.range(1, 5))
  4.   .filter(new Func1<Integer, Boolean>() {
  5.     @Override
  6.     public Boolean call(Integer v) {
  7.       return v % 2 == 0;
  8.     }
  9.   })
  10.   .test()
  11.   .assertResult(2, 4);
  12. }

代码示例来源：origin: akarnokd/RxJava2Interop

  1. @Test
  2. public void fo2ToFo1Crash() {
  3.   FlowableOperator<Integer, Integer> transformer = new FlowableOperator<Integer, Integer>() {
  4.     @Override
  5.     public org.reactivestreams.Subscriber<? super Integer> apply(final org.reactivestreams.Subscriber<? super Integer> o) {
  6.       throw new IllegalArgumentException();
  7.     }
  8.   };
  9.   Observable.just(1)
  10.   .lift(toV1Operator(transformer))
  11.   .test()
  12.   .assertFailure(IllegalArgumentException.class);
  13. }

代码示例来源：origin: akarnokd/RxJava2Interop

  1. @Test
  2. public void ft2ToFt1() {
  3.   FlowableTransformer<Integer, Integer> transformer = new FlowableTransformer<Integer, Integer>() {
  4.     @Override
  5.     public Flowable<Integer> apply(Flowable<Integer> o) {
  6.       return o.map(new Function<Integer, Integer>() {
  7.         @Override
  8.         public Integer apply(Integer v) {
  9.           return v + 1;
  10.         }
  11.       });
  12.     }
  13.   };
  14.   rx.Observable.just(1)
  15.   .compose(toV1Transformer(transformer))
  16.   .test()
  17.   .assertResult(2);
  18. }

代码示例来源：origin: akarnokd/RxJava2Interop

  1. @Test
  2. public void ot2ToOt1() {
  3.   ObservableTransformer<Integer, Integer> transformer = new ObservableTransformer<Integer, Integer>() {
  4.     @Override
  5.     public io.reactivex.Observable<Integer> apply(io.reactivex.Observable<Integer> o) {
  6.       return o.map(new Function<Integer, Integer>() {
  7.         @Override
  8.         public Integer apply(Integer v) {
  9.           return v + 1;
  10.         }
  11.       });
  12.     }
  13.   };
  14.   rx.Observable.just(1)
  15.     .compose(toV1Transformer(transformer, BackpressureStrategy.BUFFER))
  16.     .test()
  17.     .assertResult(2);
  18. }

代码示例来源：origin: akarnokd/RxJava2Interop

  1. @Test
  2. public void fo2ToFo1() {
  3.   FlowableOperator<Integer, Integer> transformer = new FlowableOperator<Integer, Integer>() {
  4.     @Override
  5.     public org.reactivestreams.Subscriber<? super Integer> apply(final org.reactivestreams.Subscriber<? super Integer> o) {
  6.       return new org.reactivestreams.Subscriber<Integer>() {
  7.         @Override
  8.         public void onSubscribe(Subscription s) {
  9.           o.onSubscribe(s);
  10.         }
  11.         @Override
  12.         public void onNext(Integer t) {
  13.           o.onNext(t + 1);
  14.         }
  15.         @Override
  16.         public void onError(Throwable e) {
  17.           o.onError(e);
  18.         }
  19.         @Override
  20.         public void onComplete() {
  21.           o.onComplete();
  22.         }
  23.       };
  24.     }
  25.   };
  26.   Observable.just(1)
  27.   .lift(toV1Operator(transformer))
  28.   .test()
  29.   .assertResult(2);
  30. }

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

  1. @Test
  2. public void testSendByte() {
  3.   AssertableSubscriber<String> subscriber =
  4.       subscribe(destination, byte[].class).map(Message::getPayload).cast(byte[].class).map(String::new)
  5.                         .test();
  6.   sendSubscribeOperation.sendRx(destination, byteMessage);
  7.   subscriber.assertValue(payload).assertNoErrors();
  8. }

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

  1. @Test
  2. public void testSendUser() {
  3.   AssertableSubscriber<User> subscriber =
  4.       subscribe(destination, User.class).map(Message::getPayload).cast(User.class).test();
  5.   sendSubscribeOperation.sendRx(destination, userMessage);
  6.   subscriber.assertValue(user).assertNoErrors();
  7. }

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

  1. @Test
  2. public void testSendString() {
  3.   AssertableSubscriber<String> subscriber =
  4.       subscribe(destination, String.class).map(Message::getPayload).cast(String.class).test();
  5.   sendSubscribeOperation.sendRx(destination, stringMessage);
  6.   subscriber.assertValue(payload).assertNoErrors();
  7. }

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

  1. @Test
  2. public void testSendReceiveWithRecordCheckpointMode() {
  3.   setCheckpointConfig(CheckpointConfig.builder().checkpointMode(CheckpointMode.RECORD).build());
  4.   AssertableSubscriber<User> subscriber =
  5.       subscribe(destination, User.class).map(Message::getPayload).cast(User.class).test();
  6.   Arrays.stream(messages).forEach(m -> sendSubscribeOperation.sendRx(destination, m));
  7.   subscriber.assertValueCount(messages.length).assertNoErrors();
  8.   verifyCheckpointSuccessCalled(messages.length);
  9. }

代码示例来源：origin: vert-x3/vertx-rx

  1. protected void assertTableContainsInitDataOnly() throws Exception {
  2.  client.rxGetConnection().flatMapObservable(conn -> {
  3.   return uniqueNames(conn).doAfterTerminate(conn::close);
  4.  }).test()
  5.   .awaitTerminalEvent()
  6.   .assertCompleted()
  7.   .assertValues(NAMES.stream().sorted().distinct().toArray(String[]::new));
  8. }

代码示例来源：origin: akarnokd/akarnokd-misc

  1. @Test
  2.   public void test() {
  3.     Observable.error(new IOException())
  4.     .retryWhen(new RetryWhenObservable(5, 1))
  5.     .test()
  6.     .awaitTerminalEvent()
  7.     .assertFailure(IOException.class);
  8.   }
  9. }

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

  1. @Test
  2. public void testSendReceiveWithManualCheckpointMode() {
  3.   setCheckpointConfig(CheckpointConfig.builder().checkpointMode(CheckpointMode.MANUAL).build());
  4.   Observable<Message<?>> observable = subscribe(destination, User.class);
  5.   AssertableSubscriber<User> userSubscriber = observable.map(Message::getPayload).cast(User.class).test();
  6.   sendSubscribeOperation.sendRx(destination, userMessage);
  7.   userSubscriber.assertValue(user).assertNoErrors();
  8.   verifyCheckpointSuccessCalled(0);
  9. }

代码示例来源：origin: spotify/mobius

  1. @Test
  2.  public void shouldPropagateErrorsFromUpstream() throws Exception {
  3.   final Throwable expected = new RuntimeException("expected");
  5.   AssertableSubscriber<Integer> subscriber =
  6.     input.compose(RxConnectables.toTransformer(connectable)).test();
  8.   input.onError(expected);
  10.   subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS);
  11.   subscriber.assertError(expected);
  12.  }
  13. }

代码示例来源：origin: spotify/mobius

  1. @Test
  2.  public void shouldPropagateIncomingErrorsAsUnrecoverable() throws Exception {
  3.   PublishSubject<Integer> input = PublishSubject.create();
  5.   AssertableSubscriber<String> subscriber = input.compose(loop).test();
  7.   Exception expected = new RuntimeException("expected");
  9.   input.onError(expected);
  11.   subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS);
  12.   subscriber.assertError(new UnrecoverableIncomingException(expected));
  13.  }
  14. }

代码示例来源：origin: vert-x3/vertx-rx

  1. @Test
  2. public void inTransactionFailure() throws Exception {
  3.  Exception error = new Exception();
  4.  inTransaction(error).test()
  5.   .awaitTerminalEvent()
  6.   .assertError(error)
  7.   .assertValues(namesWithExtraFolks());
  8.  assertTableContainsInitDataOnly();
  9. }

代码示例来源：origin: spotify/mobius

  1. @Test
  2. public void effectPerformerInvokesFunctionWithReceivedEffectAndEmitsReturnedEvents() {
  3.  PublishSubject<String> upstream = PublishSubject.create();
  4.  TestScheduler scheduler = new TestScheduler();
  5.  Function<String, Integer> function = s -> s.length();
  6.  AssertableSubscriber<Integer> observer =
  7.    upstream.compose(Transformers.fromFunction(function, scheduler)).test();
  8.  upstream.onNext("Hello");
  9.  scheduler.triggerActions();
  10.  observer.assertValue(5);
  11. }

代码示例来源：origin: spotify/mobius

  1. @Test
  2. public void shouldPropagateErrorsFromConnectable() throws Exception {
  3.  AssertableSubscriber<Integer> subscriber =
  4.    input.compose(RxConnectables.toTransformer(connectable)).test();
  5.  input.onNext("crash");
  6.  subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS);
  7.  subscriber.assertFailureAndMessage(RuntimeException.class, "crashing!");
  8. }

代码示例来源：origin: vert-x3/vertx-rx

  1. @Test
  2. public void inTransactionSuccess() throws Exception {
  3.  inTransaction(null).test()
  4.   .awaitTerminalEvent()
  5.   .assertCompleted()
  6.   .assertValues(namesWithExtraFolks());
  7. }

代码示例来源：origin: spotify/mobius

  1. @Test
  2. public void shouldPropagateCompletion() throws Exception {
  3.  AssertableSubscriber<Integer> subscriber =
  4.    input.compose(RxConnectables.toTransformer(connectable)).test();
  5.  input.onNext("hi");
  6.  input.onCompleted();
  7.  subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS);
  8.  subscriber.assertCompleted();
  9. }

代码示例来源：origin: spotify/mobius

  1. @Test
  2. public void consumerTransformerShouldPropagateCompletion() throws Exception {
  3.  AssertableSubscriber<Object> subscriber =
  4.    upstream.compose(Transformers.fromConsumer(consumer, scheduler)).test();
  5.  upstream.onNext("hi");
  6.  upstream.onCompleted();
  7.  scheduler.triggerActions();
  8.  subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS);
  9.  subscriber.assertCompleted();
  10. }

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

Observable

相关文章

    查看更多

    热门标签

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

    最新文章

    更多

    Observable类方法

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