io.vertx.core.AsyncResult#result ( )源码实例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());
        }
    }
}
 

private void handleTimedOutConnectionAttemptResult(final AsyncResult<ProtonConnection> conAttempt, final ProtonClientOptions clientOptions) {
    if (conAttempt.succeeded()) {
        logger.debug("ignoring successful connection attempt to AMQP 1.0 container [{}://{}:{}, role: {}]: attempt already timed out",
                clientOptions.isSsl() ? "amqps" : "amqp",
                config.getHost(),
                config.getPort(),
                config.getServerRole());
        final ProtonConnection downstreamConnection = conAttempt.result();
        downstreamConnection.disconnect();
    } else {
        logger.debug("ignoring failed connection attempt to AMQP 1.0 container [{}://{}:{}, role: {}]: attempt already timed out",
                clientOptions.isSsl() ? "amqps" : "amqp",
                config.getHost(),
                config.getPort(),
                config.getServerRole(),
                conAttempt.cause());
    }
}
 

/**
 * The returned handler first closes the SQLConnection and then passes on the AsyncResult to handler.
 *
 * <p>The returned Handler ignores (but logs) any failure when opening the connection (conn) or
 * closing the connection and always passes on the AsyncResult<T>. This is in contrast to
 * io.vertx.ext.sql.HandlerUtil.closeAndHandleResult where the connection
 * closing failure suppresses any result or failure of the AsyncResult<T> input.
 *
 * @param conn  the SQLConnection to close
 * @param handler  where to pass on the input AsyncResult
 * @return the Handler
 */
 <T> Handler<AsyncResult<T>> closeAndHandleResult(
    AsyncResult<SQLConnection> conn, Handler<AsyncResult<T>> handler) {

  return ar -> {
    if (conn.failed()) {
      log.error("Opening SQLConnection failed: " + conn.cause().getMessage(), conn.cause());
      handler.handle(ar);
      return;
    }
    SQLConnection sqlConnection = conn.result();
    if (sqlConnection.conn != null) {
      sqlConnection.conn.close();
    }
    cancelConnectionTimeoutTimer(sqlConnection);
    handler.handle(ar);
  };
}
 

private void handleSync(RemoteFileProcessor remoteFileProcessor, AsyncResult<Boolean> syncResult) {
    if (syncResult.succeeded()) {
        if (syncResult.result()) {
            logger.info("Sync service for {0}", saveFilePath);
            remoteFileProcessor.setDataPath(saveFilePath)
                    .setHandler(this::logFileProcessStatus);
        } else {
            logger.info("Sync is not required for {0}", saveFilePath);
        }
    } else {
        logger.error("Cant sync file from {0}", syncResult.cause(), downloadUrl);
    }

    // setup new update regardless of the result
    if (updatePeriod > 0) {
        vertx.setTimer(updatePeriod, idUpdateNew -> configureAutoUpdates(remoteFileProcessor));
    }
}
 

protected void onFileOpened(File file, AsyncResult<AsyncFile> ar, CompletableFuture<File> future) {
  if (ar.failed()) {
    future.completeExceptionally(ar.cause());
    return;
  }

  AsyncFile asyncFile = ar.result();
  CompletableFuture<Void> saveFuture = saveToWriteStream(asyncFile);
  saveFuture.whenComplete((v, saveException) -> {
    asyncFile.close(closeAr -> {
      if (closeAr.failed()) {
        LOGGER.error("Failed to close file {}.", file);
      }

      // whatever close success or failed
      // will not affect to result
      // result just only related to write
      if (saveException == null) {
        future.complete(file);
        return;
      }

      future.completeExceptionally(saveException);
    });
  });
}
 

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

protected <T> void responseHandler(AsyncResult<HttpResponse<T>> ar, CompletableFuture<T> promise, Predicate<Integer> expectedCodePredicate,
        String expectedCodeOrCodes, String warnMessage, boolean throwException) {
    try {
        if (ar.succeeded()) {
            HttpResponse<T> response = ar.result();
            T body = response.body();
            if (expectedCodePredicate.negate().test(response.statusCode())) {
                log.error("expected-code: {}, response-code: {}, body: {}, op: {}", expectedCodeOrCodes, response.statusCode(), response.body(), warnMessage);
                promise.completeExceptionally(new RuntimeException("Status " + response.statusCode() + " body: " + (body != null ? body.toString() : null)));
            } else if (response.statusCode() < HTTP_OK || response.statusCode() >= HttpURLConnection.HTTP_MULT_CHOICE) {
                if (throwException) {
                    promise.completeExceptionally(new RuntimeException(body == null ? "null" : body.toString()));
                } else {
                    promise.complete(ar.result().body());
                }
            } else {
                promise.complete(ar.result().body());
            }
        } else {
            log.warn("Request failed: {}", warnMessage, ar.cause());
            promise.completeExceptionally(ar.cause());
        }
    } catch (io.vertx.core.json.DecodeException decEx) {
        if (ar.result().bodyAsString().toLowerCase().contains("application is not available")) {
            log.warn("'{}' is not available.", apiClientName(), ar.cause());
            throw new IllegalStateException(String.format("'%s' is not available.", apiClientName()));
        } else {
            log.warn("Unexpected object received", ar.cause());
            throw new IllegalStateException("JsonObject expected, but following object was received: " + ar.result().bodyAsString());
        }
    }
}
 

