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

/**
 * In case of failure retries HTTP request passing itself as handler.
 *
 * @param result  result of asynchronous HTTP request execution
 */
@Override
@SuppressWarnings("unchecked")
public void handle(AsyncResult<T> result) {
  if (result.succeeded()) {
    this.resultFuture.complete(result.result());
  } else {
    try {
      throw result.cause();
    } catch (final RetryableException retryableException) {
      try {
        this.retryer.continueOrPropagate(retryableException);
        logRetry();
        ((Future<T>) executeAndDecode(this.template)).setHandler(this);
      } catch (final RetryableException noMoreRetryAttempts) {
        this.resultFuture.fail(noMoreRetryAttempts);
      }
    } catch (final Throwable otherException) {
      this.resultFuture.fail(otherException);
    }
  }
}
 

/**
 * 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);
        }
    }
}
 

private PreBidResponse bidResponseOrError(AsyncResult<PreBidResponse> responseResult) {
    final MetricName responseStatus;

    final PreBidResponse result;

    if (responseResult.succeeded()) {
        responseStatus = MetricName.ok;
        result = responseResult.result();
    } else {
        final Throwable exception = responseResult.cause();
        final boolean isRequestInvalid = exception instanceof InvalidRequestException;

        responseStatus = isRequestInvalid ? MetricName.badinput : MetricName.err;

        if (!isRequestInvalid) {
            logger.error("Failed to process /auction request", exception);
        }

        result = error(isRequestInvalid || exception instanceof PreBidException
                ? exception.getMessage()
                : "Unexpected server error");
    }

    updateRequestMetric(responseStatus);

    return result;
}
 

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

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

private void forwardErrorCode(RoutingContext routingContext, AsyncResult<Message<Object>> reply) {
    ReplyException ex = (ReplyException) reply.cause();
    ex.printStackTrace();
    HttpServerResponse response = routingContext.response();
    response.setStatusCode(ex.failureCode());
    response.end();
}
 

private static void handleStartResult(AsyncResult<HttpServer> startResult,
    CompletableFuture<Integer> startFuture, int requestedPort) {
  // Complete the future
  completeFuture(startResult.map(HttpServer::actualPort), startFuture);

  // Log the event
  if (startResult.succeeded()) {
    HttpServer server = startResult.result();
    logger.info("Java server is listening at port {}", server.actualPort());
  } else {
    Throwable failureCause = startResult.cause();
    logger.error("Java server failed to start listening at port {}", requestedPort,
        failureCause);
  }
}
 

/**
 * Combines two completed AsyncResults, at least one of which has failed, returning
 * a single cause, possibly with suppressed exception.
 * If both AsyncResults have failed {@code primary} will be the main cause of failure and
 * {@code secondary} will be a suppressed exception.
 * @param primary The primary failure.
 * @param secondary The secondary failure.
 * @return The cause.
 */
Throwable collectCauses(AsyncResult<? extends Object> primary,
                        AsyncResult<? extends Object> secondary) {
    Throwable cause = primary.cause();
    if (cause == null) {
        cause = secondary.cause();
    } else {
        if (secondary.failed()) {
            cause.addSuppressed(secondary.cause());
        }
    }
    return 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());
    }
}
 

public static <T> FutureHandler<T, AsyncResult<T>> asyncResult() {
  return new FutureHandler<T, AsyncResult<T>>() {
    @Override
    synchronized public void handle(AsyncResult<T> t) {
      if (t.succeeded()) {
        result = t.result();
      } else {
        exception = new ExecutionException(t.cause());
      }
      latch.countDown();
    }
  };
}
 

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 static <T> ThrowableSupplier<T> createSupplier(
    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,
    RetryExecutor<T> retry,
    AsyncResult<Message<Object>> event) {
  return () -> {
    T resp = null;
    if (event.failed()) {
      if (retryCount > 0) {
        retryFunction(
            methodId, targetId,
            message,
            function,
            deliveryOptions,
            vxmsShared,
            event.cause(),
            errorMethodHandler,
            requestMessage,
            encoder,
            errorHandler,
            onFailureRespond,
            responseDeliveryOptions,
            retryCount,
            timeout,
            delay,
            circuitBreakerTimeout,
            retry);
      } else {
        throw event.cause();
      }
    } else {
      resp = function.apply(event);
    }

    return resp;
  };
}
 

private static <T> ThrowableSupplier<T> createSupplier(
    String methodId,
    String targetId,
    Object message,
    ThrowableFunction<AsyncResult<Message<Object>>, T> function,
    DeliveryOptions deliveryOptions,
    VxmsShared vxmsShared,
    Throwable t,
    Consumer<Throwable> errorMethodHandler,
    RoutingContext context,
    Map<String, String> headers,
    Encoder encoder,
    Consumer<Throwable> errorHandler,
    ThrowableFunction<Throwable, T> onFailureRespond,
    int httpStatusCode,
    int httpErrorCode,
    int retryCount,
    long timeout,
    long delay,
    long circuitBreakerTimeout,
    RetryExecutor<T> retry,
    AsyncResult<Message<Object>> event) {
  return () -> {
    T resp = null;
    if (event.failed()) {
      if (retryCount > 0) {
        retryOperation(
            methodId,
            targetId,
            message,
            function,
            deliveryOptions,
            vxmsShared,
            t,
            errorMethodHandler,
            context,
            headers,
            encoder,
            errorHandler,
            onFailureRespond,
            httpStatusCode,
            httpErrorCode,
            retryCount,
            timeout,
            delay,
            circuitBreakerTimeout,
            retry);
      } else {
        throw event.cause();
      }
    } else {
      resp = function.apply(event);
    }

    return resp;
  };
}
 

