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

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

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

Observable.using介绍

[英]Constructs an ObservableSource that creates a dependent resource object which is disposed of when the downstream calls dispose().

Scheduler: using does not operate by default on a particular Scheduler.
[中]构造一个ObservieSource，该资源创建一个依赖资源对象，当下游调用dispose（）时，该对象将被释放。
调度程序：默认情况下，使用不会在特定调度程序上运行。

代码示例

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingResourceSupplierNull() {
  3.   Observable.using(null, new Function<Object, Observable<Integer>>() {
  4.     @Override
  5.     public Observable<Integer> apply(Object d) {
  6.       return just1;
  7.     }
  8.   }, new Consumer<Object>() {
  9.     @Override
  10.     public void accept(Object d) { }
  11.   });
  12. }

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

  1. /**
  2.  * Constructs an ObservableSource that creates a dependent resource object which is disposed of when the downstream
  3.  * calls dispose().
  4.  * <p>
  5.  * <img width="640" height="400" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/using.png" alt="">
  6.  * <dl>
  7.  *  <dt><b>Scheduler:</b></dt>
  8.  *  <dd>{@code using} does not operate by default on a particular {@link Scheduler}.</dd>
  9.  * </dl>
  10.  *
  11.  * @param <T> the element type of the generated ObservableSource
  12.  * @param <D> the type of the resource associated with the output sequence
  13.  * @param resourceSupplier
  14.  *            the factory function to create a resource object that depends on the ObservableSource
  15.  * @param sourceSupplier
  16.  *            the factory function to create an ObservableSource
  17.  * @param disposer
  18.  *            the function that will dispose of the resource
  19.  * @return the ObservableSource whose lifetime controls the lifetime of the dependent resource object
  20.  * @see <a href="http://reactivex.io/documentation/operators/using.html">ReactiveX operators documentation: Using</a>
  21.  */
  22. @CheckReturnValue
  23. @SchedulerSupport(SchedulerSupport.NONE)
  24. public static <T, D> Observable<T> using(Callable<? extends D> resourceSupplier, Function<? super D, ? extends ObservableSource<? extends T>> sourceSupplier, Consumer<? super D> disposer) {
  25.   return using(resourceSupplier, sourceSupplier, disposer, true);
  26. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingObservableSupplierNull() {
  3.   Observable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() {
  6.       return 1;
  7.     }
  8.   }, null, new Consumer<Object>() {
  9.     @Override
  10.     public void accept(Object d) { }
  11.   });
  12. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingDisposeNull() {
  3.   Observable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, Observable<Integer>>() {
  9.     @Override
  10.     public Observable<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, Observable<Integer>> observableFactory = new Function<Disposable, Observable<Integer>>() {
  9.     @Override
  10.     public Observable<Integer> apply(Disposable d) {
  11.       return Observable.empty();
  12.     }
  13.   };
  14.   Observable.using(resourceFactory, observableFactory, disposeSubscription)
  15.   .blockingLast();
  16. }

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

  1. private void performTestUsingWithObservableFactoryError(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, Observable<Integer>> observableFactory = new Function<Disposable, Observable<Integer>>() {
  10.     @Override
  11.     public Observable<Integer> apply(Disposable subscription) {
  12.       throw new TestException();
  13.     }
  14.   };
  15.   try {
  16.     Observable.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. private void performTestUsingWithObservableFactoryErrorInOnSubscribe(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, Observable<Integer>> observableFactory = new Function<Disposable, Observable<Integer>>() {
  10.     @Override
  11.     public Observable<Integer> apply(Disposable subscription) {
  12.       return Observable.unsafeCreate(new ObservableSource<Integer>() {
  13.         @Override
  14.         public void subscribe(Observer<? super Integer> t1) {
  15.           throw new TestException();
  16.         }
  17.       });
  18.     }
  19.   };
  20.   try {
  21.     Observable
  22.     .using(resourceFactory, observableFactory, disposeSubscription, disposeEagerly)
  23.     .blockingLast();
  24.     fail("Should throw a TestException when the observableFactory throws it");
  25.   } catch (TestException e) {
  26.     // Make sure that unsubscribe is called so that users can close
  27.     // the resource if some error happens.
  28.     verify(unsubscribe, times(1)).run();
  29.   }
  30. }

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

  1. @Test(expected = NullPointerException.class)
  2. public void usingObservableSupplierReturnsNull() {
  3.   Observable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, Observable<Object>>() {
  9.     @Override
  10.     public Observable<Object> apply(Object d) {
  11.       return null;
  12.     }
  13.   }, new Consumer<Object>() {
  14.     @Override
  15.     public void accept(Object d) { }
  16.   }).blockingLast();
  17. }

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

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

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

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

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

  1. @Test
  2. public void eagerDisposedOnComplete() {
  3.   final TestObserver<Integer> to = new TestObserver<Integer>();
  4.   Observable.using(Functions.justCallable(1), Functions.justFunction(new Observable<Integer>() {
  5.     @Override
  6.     protected void subscribeActual(Observer<? super Integer> observer) {
  7.       observer.onSubscribe(Disposables.empty());
  8.       to.cancel();
  9.       observer.onComplete();
  10.     }
  11.   }), Functions.emptyConsumer(), true)
  12.   .subscribe(to);
  13. }

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

  1. @Test
  2.   public void eagerDisposedOnError() {
  3.     final TestObserver<Integer> to = new TestObserver<Integer>();
  5.     Observable.using(Functions.justCallable(1), Functions.justFunction(new Observable<Integer>() {
  6.       @Override
  7.       protected void subscribeActual(Observer<? super Integer> observer) {
  8.         observer.onSubscribe(Disposables.empty());
  9.         to.cancel();
  10.         observer.onError(new TestException());
  11.       }
  12.     }), Functions.emptyConsumer(), true)
  13.     .subscribe(to);
  14.   }
  15. }

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

  1. @Test
  2. public void eagerOnCompleteDisposerCrash() {
  3.   Observable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() throws Exception {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, ObservableSource<Object>>() {
  9.     @Override
  10.     public ObservableSource<Object> apply(Object v) throws Exception {
  11.       return Observable.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. @Test
  2. public void sourceSupplierReturnsNull() {
  3.   Observable.using(Functions.justCallable(1),
  4.       Functions.justFunction((Observable<Object>)null),
  5.       Functions.emptyConsumer())
  6.   .test()
  7.   .assertFailureAndMessage(NullPointerException.class, "The sourceSupplier returned a null ObservableSource")
  8.   ;
  9. }

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

  1. @Test
  2. public void eagerOnErrorDisposerCrash() {
  3.   TestObserver<Object> to = Observable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() throws Exception {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, ObservableSource<Object>>() {
  9.     @Override
  10.     public ObservableSource<Object> apply(Object v) throws Exception {
  11.       return Observable.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(to.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, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
  8.     @Override
  9.     public Observable<String> apply(Resource resource) {
  10.       return Observable.fromArray(resource.getTextFromWeb().split(" "));
  11.     }
  12.   };
  13.   Observer<String> observer = TestHelper.mockObserver();
  14.   Observable<String> o = Observable.using(resourceFactory, observableFactory,
  15.       new DisposeAction(), true)
  16.   .doOnDispose(unsub)
  17.   .doOnComplete(completion);
  18.   o.safeSubscribe(observer);
  19.   assertEquals(Arrays.asList("disposed", "completed" /* , "unsub" */), 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, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
  8.     @Override
  9.     public Observable<String> apply(Resource resource) {
  10.       return Observable.fromArray(resource.getTextFromWeb().split(" "));
  11.     }
  12.   };
  13.   Observer<String> observer = TestHelper.mockObserver();
  14.   Observable<String> o = Observable.using(resourceFactory, observableFactory,
  15.       new DisposeAction(), false)
  16.   .doOnDispose(unsub)
  17.   .doOnComplete(completion);
  18.   o.safeSubscribe(observer);
  19.   assertEquals(Arrays.asList("completed", /*"unsub",*/ "disposed"), events);
  20. }

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

  1. @Test
  2. public void supplierDisposerCrash() {
  3.   TestObserver<Object> to = Observable.using(new Callable<Object>() {
  4.     @Override
  5.     public Object call() throws Exception {
  6.       return 1;
  7.     }
  8.   }, new Function<Object, ObservableSource<Object>>() {
  9.     @Override
  10.     public ObservableSource<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(to.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, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
  8.     @Override
  9.     public Observable<String> apply(Resource resource) {
  10.       return Observable.fromArray(resource.getTextFromWeb().split(" "))
  11.           .concatWith(Observable.<String>error(new RuntimeException()));
  12.     }
  13.   };
  14.   Observer<String> observer = TestHelper.mockObserver();
  15.   Observable<String> o = Observable.using(resourceFactory, observableFactory,
  16.       new DisposeAction(), true)
  17.   .doOnDispose(unsub)
  18.   .doOnError(onError);
  19.   o.safeSubscribe(observer);
  20.   assertEquals(Arrays.asList("disposed", "error" /*, "unsub"*/), events);
  21. }

代码示例来源：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, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
  8.     @Override
  9.     public Observable<String> apply(Resource resource) {
  10.       return Observable.fromArray(resource.getTextFromWeb().split(" "))
  11.           .concatWith(Observable.<String>error(new RuntimeException()));
  12.     }
  13.   };
  14.   Observer<String> observer = TestHelper.mockObserver();
  15.   Observable<String> o = Observable.using(resourceFactory, observableFactory,
  16.       new DisposeAction(), false)
  17.   .doOnDispose(unsub)
  18.   .doOnError(onError);
  19.   o.safeSubscribe(observer);
  20.   assertEquals(Arrays.asList("error", /* "unsub",*/ "disposed"), events);
  21. }

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

Observable

相关文章

    查看更多

    热门标签

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

    最新文章

    更多

    Observable类方法

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