io.vertx.core.AsyncResult#succeeded ( )源码实例Demo

/**
 * Handler for connection opened by remote
 *
 * @param ar async result with info on related Proton connection
 */
private void processOpenConnection(AsyncResult<ProtonConnection> ar) {

    if (ar.succeeded()) {

        log.info("Connection opened by {} {}", ar.result().getRemoteHostname(), ar.result().getRemoteContainer());

        ProtonConnection connection = ar.result();
        connection.open();

        // new connection, preparing for hosting related sink/source endpoints
        if (!this.endpoints.containsKey(connection)) {
            this.endpoints.put(connection, new ConnectionEndpoint());
        }
    }
}
 

/**
 * Log the reconciliation outcome.
 */
private void handleResult(Reconciliation reconciliation, AsyncResult<Void> result, Timer.Sample reconciliationTimerSample) {
    if (result.succeeded()) {
        successfulReconciliationsCounter.increment();
        reconciliationTimerSample.stop(reconciliationsTimer);
        log.info("{}: reconciled", reconciliation);
    } else {
        Throwable cause = result.cause();
        if (cause instanceof InvalidConfigParameterException) {
            failedReconciliationsCounter.increment();
            reconciliationTimerSample.stop(reconciliationsTimer);
            log.warn("{}: Failed to reconcile {}", reconciliation, cause.getMessage());
        } else if (cause instanceof UnableToAcquireLockException) {
            lockedReconciliationsCounter.increment();
        } else  {
            failedReconciliationsCounter.increment();
            reconciliationTimerSample.stop(reconciliationsTimer);
            log.warn("{}: Failed to reconcile", reconciliation, cause);
        }
    }
}
 

@Override
public Future<KafkaTopic> createResource(KafkaTopic topicResource) {
    Promise<KafkaTopic> handler = Promise.promise();
    AsyncResult<Void> response = createResponse.apply(new ResourceName(topicResource));
    if (response.succeeded()) {
        AsyncResult<KafkaTopic> old = byName.put(new ResourceName(topicResource), Future.succeededFuture(topicResource));
        if (old != null) {
            handler.handle(Future.failedFuture("resource already existed: " + topicResource.getMetadata().getName()));
            return handler.future();
        }
    }
    if (response.succeeded()) {
        handler.complete(new KafkaTopicBuilder(topicResource).editMetadata().withGeneration(1L).endMetadata().build());
    } else {
        handler.fail(response.cause());
    }
    return handler.future();
}
 

@Override
public Future<KafkaTopic> updateResource(KafkaTopic topicResource) {
    Promise<KafkaTopic> handler = Promise.promise();
    AsyncResult<Void> response = modifyResponse.apply(new ResourceName(topicResource));
    if (response.succeeded()) {
        AsyncResult<KafkaTopic> old = byName.put(new ResourceName(topicResource), Future.succeededFuture(topicResource));
        if (old == null) {
            handler.handle(Future.failedFuture("resource does not exist, cannot be updated: " + topicResource.getMetadata().getName()));
            return handler.future();
        }
    }
    if (response.succeeded()) {
        Long generation = topicResource.getMetadata().getGeneration();
        handler.complete(new KafkaTopicBuilder(topicResource)
                .editMetadata()
                    .withGeneration(generation != null ? generation + 1 : 1)
                .endMetadata()
            .build());
    } else {
        handler.fail(response.cause());
    }
    return handler.future();
}
 

/**
 * Completes this promise with the given result and stops the expiration timer.
 *
 * @param connectionResult The connection result to complete this promise with.
 */
public void tryCompleteAndCancelTimer(final AsyncResult<HonoConnection> connectionResult) {
    if (timerId != null) {
        vertx.cancelTimer(timerId);
    }
    if (connectionResult.succeeded()) {
        promise.tryComplete();
    } else {
        promise.tryFail(connectionResult.cause());
    }
}
 

private void handleRetryResult(long retryInterval, long next, AsyncResult<Void> retryResult,
                               Promise<Void> promise) {
    if (retryResult.succeeded()) {
        promise.complete();
    } else {
        retryDownload(promise, retryInterval, next);
    }
}
 

private void completeFromAsyncResult(AsyncResult<T> ar) {
    if (ar.succeeded()) {
        super.complete(ar.result());
    } else {
        super.completeExceptionally(ar.cause());
    }
}
 

private void completeFromAsyncResult(AsyncResult<T> ar) {
  if (ar.succeeded()) {
    super.complete(ar.result());
  } else {
    super.completeExceptionally(ar.cause());
  }
}
 

private void completeFromAsyncResult(AsyncResult<T> ar) {
    if (ar.succeeded()) {
        super.complete(ar.result());
    } else {
        super.completeExceptionally(ar.cause());
    }
}
 

private void onCreateCacheResponse(AsyncResult<Void> ar) {
  if (ar.succeeded()) {
    future.complete(createResult);
    return;
  }

  future.completeExceptionally(ar.cause());
}
 