private void completeFromAsyncResult(AsyncResult<T> ar) {
    if (ar.succeeded()) {
        super.complete(ar.result());
    } else {
        super.completeExceptionally(ar.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;
}
 

/**
 * low-level getter based on CQLWrapper
 * @param <T>
 * @param conn
 * @param table
 * @param clazz
 * @param fieldName
 * @param wrapper
 * @param returnCount
 * @param returnIdField
 * @param facets
 * @param distinctOn
 * @param replyHandler
 */
private <T> void doGet(
  AsyncResult<SQLConnection> conn, String table, Class<T> clazz,
  String fieldName, CQLWrapper wrapper, boolean returnCount, boolean returnIdField,
  List<FacetField> facets, String distinctOn, Handler<AsyncResult<Results<T>>> replyHandler
) {

  if (conn.failed()) {
    log.error(conn.cause().getMessage(), conn.cause());
    replyHandler.handle(Future.failedFuture(conn.cause()));
    return;
  }
  PgConnection connection = conn.result().conn;
  try {
    QueryHelper queryHelper = buildQueryHelper(table, fieldName, wrapper, returnIdField, facets, distinctOn);
    if (returnCount) {
      processQueryWithCount(connection, queryHelper, GET_STAT_METHOD,
        totaledResults -> processResults(totaledResults.set, totaledResults.total, queryHelper.offset, queryHelper.limit, clazz), replyHandler);
    } else {
      processQuery(connection, queryHelper, null, GET_STAT_METHOD,
        totaledResults -> processResults(totaledResults.set, totaledResults.total, queryHelper.offset, queryHelper.limit, clazz), replyHandler);
    }
  } catch (Exception e) {
    log.error(e.getMessage(), e);
    replyHandler.handle(Future.failedFuture(e));
  }
}
 

private void handleFileExistsWithDelete(String filePath, AsyncResult<Boolean> existResult, Promise<Void> promise) {
    if (existResult.succeeded()) {
        if (existResult.result()) {
            fileSystem.delete(filePath, promise);
        } else {
            promise.complete();
        }
    } else {
        promise.fail(new PreBidException(String.format("Cant check if file exists %s", filePath)));
    }
}
 

private void handleFileOpenWithDownload(AsyncResult<AsyncFile> openResult, Promise<Void> promise) {
    if (openResult.succeeded()) {
        final AsyncFile asyncFile = openResult.result();
        try {
            httpClient.getAbs(downloadUrl, response -> pumpFileFromRequest(response, asyncFile, promise)).end();
        } catch (Exception e) {
            promise.fail(e);
        }
    } else {
        promise.fail(openResult.cause());
    }
}
 

private void onGetResponse(AsyncResult<String> ar) {
  if (ar.succeeded()) {
    if (ar.result() == null) {
      createCache();
      return;
    }

    future.complete(ar.result());
    return;
  }

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

private void saveBatchInternal(AsyncResult<SQLConnection> sqlConnection, boolean upsert, String table,
                                List<Tuple> batch, Handler<AsyncResult<RowSet<Row>>> replyHandler) {

  try {
    long start = System.nanoTime();
    log.info("starting: saveBatch size=" + batch.size());
    String sql = INSERT_CLAUSE + schemaName + DOT + table + " (id, jsonb) VALUES ($1, $2)"
        + (upsert ? " ON CONFLICT (id) DO UPDATE SET jsonb = EXCLUDED.jsonb" : "")
        + RETURNING_ID;
    if (sqlConnection.failed()) {
      replyHandler.handle(Future.failedFuture(sqlConnection.cause()));
      return;
    }
    PgConnection connection = sqlConnection.result().conn;

    connection.preparedQuery(sql).executeBatch(batch, queryRes -> {
      if (queryRes.failed()) {
        log.error("saveBatch size=" + batch.size()
                + SPACE
                + queryRes.cause().getMessage(),
            queryRes.cause());
        statsTracker("saveBatchFailed", table, start);
        replyHandler.handle(Future.failedFuture(queryRes.cause()));
        return;
      }
      statsTracker("saveBatch", table, start);
      if (queryRes.result() != null) {
        replyHandler.handle(Future.succeededFuture(queryRes.result()));
      } else {
        replyHandler.handle(Future.succeededFuture(new LocalRowSet(0)));
      }
    });
  } catch (Exception e) {
    log.error(e.getMessage(), e);
    replyHandler.handle(Future.failedFuture(e));
  }
}
 

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();
  }
}
 

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 static <T> void handleStatefulError(
    String _methodId,
    ThrowableSupplier<T> _supplier,
    Promise<ExecutionResult<T>> _blockingHandler,
    Consumer<Throwable> _errorHandler,
    ThrowableFunction<Throwable, T> _onFailureRespond,
    Consumer<Throwable> _errorMethodHandler,
    VxmsShared vxmsShared,
    Throwable _t,
    int _retry,
    long _timeout,
    long _circuitBreakerTimeout,
    long _delay,
    Throwable e,
    Lock lck,
    Counter counter,
    AsyncResult<Long> valHandler) {
  //////////////////////////////////////////
  long count = valHandler.result();
  if (count <= DEFAULT_VALUE) {
    setCircuitBreakerReleaseTimer(vxmsShared, _retry, _circuitBreakerTimeout, counter);
    openCircuitBreakerAndHandleError(
        _blockingHandler,
        _errorHandler,
        _onFailureRespond,
        _errorMethodHandler,
        vxmsShared,
        e,
        lck,
        counter);
  } else {
    lck.release();
    org.jacpfx.vxms.event.response.basic.ResponseExecution.handleError(_errorHandler, e);
    handleDelay(_delay);
    createResponseBlocking(
        _methodId,
        _supplier,
        _blockingHandler,
        _errorHandler,
        _onFailureRespond,
        _errorMethodHandler,
        vxmsShared,
        _t,
        _retry,
        _timeout,
        _circuitBreakerTimeout,
        _delay);
  }
  ////////////////////////////////////////
}
 

private static <T, V> void handleStatefulError(
        String _methodId,
        ThrowableFunction<T, V> step,
        T value,
        Promise<ExecutionResult<V>> _blockingHandler,
        Consumer<Throwable> _errorHandler,
        ThrowableFunction<Throwable, V> _onFailureRespond,
        Consumer<Throwable> _errorMethodHandler,
        VxmsShared vxmsShared,
        Throwable _t,
        int _retry,
        long _timeout,
        long _circuitBreakerTimeout,
        long _delay,
        Throwable e,
        Lock lck,
        Counter counter,
        AsyncResult<Long> valHandler) {
    //////////////////////////////////////////
    long count = valHandler.result();
    if (count <= DEFAULT_LONG_VALUE) {
        setCircuitBreakerReleaseTimer(vxmsShared, _retry, _circuitBreakerTimeout, counter);
        openCircuitBreakerAndHandleError(
                _blockingHandler,
                _errorHandler,
                _onFailureRespond,
                _errorMethodHandler,
                vxmsShared,
                e,
                lck,
                counter);
    } else {
        lck.release();
        org.jacpfx.vxms.rest.base.response.basic.ResponseExecution.handleError(_errorHandler, e);
        handleDelay(_delay);
        executeRetryAndCatchAsync(
                _methodId,
                step,
                value,
                _blockingHandler,
                _errorHandler,
                _onFailureRespond,
                _errorMethodHandler,
                vxmsShared,
                _t,
                _retry,
                _timeout,
                _circuitBreakerTimeout,
                _delay);
    }
    ////////////////////////////////////////
}
 

private static <T> void handleStatefulError(
    String _methodId,
    ThrowableSupplier<T> _supplier,
    Promise<ExecutionResult<T>> _blockingHandler,
    Consumer<Throwable> _errorHandler,
    ThrowableFunction<Throwable, T> _onFailureRespond,
    Consumer<Throwable> _errorMethodHandler,
    VxmsShared vxmsShared,
    Throwable _t,
    int _retry,
    long _timeout,
    long _circuitBreakerTimeout,
    long _delay,
    Throwable e,
    Lock lck,
    Counter counter,
    AsyncResult<Long> valHandler) {
  //////////////////////////////////////////
  long count = valHandler.result();
  if (count <= DEFAULT_LONG_VALUE) {
    setCircuitBreakerReleaseTimer(vxmsShared, _retry, _circuitBreakerTimeout, counter);
    openCircuitBreakerAndHandleError(
        _blockingHandler,
        _errorHandler,
        _onFailureRespond,
        _errorMethodHandler,
        vxmsShared,
        e,
        lck,
        counter);
  } else {
    lck.release();
    org.jacpfx.vxms.rest.base.response.basic.ResponseExecution.handleError(_errorHandler, e);
    handleDelay(_delay);
    executeRetryAndCatchAsync(
        _methodId,
        _supplier,
        _blockingHandler,
        _errorHandler,
        _onFailureRespond,
        _errorMethodHandler,
        vxmsShared,
        _t,
        _retry,
        _timeout,
        _circuitBreakerTimeout,
        _delay);
  }
  ////////////////////////////////////////
}