本文整理了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

@Test(expected = NullPointerException.class)
public void usingResourceSupplierNull() {
  Observable.using(null, new Function<Object, Observable<Integer>>() {
    @Override
    public Observable<Integer> apply(Object d) {
      return just1;
    }
  }, new Consumer<Object>() {
    @Override
    public void accept(Object d) { }
  });
}

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

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

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

@Test(expected = NullPointerException.class)
public void usingObservableSupplierNull() {
  Observable.using(new Callable<Object>() {
    @Override
    public Object call() {
      return 1;
    }
  }, null, new Consumer<Object>() {
    @Override
    public void accept(Object d) { }
  });
}

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

@Test(expected = NullPointerException.class)
public void usingDisposeNull() {
  Observable.using(new Callable<Object>() {
    @Override
    public Object call() {
      return 1;
    }
  }, new Function<Object, Observable<Integer>>() {
    @Override
    public Observable<Integer> apply(Object d) {
      return just1;
    }
  }, null);
}

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

private void performTestUsingWithResourceFactoryError(boolean disposeEagerly) {
  Callable<Disposable> resourceFactory = new Callable<Disposable>() {
    @Override
    public Disposable call() {
      throw new TestException();
    }
  };
  Function<Disposable, Observable<Integer>> observableFactory = new Function<Disposable, Observable<Integer>>() {
    @Override
    public Observable<Integer> apply(Disposable d) {
      return Observable.empty();
    }
  };
  Observable.using(resourceFactory, observableFactory, disposeSubscription)
  .blockingLast();
}

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

private void performTestUsingWithObservableFactoryError(boolean disposeEagerly) {
  final Runnable unsubscribe = mock(Runnable.class);
  Callable<Disposable> resourceFactory = new Callable<Disposable>() {
    @Override
    public Disposable call() {
      return Disposables.fromRunnable(unsubscribe);
    }
  };
  Function<Disposable, Observable<Integer>> observableFactory = new Function<Disposable, Observable<Integer>>() {
    @Override
    public Observable<Integer> apply(Disposable subscription) {
      throw new TestException();
    }
  };
  try {
    Observable.using(resourceFactory, observableFactory, disposeSubscription).blockingLast();
    fail("Should throw a TestException when the observableFactory throws it");
  } catch (TestException e) {
    // Make sure that unsubscribe is called so that users can close
    // the resource if some error happens.
    verify(unsubscribe, times(1)).run();
  }
}

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

private void performTestUsingWithObservableFactoryErrorInOnSubscribe(boolean disposeEagerly) {
  final Runnable unsubscribe = mock(Runnable.class);
  Callable<Disposable> resourceFactory = new Callable<Disposable>() {
    @Override
    public Disposable call() {
      return Disposables.fromRunnable(unsubscribe);
    }
  };
  Function<Disposable, Observable<Integer>> observableFactory = new Function<Disposable, Observable<Integer>>() {
    @Override
    public Observable<Integer> apply(Disposable subscription) {
      return Observable.unsafeCreate(new ObservableSource<Integer>() {
        @Override
        public void subscribe(Observer<? super Integer> t1) {
          throw new TestException();
        }
      });
    }
  };
  try {
    Observable
    .using(resourceFactory, observableFactory, disposeSubscription, disposeEagerly)
    .blockingLast();
    fail("Should throw a TestException when the observableFactory throws it");
  } catch (TestException e) {
    // Make sure that unsubscribe is called so that users can close
    // the resource if some error happens.
    verify(unsubscribe, times(1)).run();
  }
}

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

@Test(expected = NullPointerException.class)
public void usingObservableSupplierReturnsNull() {
  Observable.using(new Callable<Object>() {
    @Override
    public Object call() {
      return 1;
    }
  }, new Function<Object, Observable<Object>>() {
    @Override
    public Observable<Object> apply(Object d) {
      return null;
    }
  }, new Consumer<Object>() {
    @Override
    public void accept(Object d) { }
  }).blockingLast();
}

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

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

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

@Test
public void dispose() {
  TestHelper.checkDisposed(Observable.using(
      new Callable<Object>() {
        @Override
        public Object call() throws Exception {
          return 1;
        }
      },
      new Function<Object, ObservableSource<Object>>() {
        @Override
        public ObservableSource<Object> apply(Object v) throws Exception {
          return Observable.never();
        }
      },
      Functions.emptyConsumer()
  ));
}

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

@Test
public void eagerDisposedOnComplete() {
  final TestObserver<Integer> to = new TestObserver<Integer>();
  Observable.using(Functions.justCallable(1), Functions.justFunction(new Observable<Integer>() {
    @Override
    protected void subscribeActual(Observer<? super Integer> observer) {
      observer.onSubscribe(Disposables.empty());
      to.cancel();
      observer.onComplete();
    }
  }), Functions.emptyConsumer(), true)
  .subscribe(to);
}

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

@Test
  public void eagerDisposedOnError() {
    final TestObserver<Integer> to = new TestObserver<Integer>();

    Observable.using(Functions.justCallable(1), Functions.justFunction(new Observable<Integer>() {
      @Override
      protected void subscribeActual(Observer<? super Integer> observer) {
        observer.onSubscribe(Disposables.empty());
        to.cancel();
        observer.onError(new TestException());
      }
    }), Functions.emptyConsumer(), true)
    .subscribe(to);
  }
}

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