@Override
public void handle(AsyncResult<Connection> ar) {
  if (ar.succeeded()) {
    conn = ar.result();
  } else {
    failure = ar.cause();
  }
}
 

/**
 * Invoked when a client closes the connection with this server.
 * <p>
 * This implementation closes and disconnects the connection.
 *
 * @param con The connection to close.
 * @param res The client's close frame.
 */
protected void handleRemoteConnectionClose(final ProtonConnection con, final AsyncResult<ProtonConnection> res) {
    if (res.succeeded()) {
        log.debug("client [container: {}] closed connection", con.getRemoteContainer());
    } else {
        log.debug("client [container: {}] closed connection with error", con.getRemoteContainer(), res.cause());
    }
    con.close();
    con.disconnect();
    publishConnectionClosedEvent(con);
}
 

private void completeFromAsyncResult(AsyncResult<T> ar) {
    if (ar.succeeded()) {
        super.complete(ar.result());
    } else {
        super.completeExceptionally(ar.cause());
    }
}
 

/**
 * Handler for connection closed by remote
 *
 * @param ar async result with info on related Proton connection
 */
private void processCloseConnection(AsyncResult<ProtonConnection> ar) {

    if (ar.succeeded()) {
        log.info("Connection closed by {} {}", ar.result().getRemoteHostname(), ar.result().getRemoteContainer());
        this.closeConnectionEndpoint(ar.result());
    }
}
 

/**
 * Invoked when a client closes the connection with this server.
 *
 * @param con The connection to close.
 * @param res The client's close frame.
 */
private void handleRemoteConnectionClose(final ProtonConnection con, final AsyncResult<ProtonConnection> res) {

    if (res.succeeded()) {
        log.debug("client [container: {}] closed connection", con.getRemoteContainer());
    } else {
        log.debug("client [container: {}] closed connection with error", con.getRemoteContainer(), res.cause());
    }
    con.disconnectHandler(null);
    con.close();
    con.disconnect();
}
 

private static <T> void statefulExecution(
    String methodId,
    String targetId,
    Object message,
    ThrowableFutureBiConsumer<AsyncResult<Message<Object>>, T> objectFunction,
    DeliveryOptions requestDeliveryOptions,
    VxmsShared vxmsShared,
    Consumer<Throwable> errorMethodHandler,
    Message<Object> requestMessage,
    Encoder encoder,
    Consumer<Throwable> errorHandler,
    ThrowableErrorConsumer<Throwable, T> onFailureRespond,
    DeliveryOptions responseDeliveryOptions,
    int retryCount,
    long timeout,
    long circuitBreakerTimeout,
    RecursiveExecutor<T> executor,
    RetryExecutor<T> retry,
    AsyncResult<Message<Object>> event,
    ThrowableFutureConsumer<T> objectConsumer) {
  if (event.succeeded()) {
    executor.execute(
        methodId,
        vxmsShared,
        event.cause(),
        errorMethodHandler,
        requestMessage,
        objectConsumer,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        circuitBreakerTimeout);
  } else {
    statefulErrorHandling(
        methodId,
        targetId,
        message,
        objectFunction,
        requestDeliveryOptions,
        vxmsShared,
        event.cause(),
        errorMethodHandler,
        requestMessage,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        circuitBreakerTimeout,
        executor,
        retry,
        event);
  }
}
 

private static <T> void statelessExecution(
    String methodId,
    String targetId,
    Object message,
    ThrowableFunction<AsyncResult<Message<Object>>, T> function,
    DeliveryOptions deliveryOptions,
    VxmsShared vxmsShared,
    Consumer<Throwable> errorMethodHandler,
    Message<Object> requestMessage,
    Encoder encoder,
    Consumer<Throwable> errorHandler,
    ThrowableFunction<Throwable, T> onFailureRespond,
    DeliveryOptions responseDeliveryOptions,
    int retryCount,
    long timeout,
    long delay,
    long circuitBreakerTimeout,
    RecursiveExecutor<T> executor,
    RetryExecutor<T> retry,
    AsyncResult<Message<Object>> event,
    ThrowableSupplier<T> byteSupplier) {
  if (event.succeeded() || (event.failed() && retryCount <= 0)) {
    executor.execute(
        methodId,
        vxmsShared,
        event.cause(),
        errorMethodHandler,
        requestMessage,
        byteSupplier,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        delay,
        circuitBreakerTimeout);
  } else if (event.failed() && retryCount > 0) {
    retryFunction(
        methodId, targetId,
        message,
        function,
        deliveryOptions,
        vxmsShared,
        event.cause(),
        errorMethodHandler,
        requestMessage,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        delay,
        circuitBreakerTimeout,
        retry);
  }
}
 

