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

Observable.single介绍

[英]Returns an Observable that emits the single item emitted by the source Observable, if that Observable emits only a single item. If the source Observable emits more than one item or no items, notify of an IllegalArgumentException or NoSuchElementException respectively.

Scheduler: single does not operate by default on a particular Scheduler.
[中]返回一个可观察对象，如果该可观察对象仅发射单个项，则该可观察对象发射源可观察对象发射的单个项。如果源Observable发出多个项或没有项，则分别通知IllegalArgumentException或NoSuchElementException。
调度程序：默认情况下，single不会在特定调度程序上运行。

代码示例

代码示例来源：origin: PipelineAI/pipeline

@Override
public Observable<Void> mapResponseToRequests(Observable<BatchReturnType> batchResponse, final Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests) {
  return batchResponse.single().doOnNext(new Action1<BatchReturnType>() {
    @Override
    public void call(BatchReturnType batchReturnType) {
      // this is a blocking call in HystrixCollapser
      self.mapResponseToRequests(batchReturnType, requests);
    }
  }).ignoreElements().cast(Void.class);
}

代码示例来源：origin: HotelsDotCom/styx

public static <T> CompletableFuture<T> fromSingleObservable(Observable<T> observable) {
  CompletableFuture<T> future = new CompletableFuture<>();
  observable.single().subscribe(future::complete, future::completeExceptionally);
  return future;
}

代码示例来源：origin: leeowenowen/rxjava-examples

@Override
 public void run() {
  Observable.just(1).single().subscribe(new Action1<Integer>() {
   @Override
   public void call(Integer integer) {
    log(integer);
   }
  });
 }
});

代码示例来源：origin: com.couchbase.client/java-client

@Override
public N1qlQueryResult query(N1qlQuery query, long timeout, TimeUnit timeUnit) {
  return Blocking.blockForSingle(
      couchbaseAsyncCluster
          .query(query)
          .flatMap(N1qlQueryExecutor.ASYNC_RESULT_TO_SYNC)
          .single(), timeout, timeUnit);
}

代码示例来源：origin: Qkyrie/spring-boot-netflix-example

@RequestMapping(method = GET, value = "/status")
public DeferredResult<String> getStatusPageUrl() {
  DeferredResult<String> result = new DeferredResult<>();
  notificationService.statusPageUrl().single()
      .subscribe(
          result::setResult,
          result::setErrorResult
      );
  return result;
}

代码示例来源：origin: alaisi/postgres-async-driver

Observable<Connection> connect(String username, String password, String database) {
  return stream.connect(new StartupMessage(username, database))
      .flatMap(message -> authenticate(username, password, message))
      .single(message -> message == ReadyForQuery.INSTANCE)
      .map(ready -> this);
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public BucketSettings updateBucket(BucketSettings settings, long timeout, TimeUnit timeUnit) {
  return Blocking.blockForSingle(asyncClusterManager.updateBucket(settings).single(), timeout, timeUnit);
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public DesignDocument upsertDesignDocument(final DesignDocument designDocument, final long timeout,
  final TimeUnit timeUnit) {
  return Blocking.blockForSingle(
    asyncBucketManager.upsertDesignDocument(designDocument).single(), timeout, timeUnit
  );
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public DesignDocument insertDesignDocument(final DesignDocument designDocument, final long timeout,
  final TimeUnit timeUnit) {
  return Blocking.blockForSingle(
    asyncBucketManager.insertDesignDocument(designDocument).single(), timeout, timeUnit
  );
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public DesignDocument insertDesignDocument(final DesignDocument designDocument, final boolean development,
  final long timeout, final TimeUnit timeUnit) {
  return Blocking.blockForSingle(
    asyncBucketManager.insertDesignDocument(designDocument, development).single(), timeout, timeUnit
  );
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public Boolean removeDesignDocument(final String name, final long timeout, final TimeUnit timeUnit) {
  return Blocking.blockForSingle(
    asyncBucketManager.removeDesignDocument(name).single(), timeout, timeUnit
  );
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public DesignDocument publishDesignDocument(final String name, final long timeout, final TimeUnit timeUnit) {
  return Blocking.blockForSingle(
    asyncBucketManager.publishDesignDocument(name).single(), timeout, timeUnit
  );
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public DesignDocument publishDesignDocument(final String name, final boolean overwrite, final long timeout,
  final TimeUnit timeUnit) {
  return Blocking.blockForSingle(
    asyncBucketManager.publishDesignDocument(name, overwrite).single(), timeout, timeUnit
  );
}

代码示例来源：origin: com.github.alaisi.pgasync/postgres-async-driver

Observable<Connection> connect(String username, String password, String database) {
  return stream.connect(new StartupMessage(username, database))
      .flatMap(message -> authenticate(username, password, message))
      .single(message -> message == ReadyForQuery.INSTANCE)
      .map(ready -> this);
}

代码示例来源：origin: com.microsoft.azure/azure-cosmosdb-gateway

protected Observable<DocumentProducerFeedResponse> produceOnSplit(Observable<DocumentProducer<T>> replacementProducers) {
  Observable<DocumentProducerFeedResponse> res = replacementProducers.toList().single().flatMap(documentProducers -> {
    RequestChargeTracker tracker = new RequestChargeTracker();
    return OrderByUtils.orderedMerge(resourceType, consumeComparer, tracker, documentProducers)
        .map(orderByQueryResult -> resultPageFrom(tracker, orderByQueryResult));
  });
  return res;
}

代码示例来源：origin: com.couchbase.client/java-client

@Override
public List<User> getUsers(AuthDomain domain, long timeout, TimeUnit timeUnit) {
  return Blocking.blockForSingle(asyncClusterManager.getUsers(domain).toList().single(), timeout, timeUnit);
}

代码示例来源：origin: com.netflix.hystrix/hystrix-core

@Override
public Observable<Void> mapResponseToRequests(Observable<BatchReturnType> batchResponse, final Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests) {
  return batchResponse.single().doOnNext(new Action1<BatchReturnType>() {
    @Override
    public void call(BatchReturnType batchReturnType) {
      // this is a blocking call in HystrixCollapser
      self.mapResponseToRequests(batchReturnType, requests);
    }
  }).ignoreElements().cast(Void.class);
}

代码示例来源：origin: nurkiewicz/rxjava-book-examples

@Test
public void sample_122() throws Exception {
  Observable<BigDecimal> totalPrice = Observable
      .just("bread", "butter", "milk", "tomato", "cheese")
      .subscribeOn(schedulerA)  //BROKEN!!!
      .map(prod -> rxGroceries.doPurchase(prod, 1))
      .reduce(BigDecimal::add)
      .single();
}

代码示例来源：origin: nurkiewicz/rxjava-book-examples

@Test
public void sample_135() throws Exception {
  final Observable<BigDecimal> totalPrice = Observable
      .just("bread", "butter", "milk", "tomato", "cheese")
      .subscribeOn(schedulerA)
      .flatMap(prod -> rxGroceries.purchase(prod, 1))
      .reduce(BigDecimal::add)
      .single();
}

代码示例来源：origin: nurkiewicz/rxjava-book-examples

@Test
public void sample_145() throws Exception {
  Observable<BigDecimal> totalPrice = Observable
      .just("bread", "butter", "milk", "tomato", "cheese")
      .flatMap(prod ->
          rxGroceries
              .purchase(prod, 1)
              .subscribeOn(schedulerA))
      .reduce(BigDecimal::add)
      .single();
}