void processLoadEvent(Callback<ConditionalOperation> callback, LoadFeaturesEvent event, AsyncResult<XyzResponse> r) {
  final Map<String, String> idsMap = event.getIdsMap();
  if (r.failed()) {
    if (r.cause() instanceof Exception) {
      callback.exception((Exception) r.cause());
    } else {
      callback.exception(new Exception(r.cause()));
    }
    return;
  }

  try {
    final XyzResponse response = r.result();
    if (!(response instanceof FeatureCollection)) {
      callback.exception(Api.responseToHttpException(response));
      return;
    }
    final FeatureCollection collection = (FeatureCollection) response;
    final List<Feature> features = collection.getFeatures();

    // For each input feature there could be 0, 1(head state) or 2 (head state and base state) features in the response
    if (features == null) {
      callback.call(this);
      return;
    }

    for (final Feature feature : features) {
      final String id = feature.getId();

      // The uuid the client has requested.
      final String requestedUuid = idsMap.get(id);

      int position = getPositionForId(feature.getId());
      if (position == -1) { // There is no object with this ID in the input states
        continue;
      }

      if (feature.getProperties() == null || feature.getProperties().getXyzNamespace() == null) {
        throw new IllegalStateException("Received a feature with missing space namespace properties for object '" + id + "'");
      }

      String uuid = feature.getProperties().getXyzNamespace().getUuid();

      // Set the head state( i.e. the latest version in the database )
      if (modifyOp.entries.get(position).head == null || uuid != null && !uuid.equals(requestedUuid)) {
        modifyOp.entries.get(position).head = feature;
      }

      // Set the base state( i.e. the original version that the user was editing )
      // Note: The base state must not be empty. If the connector doesn't support history and doesn't return the base state, use the
      // head state instead.
      if (modifyOp.entries.get(position).base == null || uuid != null && uuid.equals(requestedUuid)) {
        modifyOp.entries.get(position).base = feature;
      }
    }

    callback.call(this);
  } catch (Exception e) {
    callback.exception(e);
  }
}
 

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,
    ThrowableFunction<AsyncResult<Message<Object>>, T> function,
    DeliveryOptions deliveryOptions,
    VxmsShared vxmsShared,
    Throwable t,
    Consumer<Throwable> errorMethodHandler,
    RoutingContext context,
    Map<String, String> headers,
    Encoder encoder,
    Consumer<Throwable> errorHandler,
    ThrowableFunction<Throwable, T> onFailureRespond,
    int httpStatusCode,
    int httpErrorCode,
    int retryCount,
    long timeout,
    long delay,
    long circuitBreakerTimeout,
    RecursiveExecutor<T> executor,
    RetryExecutor<T> retry,
    AsyncResult<Message<Object>> event,
    ThrowableSupplier<T> supplier) {
  if (event.succeeded() || (event.failed() && retryCount <= 0)) {
    executor.execute(
        methodId,
        vxmsShared,
        t,
        errorMethodHandler,
        context,
        headers,
        supplier,
        encoder,
        errorHandler,
        onFailureRespond,
        httpStatusCode,
        httpErrorCode,
        retryCount,
        timeout,
        delay,
        circuitBreakerTimeout);
  } else if (event.failed() && retryCount > 0) {
    // retry operation
    final Throwable cause = event.cause();
    retryOperation(
        methodId,
        id,
        message,
        function,
        deliveryOptions,
        vxmsShared,
        cause,
        errorMethodHandler,
        context,
        headers,
        encoder,
        errorHandler,
        onFailureRespond,
        httpStatusCode,
        httpErrorCode,
        retryCount,
        timeout,
        delay,
        circuitBreakerTimeout,
        retry);
  }
}
 

private void handleConnectionRequestResult(final MqttEndpoint endpoint,
        final Span currentSpan,
        final AsyncResult<Device> authenticationAttempt) {

    if (authenticationAttempt.succeeded()) {

        final Device authenticatedDevice = authenticationAttempt.result();
        TracingHelper.TAG_AUTHENTICATED.set(currentSpan, authenticatedDevice != null);

        sendConnectedEvent(endpoint.clientIdentifier(), authenticatedDevice)
                .onComplete(sendAttempt -> {
                    if (sendAttempt.succeeded()) {
                        // we NEVER maintain session state
                        endpoint.accept(false);
                        if (authenticatedDevice != null) {
                            TracingHelper.setDeviceTags(
                                    currentSpan,
                                    authenticationAttempt.result().getTenantId(),
                                    authenticationAttempt.result().getDeviceId());
                        }
                        currentSpan.log("connection accepted");
                    } else {
                        log.warn(
                                "connection request from client [clientId: {}] rejected due to connection event "
                                        + "failure: {}",
                                endpoint.clientIdentifier(),
                                MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE,
                                sendAttempt.cause());
                        endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
                        TracingHelper.logError(currentSpan, sendAttempt.cause());
                    }
                });

    } else {

        final Throwable t = authenticationAttempt.cause();
        TracingHelper.TAG_AUTHENTICATED.set(currentSpan, false);
        log.debug("connection request from client [clientId: {}] rejected due to {} ",
                endpoint.clientIdentifier(), t.getMessage());

        final MqttConnectReturnCode code = getConnectReturnCode(t);
        endpoint.reject(code);
        currentSpan.log("connection rejected");
        TracingHelper.logError(currentSpan, authenticationAttempt.cause());
    }
    currentSpan.finish();
}