private void handleResult(AsyncResult<VideoResponse> responseResult, VideoEvent.VideoEventBuilder videoEventBuilder,
                          RoutingContext context, long startTime) {
    final boolean responseSucceeded = responseResult.succeeded();
    final MetricName metricRequestStatus;
    final List<String> errorMessages;
    final int status;
    final String body;

    if (responseSucceeded) {
        metricRequestStatus = MetricName.ok;
        errorMessages = Collections.emptyList();

        status = HttpResponseStatus.OK.code();
        context.response().headers().add(HttpUtil.CONTENT_TYPE_HEADER, HttpHeaderValues.APPLICATION_JSON);
        body = mapper.encode(responseResult.result());
    } else {
        final Throwable exception = responseResult.cause();
        if (exception instanceof InvalidRequestException) {
            metricRequestStatus = MetricName.badinput;
            errorMessages = ((InvalidRequestException) exception).getMessages();
            logger.info("Invalid request format: {0}", errorMessages);

            status = HttpResponseStatus.BAD_REQUEST.code();
            body = errorMessages.stream()
                    .map(msg -> String.format("Invalid request format: %s", msg))
                    .collect(Collectors.joining("\n"));
        } else if (exception instanceof UnauthorizedAccountException) {
            metricRequestStatus = MetricName.badinput;
            final String errorMessage = exception.getMessage();
            logger.info("Unauthorized: {0}", errorMessage);

            errorMessages = Collections.singletonList(errorMessage);

            status = HttpResponseStatus.UNAUTHORIZED.code();
            body = String.format("Unauthorised: %s", errorMessage);
        } else {
            metricRequestStatus = MetricName.err;
            logger.error("Critical error while running the auction", exception);

            final String message = exception.getMessage();
            errorMessages = Collections.singletonList(message);

            status = HttpResponseStatus.INTERNAL_SERVER_ERROR.code();
            body = String.format("Critical error while running the auction: %s", message);
        }
    }
    final VideoEvent videoEvent = videoEventBuilder.status(status).errors(errorMessages).build();
    respondWith(context, status, body, startTime, metricRequestStatus, videoEvent);
}
 

private static <T> void statelessExecution(
    String methodId,
    String id,
    Object message,
    ThrowableFutureBiConsumer<AsyncResult<Message<Object>>, T> stringFunction,
    DeliveryOptions options,
    VxmsShared vxmsShared,
    Throwable t,
    Consumer<Throwable> errorMethodHandler,
    RoutingContext context,
    Map<String, String> headers,
    Encoder encoder,
    Consumer<Throwable> errorHandler,
    ThrowableErrorConsumer<Throwable, T> onFailureRespond,
    int httpStatusCode,
    int httpErrorCode,
    int retryCount,
    long timeout,
    long circuitBreakerTimeout,
    RecursiveExecutor<T> executor,
    RetryExecutor<T> retry,
    AsyncResult<Message<Object>> event,
    ThrowableFutureConsumer<T> stringSupplier) {
  if (event.succeeded() || (event.failed() && retryCount <= 0)) {
    executor.execute(
        methodId,
        vxmsShared,
        t,
        errorMethodHandler,
        context,
        headers,
        stringSupplier,
        null,
        encoder,
        errorHandler,
        onFailureRespond,
        httpStatusCode,
        httpErrorCode,
        retryCount,
        timeout,
        circuitBreakerTimeout);
  } else if (event.failed() && retryCount > 0) {
    // retry operation
    final Throwable cause = event.cause();
    retryOperation(
        methodId,
        id,
        message,
        stringFunction,
        options,
        vxmsShared,
        cause,
        errorMethodHandler,
        context,
        headers,
        encoder,
        errorHandler,
        onFailureRespond,
        httpStatusCode,
        httpErrorCode,
        retryCount,
        timeout,
        circuitBreakerTimeout,
        retry);
  }
}
 

private static <T> void statefulExecution(
    String methodId,
    String targetId,
    Object message,
    ThrowableFunction<AsyncResult<Message<Object>>, T> function,
    DeliveryOptions deliveryOptions,
    VxmsShared vxmsShared,
    Consumer<Throwable> errorMethodHandler,
    Message<Object> requestMessage,
    Encoder encoder,
    Consumer<Throwable> errorHandler,
    ThrowableFunction<Throwable, T> onFailureRespond,
    DeliveryOptions responseDeliveryOptions,
    int retryCount,
    long timeout,
    long delay,
    long circuitBreakerTimeout,
    RecursiveExecutor<T> executor,
    RetryExecutor<T> retry,
    AsyncResult<Message<Object>> event,
    ThrowableSupplier<T> supplier) {
  if (event.succeeded()) {
    executor.execute(
        methodId,
        vxmsShared,
        event.cause(),
        errorMethodHandler,
        requestMessage,
        supplier,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        delay,
        circuitBreakerTimeout);
  } else {
    statefulErrorHandling(
        methodId, targetId,
        message,
        function,
        deliveryOptions,
        vxmsShared,
        errorMethodHandler,
        requestMessage,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        delay,
        circuitBreakerTimeout,
        executor,
        retry,
        event);
  }
}