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

@Test(expected = NullPointerException.class)
public void usingDisposeNull() {
  Flowable.using(new Callable<Object>() {
    @Override
    public Object call() {
      return 1;
    }
  }, new Function<Object, Publisher<Integer>>() {
    @Override
    public Publisher<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, Flowable<Integer>> observableFactory = new Function<Disposable, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Disposable d) {
      return Flowable.empty();
    }
  };
  Flowable.using(resourceFactory, observableFactory, disposeSubscription)
  .blockingLast();
}

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

@Test(expected = NullPointerException.class)
public void usingFlowableSupplierNull() {
  Flowable.using(new Callable<Object>() {
    @Override
    public Object call() {
      return 1;
    }
  }, null, Functions.emptyConsumer());
}

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

private void performTestUsingWithFlowableFactoryError(boolean disposeEagerly) {
  final Runnable unsubscribe = mock(Runnable.class);
  Callable<Disposable> resourceFactory = new Callable<Disposable>() {
    @Override
    public Disposable call() {
      return Disposables.fromRunnable(unsubscribe);
    }
  };
  Function<Disposable, Flowable<Integer>> observableFactory = new Function<Disposable, Flowable<Integer>>() {
    @Override
    public Flowable<Integer> apply(Disposable subscription) {
      throw new TestException();
    }
  };
  try {
    Flowable.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

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

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

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

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

@Test(expected = NullPointerException.class)
public void usingFlowableSupplierReturnsNull() {
  Flowable.using(new Callable<Object>() {
    @Override
    public Object call() {
      return 1;
    }
  }, new Function<Object, Publisher<Object>>() {
    @Override
    public Publisher<Object> apply(Object d) {
      return null;
    }
  }, Functions.emptyConsumer()).blockingLast();
}

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

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

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

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

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

@Override
  public Publisher<Long> createPublisher(long elements) {
    return
      Flowable.using(Functions.justCallable(1),
          Functions.justFunction(Flowable.fromIterable(iterate(elements))),
          Functions.emptyConsumer()
      )
    ;
  }
}

代码示例来源：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 eagerDisposedOnError() {
    final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();

    Flowable.using(Functions.justCallable(1), Functions.justFunction(new Flowable<Integer>() {
      @Override
      protected void subscribeActual(Subscriber<? super Integer> subscriber) {
        subscriber.onSubscribe(new BooleanSubscription());
        ts.cancel();
        subscriber.onError(new TestException());
      }
    }), Functions.emptyConsumer(), true)
    .subscribe(ts);
  }
}

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

@Test
public void eagerDisposedOnComplete() {
  final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  Flowable.using(Functions.justCallable(1), Functions.justFunction(new Flowable<Integer>() {
    @Override
    protected void subscribeActual(Subscriber<? super Integer> subscriber) {
      subscriber.onSubscribe(new BooleanSubscription());
      ts.cancel();
      subscriber.onComplete();
    }
  }), Functions.emptyConsumer(), true)
  .subscribe(ts);
}

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

@Test
public void supplierDisposerCrash() {
  TestSubscriber<Object> ts = Flowable.using(new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      return 1;
    }
  }, new Function<Object, Flowable<Object>>() {
    @Override
    public Flowable<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(ts.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, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
    @Override
    public Flowable<String> apply(Resource resource) {
      return Flowable.fromArray(resource.getTextFromWeb().split(" "))
          .concatWith(Flowable.<String>error(new RuntimeException()));
    }
  };
  Subscriber<String> subscriber = TestHelper.mockSubscriber();
  Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
      new DisposeAction(), true)
  .doOnCancel(unsub)
  .doOnError(onError);
  flowable.safeSubscribe(subscriber);
  assertEquals(Arrays.asList("disposed", "error"), events);
}

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

@Test
public void eagerOnErrorDisposerCrash() {
  TestSubscriber<Object> ts = Flowable.using(new Callable<Object>() {
    @Override
    public Object call() throws Exception {
      return 1;
    }
  }, new Function<Object, Flowable<Object>>() {
    @Override
    public Flowable<Object> apply(Object v) throws Exception {
      return Flowable.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(ts.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, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
    @Override
    public Flowable<String> apply(Resource resource) {
      return Flowable.fromArray(resource.getTextFromWeb().split(" "));
    }
  };
  Subscriber<String> subscriber = TestHelper.mockSubscriber();
  Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
      new DisposeAction(), true)
  .doOnCancel(unsub)
  .doOnComplete(completion);
  flowable.safeSubscribe(subscriber);
  assertEquals(Arrays.asList("disposed", "completed"), 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, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
    @Override
    public Flowable<String> apply(Resource resource) {
      return Flowable.fromArray(resource.getTextFromWeb().split(" "));
    }
  };
  Subscriber<String> subscriber = TestHelper.mockSubscriber();
  Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
      new DisposeAction(), false)
  .doOnCancel(unsub)
  .doOnComplete(completion);
  flowable.safeSubscribe(subscriber);
  assertEquals(Arrays.asList("completed", "disposed"), 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, Flowable<String>> observableFactory = new Function<Resource, Flowable<String>>() {
    @Override
    public Flowable<String> apply(Resource resource) {
      return Flowable.fromArray(resource.getTextFromWeb().split(" "))
          .concatWith(Flowable.<String>error(new RuntimeException()));
    }
  };
  Subscriber<String> subscriber = TestHelper.mockSubscriber();
  Flowable<String> flowable = Flowable.using(resourceFactory, observableFactory,
      new DisposeAction(), false)
  .doOnCancel(unsub)
  .doOnError(onError);
  flowable.safeSubscribe(subscriber);
  assertEquals(Arrays.asList("error", "disposed"), events);
}

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

@Test
public void nonEagerDisposerCrash() {
  List<Throwable> errors = TestHelper.trackPluginErrors();
  try {
    Flowable.using(new Callable<Object>() {
      @Override
      public Object call() throws Exception {
        return 1;
      }
    }, new Function<Object, Flowable<Object>>() {
      @Override
      public Flowable<Object> apply(Object v) throws Exception {
        return Flowable.empty();
      }
    }, new Consumer<Object>() {
      @Override
      public void accept(Object e) throws Exception {
        throw new TestException("Second");
      }
    }, false)
    .test()
    .assertResult();
    TestHelper.assertUndeliverable(errors, 0, TestException.class, "Second");
  } finally {
    RxJavaPlugins.reset();
  }
}