@Test
public void eagerOnCompleteDisposerCrash() {
  Observable.using(new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      return 1;
    }
  }, new Function<Object, ObservableSource<Object>>() {
    @Override
    public ObservableSource<Object> apply(Object v) throws Exception {
      return Observable.empty();
    }
  }, new Consumer<Object>() {
    @Override
    public void accept(Object e) throws Exception {
      throw new TestException("Second");
    }
  })
  .test()
  .assertFailureAndMessage(TestException.class, "Second");
}

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

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

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

@Test
public void eagerOnErrorDisposerCrash() {
  TestObserver<Object> to = Observable.using(new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      return 1;
    }
  }, new Function<Object, ObservableSource<Object>>() {
    @Override
    public ObservableSource<Object> apply(Object v) throws Exception {
      return Observable.error(new TestException("First"));
    }
  }, new Consumer<Object>() {
    @Override
    public void accept(Object e) throws Exception {
      throw new TestException("Second");
    }
  })
  .test()
  .assertFailure(CompositeException.class);
  List<Throwable> errors = TestHelper.compositeList(to.errors().get(0));
  TestHelper.assertError(errors, 0, TestException.class, "First");
  TestHelper.assertError(errors, 1, TestException.class, "Second");
}

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

@Test
public void testUsingDisposesEagerlyBeforeCompletion() {
  final List<String> events = new ArrayList<String>();
  Callable<Resource> resourceFactory = createResourceFactory(events);
  final Action completion = createOnCompletedAction(events);
  final Action unsub = createUnsubAction(events);
  Function<Resource, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
    @Override
    public Observable<String> apply(Resource resource) {
      return Observable.fromArray(resource.getTextFromWeb().split(" "));
    }
  };
  Observer<String> observer = TestHelper.mockObserver();
  Observable<String> o = Observable.using(resourceFactory, observableFactory,
      new DisposeAction(), true)
  .doOnDispose(unsub)
  .doOnComplete(completion);
  o.safeSubscribe(observer);
  assertEquals(Arrays.asList("disposed", "completed" /* , "unsub" */), events);
}

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

@Test
public void testUsingDoesNotDisposesEagerlyBeforeCompletion() {
  final List<String> events = new ArrayList<String>();
  Callable<Resource> resourceFactory = createResourceFactory(events);
  final Action completion = createOnCompletedAction(events);
  final Action unsub = createUnsubAction(events);
  Function<Resource, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
    @Override
    public Observable<String> apply(Resource resource) {
      return Observable.fromArray(resource.getTextFromWeb().split(" "));
    }
  };
  Observer<String> observer = TestHelper.mockObserver();
  Observable<String> o = Observable.using(resourceFactory, observableFactory,
      new DisposeAction(), false)
  .doOnDispose(unsub)
  .doOnComplete(completion);
  o.safeSubscribe(observer);
  assertEquals(Arrays.asList("completed", /*"unsub",*/ "disposed"), events);
}

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

@Test
public void supplierDisposerCrash() {
  TestObserver<Object> to = Observable.using(new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      return 1;
    }
  }, new Function<Object, ObservableSource<Object>>() {
    @Override
    public ObservableSource<Object> apply(Object v) throws Exception {
      throw new TestException("First");
    }
  }, new Consumer<Object>() {
    @Override
    public void accept(Object e) throws Exception {
      throw new TestException("Second");
    }
  })
  .test()
  .assertFailure(CompositeException.class);
  List<Throwable> errors = TestHelper.compositeList(to.errors().get(0));
  TestHelper.assertError(errors, 0, TestException.class, "First");
  TestHelper.assertError(errors, 1, TestException.class, "Second");
}

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

@Test
public void testUsingDisposesEagerlyBeforeError() {
  final List<String> events = new ArrayList<String>();
  Callable<Resource> resourceFactory = createResourceFactory(events);
  final Consumer<Throwable> onError = createOnErrorAction(events);
  final Action unsub = createUnsubAction(events);
  Function<Resource, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
    @Override
    public Observable<String> apply(Resource resource) {
      return Observable.fromArray(resource.getTextFromWeb().split(" "))
          .concatWith(Observable.<String>error(new RuntimeException()));
    }
  };
  Observer<String> observer = TestHelper.mockObserver();
  Observable<String> o = Observable.using(resourceFactory, observableFactory,
      new DisposeAction(), true)
  .doOnDispose(unsub)
  .doOnError(onError);
  o.safeSubscribe(observer);
  assertEquals(Arrays.asList("disposed", "error" /*, "unsub"*/), events);
}

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

@Test
public void testUsingDoesNotDisposesEagerlyBeforeError() {
  final List<String> events = new ArrayList<String>();
  final Callable<Resource> resourceFactory = createResourceFactory(events);
  final Consumer<Throwable> onError = createOnErrorAction(events);
  final Action unsub = createUnsubAction(events);
  Function<Resource, Observable<String>> observableFactory = new Function<Resource, Observable<String>>() {
    @Override
    public Observable<String> apply(Resource resource) {
      return Observable.fromArray(resource.getTextFromWeb().split(" "))
          .concatWith(Observable.<String>error(new RuntimeException()));
    }
  };
  Observer<String> observer = TestHelper.mockObserver();
  Observable<String> o = Observable.using(resourceFactory, observableFactory,
      new DisposeAction(), false)
  .doOnDispose(unsub)
  .doOnError(onError);
  o.safeSubscribe(observer);
  assertEquals(Arrays.asList("error", /* "unsub",*/ "disposed"), events);
}