io.reactivex.Flowable.using()方法的使用及代码示例

x33g5p2x  于2022-01-19 转载在 其他  
字(11.6k)|赞(0)|评价(0)|浏览(280)

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

Flowable.using介绍

[英]Constructs a Publisher that creates a dependent resource object which is disposed of on cancellation.

Backpressure: The operator is a pass-through for backpressure and otherwise depends on the backpressure support of the Publisher returned by the resourceFactory. Scheduler: using does not operate by default on a particular Scheduler.
[中]构造一个发布服务器，该发布服务器创建从属资源对象，该对象在取消时被释放。
背压：操作员是背压的传递，否则取决于resourceFactory返回的发布服务器的背压支持。调度程序：默认情况下，使用不会在特定调度程序上运行。

代码示例

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingDisposeNull() {
  3.   Flowable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, Publisher<Integer>>() {
  9.     @Override
  10.     public Publisher<Integer> apply(Object d) {
  11.       return just1;
  12.     }
  13.   }, null);
  14. }

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

  1. private void performTestUsingWithResourceFactoryError(boolean disposeEagerly) {
  2.   Callable<Disposable> resourceFactory = new Callable<Disposable>() {
  3.     @Override
  4.     public Disposable call() {
  5.       throw new TestException();
  6.     }
  7.   };
  8.   Function<Disposable, Flowable<Integer>> observableFactory = new Function<Disposable, Flowable<Integer>>() {
  9.     @Override
  10.     public Flowable<Integer> apply(Disposable d) {
  11.       return Flowable.empty();
  12.     }
  13.   };
  14.   Flowable.using(resourceFactory, observableFactory, disposeSubscription)
  15.   .blockingLast();
  16. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingFlowableSupplierNull() {
  3.   Flowable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() {
  6.       return 1;
  7.     }
  8.   }, null, Functions.emptyConsumer());
  9. }

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

  1. private void performTestUsingWithFlowableFactoryError(boolean disposeEagerly) {
  2.   final Runnable unsubscribe = mock(Runnable.class);
  3.   Callable<Disposable> resourceFactory = new Callable<Disposable>() {
  4.     @Override
  5.     public Disposable call() {
  6.       return Disposables.fromRunnable(unsubscribe);
  7.     }
  8.   };
  9.   Function<Disposable, Flowable<Integer>> observableFactory = new Function<Disposable, Flowable<Integer>>() {
  10.     @Override
  11.     public Flowable<Integer> apply(Disposable subscription) {
  12.       throw new TestException();
  13.     }
  14.   };
  15.   try {
  16.     Flowable.using(resourceFactory, observableFactory, disposeSubscription).blockingLast();
  17.     fail("Should throw a TestException when the observableFactory throws it");
  18.   } catch (TestException e) {
  19.     // Make sure that unsubscribe is called so that users can close
  20.     // the resource if some error happens.
  21.     verify(unsubscribe, times(1)).run();
  22.   }
  23. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingResourceSupplierNull() {
  3.   Flowable.using(null, new Function<Object, Publisher<Integer>>() {
  4.     @Override
  5.     public Publisher<Integer> apply(Object d) {
  6.       return just1;
  7.     }
  8.   }, Functions.emptyConsumer());
  9. }

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

  1. @Override
  2.   public Flowable<Object> apply(Flowable<Object> f)
  3.       throws Exception {
  4.     return Flowable.using(Functions.justCallable(1), Functions.justFunction(f), Functions.emptyConsumer());
  5.   }
  6. });

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingFlowableSupplierReturnsNull() {
  3.   Flowable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, Publisher<Object>>() {
  9.     @Override
  10.     public Publisher<Object> apply(Object d) {
  11.       return null;
  12.     }
  13.   }, Functions.emptyConsumer()).blockingLast();
  14. }

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

  1. @Test
  2. public void dispose() {
  3.   TestHelper.checkDisposed(Flowable.using(
  4.       new Callable<Object>() {
  5.         @Override
  6.         public Object call() throws Exception {
  7.           return 1;
  8.         }
  9.       },
  10.       new Function<Object, Flowable<Object>>() {
  11.         @Override
  12.         public Flowable<Object> apply(Object v) throws Exception {
  13.           return Flowable.never();
  14.         }
  15.       },
  16.       Functions.emptyConsumer()
  17.   ));
  18. }

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

  1. @Test
  2. public void eagerOnCompleteDisposerCrash() {
  3.   Flowable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() throws Exception {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, Flowable<Object>>() {
  9.     @Override
  10.     public Flowable<Object> apply(Object v) throws Exception {
  11.       return Flowable.empty();
  12.     }
  13.   }, new Consumer<Object>() {
  14.     @Override
  15.     public void accept(Object e) throws Exception {
  16.       throw new TestException("Second");
  17.     }
  18.   })
  19.   .test()
  20.   .assertFailureAndMessage(TestException.class, "Second");
  21. }

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

  1. @Override
  2.   public Publisher<Long> createPublisher(long elements) {
  3.     return
  4.       Flowable.using(Functions.justCallable(1),
  5.           Functions.justFunction(Flowable.fromIterable(iterate(elements))),
  6.           Functions.emptyConsumer()
  7.       )
  8.     ;
  9.   }
  10. }

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

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

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

  1. @Test
  2.   public void eagerDisposedOnError() {
  3.     final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  5.     Flowable.using(Functions.justCallable(1), Functions.justFunction(new Flowable<Integer>() {
  6.       @Override
  7.       protected void subscribeActual(Subscriber<? super Integer> subscriber) {
  8.         subscriber.onSubscribe(new BooleanSubscription());
  9.         ts.cancel();
  10.         subscriber.onError(new TestException());
  11.       }
  12.     }), Functions.emptyConsumer(), true)
  13.     .subscribe(ts);
  14.   }
  15. }

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

  1. @Test
  2. public void eagerDisposedOnComplete() {
  3.   final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  4.   Flowable.using(Functions.justCallable(1), Functions.justFunction(new Flowable<Integer>() {
  5.     @Override
  6.     protected void subscribeActual(Subscriber<? super Integer> subscriber) {
  7.       subscriber.onSubscribe(new BooleanSubscription());
  8.       ts.cancel();
  9.       subscriber.onComplete();
  10.     }
  11.   }), Functions.emptyConsumer(), true)
  12.   .subscribe(ts);
  13. }

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

  1. @Test
  2. public void supplierDisposerCrash() {
  3.   TestSubscriber<Object> ts = Flowable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() throws Exception {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, Flowable<Object>>() {
  9.     @Override
  10.     public Flowable<Object> apply(Object v) throws Exception {
  11.       throw new TestException("First");
  12.     }
  13.   }, new Consumer<Object>() {
  14.     @Override
  15.     public void accept(Object e) throws Exception {
  16.       throw new TestException("Second");
  17.     }
  18.   })
  19.   .test()
  20.   .assertFailure(CompositeException.class);
  21.   List<Throwable> errors = TestHelper.compositeList(ts.errors().get(0));
  22.   TestHelper.assertError(errors, 0, TestException.class, "First");
  23.   TestHelper.assertError(errors, 1, TestException.class, "Second");
  24. }

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

  1. @Test
  2. public void testUsingDisposesEagerlyBeforeError() {
  3.   final List<String> events = new ArrayList<String>();
  4.   Callable<Resource> resourceFactory = createResourceFactory(events);
  5.   final Consumer<Throwable> onError = createOnErrorAction(events);
  6.   final Action unsub = createUnsubAction(events);
  7.   Function<Resource, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
  8.     @Override
  9.     public Flowable<String> apply(Resource resource) {
  10.       return Flowable.fromArray(resource.getTextFromWeb().split(" "))
  11.           .concatWith(Flowable.<String>error(new RuntimeException()));
  12.     }
  13.   };
  14.   Subscriber<String> subscriber = TestHelper.mockSubscriber();
  15.   Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
  16.       new DisposeAction(), true)
  17.   .doOnCancel(unsub)
  18.   .doOnError(onError);
  19.   flowable.safeSubscribe(subscriber);
  20.   assertEquals(Arrays.asList("disposed", "error"), events);
  21. }

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

  1. @Test
  2. public void eagerOnErrorDisposerCrash() {
  3.   TestSubscriber<Object> ts = Flowable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() throws Exception {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, Flowable<Object>>() {
  9.     @Override
  10.     public Flowable<Object> apply(Object v) throws Exception {
  11.       return Flowable.error(new TestException("First"));
  12.     }
  13.   }, new Consumer<Object>() {
  14.     @Override
  15.     public void accept(Object e) throws Exception {
  16.       throw new TestException("Second");
  17.     }
  18.   })
  19.   .test()
  20.   .assertFailure(CompositeException.class);
  21.   List<Throwable> errors = TestHelper.compositeList(ts.errors().get(0));
  22.   TestHelper.assertError(errors, 0, TestException.class, "First");
  23.   TestHelper.assertError(errors, 1, TestException.class, "Second");
  24. }

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

  1. @Test
  2. public void testUsingDisposesEagerlyBeforeCompletion() {
  3.   final List<String> events = new ArrayList<String>();
  4.   Callable<Resource> resourceFactory = createResourceFactory(events);
  5.   final Action completion = createOnCompletedAction(events);
  6.   final Action unsub = createUnsubAction(events);
  7.   Function<Resource, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
  8.     @Override
  9.     public Flowable<String> apply(Resource resource) {
  10.       return Flowable.fromArray(resource.getTextFromWeb().split(" "));
  11.     }
  12.   };
  13.   Subscriber<String> subscriber = TestHelper.mockSubscriber();
  14.   Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
  15.       new DisposeAction(), true)
  16.   .doOnCancel(unsub)
  17.   .doOnComplete(completion);
  18.   flowable.safeSubscribe(subscriber);
  19.   assertEquals(Arrays.asList("disposed", "completed"), events);
  20. }

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

  1. @Test
  2. public void testUsingDoesNotDisposesEagerlyBeforeCompletion() {
  3.   final List<String> events = new ArrayList<String>();
  4.   Callable<Resource> resourceFactory = createResourceFactory(events);
  5.   final Action completion = createOnCompletedAction(events);
  6.   final Action unsub = createUnsubAction(events);
  7.   Function<Resource, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
  8.     @Override
  9.     public Flowable<String> apply(Resource resource) {
  10.       return Flowable.fromArray(resource.getTextFromWeb().split(" "));
  11.     }
  12.   };
  13.   Subscriber<String> subscriber = TestHelper.mockSubscriber();
  14.   Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
  15.       new DisposeAction(), false)
  16.   .doOnCancel(unsub)
  17.   .doOnComplete(completion);
  18.   flowable.safeSubscribe(subscriber);
  19.   assertEquals(Arrays.asList("completed", "disposed"), events);
  20. }

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

  1. @Test
  2. public void testUsingDoesNotDisposesEagerlyBeforeError() {
  3.   final List<String> events = new ArrayList<String>();
  4.   final Callable<Resource> resourceFactory = createResourceFactory(events);
  5.   final Consumer<Throwable> onError = createOnErrorAction(events);
  6.   final Action unsub = createUnsubAction(events);
  7.   Function<Resource, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
  8.     @Override
  9.     public Flowable<String> apply(Resource resource) {
  10.       return Flowable.fromArray(resource.getTextFromWeb().split(" "))
  11.           .concatWith(Flowable.<String>error(new RuntimeException()));
  12.     }
  13.   };
  14.   Subscriber<String> subscriber = TestHelper.mockSubscriber();
  15.   Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
  16.       new DisposeAction(), false)
  17.   .doOnCancel(unsub)
  18.   .doOnError(onError);
  19.   flowable.safeSubscribe(subscriber);
  20.   assertEquals(Arrays.asList("error", "disposed"), events);
  21. }

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

  1. @Test
  2. public void nonEagerDisposerCrash() {
  3.   List<Throwable> errors = TestHelper.trackPluginErrors();
  4.   try {
  5.     Flowable.using(new Callable<Object>() {
  6.       @Override
  7.       public Object call() throws Exception {
  8.         return 1;
  9.       }
  10.     }, new Function<Object, Flowable<Object>>() {
  11.       @Override
  12.       public Flowable<Object> apply(Object v) throws Exception {
  13.         return Flowable.empty();
  14.       }
  15.     }, new Consumer<Object>() {
  16.       @Override
  17.       public void accept(Object e) throws Exception {
  18.         throw new TestException("Second");
  19.       }
  20.     }, false)
  21.     .test()
  22.     .assertResult();
  23.     TestHelper.assertUndeliverable(errors, 0, TestException.class, "Second");
  24.   } finally {
  25.     RxJavaPlugins.reset();
  26.   }
  27. }

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

Flowable

相关文章

    查看更多

    热门标签

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

    最新文章

    更多

    Flowable类方法

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