import ratpack.exec.Promise; //导入依赖的package包/类
/**
 * @param promise
 * @param strategy The {@link BackpressureStrategy} to use
 * @param <T>
 * @return
 * @see RxRatpack#observe(Promise)
 */
public static <T> Flowable<T> flow(Promise<T> promise, BackpressureStrategy strategy) {
  return Flowable.create(subscriber ->
      promise.onError(subscriber::onError).then(value -> {
        subscriber.onNext(value);
        subscriber.onComplete();
      }),
    strategy);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
public void accept(Throwable e) throws Exception {
  if (e instanceof OnErrorNotImplementedException) {
    Promise.error(e.getCause()).then(Action.noop());
  } else if (e instanceof UndeliverableException) {
    Promise.error(e.getCause()).then(Action.noop());
  } else {
    Promise.error(e).then(Action.noop());
  }
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
public Promise<Void> backoff() {
    return Blocking.op(() -> {
        log.debug("Circuit is OPEN.  Waiting...");
        Thread.sleep(waitTime());
        mark();
    }).promise();
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
public Promise<Result> check(Registry registry) throws Exception {
    return eurekaService.isAvailable().map(available -> {
        if (available) {
            return Result.healthy();
        }
        return Result.unhealthy("EurekaService: Not available");
    });
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
private Promise<Void> consume(ReceiveMessageResult result) {
    List<Promise<Void>> promises = result.getMessages().stream()
        .map(this::consume)
        .collect(Collectors.toList());

    return SerialBatch.of(promises)
        .yieldAll()
        .flatMap(v -> Promise.value(null));
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private Promise<Void> deleteMessage(Message message) {
    return getQueueUrl()
        .map(url -> new DeleteMessageRequest(url, message.getReceiptHandle()))
        .flatMap(sqs::deleteMessage)
        .transform(transformer.recover(t -> new DeleteMessageResult()));
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
private Promise<ReceiveMessageRequest> getReceiveMessageRequest() {
    ReceiveMessageRequest request = consumer.getReceiveMessageRequest();
    if (request.getQueueUrl() == null || request.getQueueUrl().isEmpty()) {
        return getQueueUrl()
            .map((url) -> {
               request.withQueueUrl(url);
               return request;
            });
    }
    return Promise.value(request);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
public Promise<Book> getBook(String isbn) {
  // TODO - Implement getBooks function
  // Hint - checkout Blocking#get(Block) to see how to integrate blocking code with Ratpack
  // Hint - lab07.jooq.tables.Book.BOOK is the representation for the underlying `book` table
  // Hint - checkout SelectWhereStep#where(Condition...) and ResultQuery#fetchOneInto(Class)
  return Blocking.get(() ->
    create.select().from(BOOK).where(BOOK.ISBN.equal(isbn)).fetchOneInto(Book.class)
  );
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<SetTopicAttributesResult> setTopicAttributes(
    String topicArn,
    String attributeName,
    String attributeValue
) {
    return Blocking.get(() -> sns().setTopicAttributes(topicArn, attributeName, attributeValue))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
public Promise<StreamedResponse> requestStream(URI uri, Action<? super RequestSpec> action) {
    final AtomicReference<Span> span = new AtomicReference<>();

    return delegate
        .requestStream(uri, (RequestSpec requestSpec) -> {
            WrappedRequestSpec captor = new WrappedRequestSpec(handler, injector, requestSpec, span);
            // streamed request doesn't set the http method.
            // start span here until a better solution presents itself.
            span.set(handler.handleSend(injector, captor.getHeaders(), captor));
            action.execute(captor);
        })
        .wiretap(response -> streamedResponseWithSpan(response, span));
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<CreateTopicResult> createTopic(CreateTopicRequest request) {
    LOG.trace("creating sns topic request={}", request);
    return Blocking.get(() -> sns().createTopic(request))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<DeleteTopicResult> deleteTopic(DeleteTopicRequest request) {
    LOG.debug("deleting sns topic request={}", request);
    return Blocking.get(() -> sns().deleteTopic(request))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<AddPermissionResult> addPermission(
    String topicArn,
    String label,
    List<String> aWSAccountIds,
    List<String> actionNames
) {
    return Blocking.get(() -> sns().addPermission(topicArn, label, actionNames, actionNames))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<CheckIfPhoneNumberIsOptedOutResult> checkIfPhoneNumberIsOptedOut(
    CheckIfPhoneNumberIsOptedOutRequest request
) {
    return Blocking.get(() -> sns().checkIfPhoneNumberIsOptedOut(request))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<ConfirmSubscriptionResult> confirmSubscription(
    String topicArn,
    String token,
    String authenticateOnUnsubscribe
) {
    return Blocking.get(() -> sns().confirmSubscription(topicArn, token, authenticateOnUnsubscribe))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
public Promise<List<Book>> getBooks() {
  // TODO - Implement getBooks function
  // Hint - checkout Blocking#get(Block) to see how to integrate blocking code with Ratpack
  // Hint - lab07.jooq.tables.Book.BOOK is the representation for the underlying `book` table
  // Hint - checkout DSLContext#select() and ResultQuery#fetchInto(Class)
  return null;
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<CreatePlatformApplicationResult> createPlatformApplication(
    CreatePlatformApplicationRequest request
) {
    return Blocking.get(() -> sns().createPlatformApplication(request))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<SetPlatformApplicationAttributesResult> setPlatformApplicationAttributes(
    SetPlatformApplicationAttributesRequest request
) {
    return Blocking.get(() -> sns().setPlatformApplicationAttributes(request))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<SetSubscriptionAttributesResult> setSubscriptionAttributes(
    SetSubscriptionAttributesRequest request
) {
    return Blocking.get(() -> sns().setSubscriptionAttributes(request))
        .transform(breaker);
}
 

import ratpack.exec.Promise; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public Promise<GetPlatformApplicationAttributesResult> getPlatformApplicationAttributes(
    GetPlatformApplicationAttributesRequest request
) {
    return Blocking.get(() -> sns().getPlatformApplicationAttributes(request))
        .transform(breaker);
}