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

/**
 * Delete by id.
 * @param connection where to run, can be within a transaction
 * @param table table name without schema
 * @param id primary key value of the record to delete
 * @param replyHandler
 */
public void delete(AsyncResult<SQLConnection> connection, String table, String id,
    Handler<AsyncResult<RowSet<Row>>> replyHandler) {

  try {
    if (connection.failed()) {
      replyHandler.handle(Future.failedFuture(connection.cause()));
      return;
    }
    connection.result().conn.preparedQuery(
        "DELETE FROM " + schemaName + DOT + table + WHERE + ID_FIELD + "=$1")
        .execute(Tuple.of(UUID.fromString(id)), replyHandler);
  } catch (Exception e) {
    replyHandler.handle(Future.failedFuture(e));
  }
}
 

private void doDelete(AsyncResult<SQLConnection> connection, String table, String where,
    Handler<AsyncResult<RowSet<Row>>> replyHandler) {
  try {
    long start = System.nanoTime();
    String sql = DELETE + FROM + schemaName + DOT + table + " " + where;
    log.debug("doDelete query = " + sql);
    if (connection.failed()) {
      replyHandler.handle(Future.failedFuture(connection.cause()));
      return;
    }
    connection.result().conn.query(sql).execute(query -> {
      statsTracker(DELETE_STAT_METHOD, table, start);
      if (query.failed()) {
        log.error(query.cause().getMessage(), query.cause());
        replyHandler.handle(Future.failedFuture(query.cause()));
        return;
      }
      replyHandler.handle(Future.succeededFuture(query.result()));
    });
  } catch (Exception e) {
    log.error(e.getMessage(), e);
    replyHandler.handle(Future.failedFuture(e));
  }
}
 

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

@Override
public void handle(AsyncResult<RowSet<Row>> ar) {
  if (ar.failed()) {
    Handler<Throwable> handler;
    synchronized (RowStreamImpl.this) {
      cursor = null;
      handler = exceptionHandler;
    }
    if (handler != null) {
      handler.handle(ar.cause());
    }
  } else {
    result = ar.result().iterator();
    checkPending();
  }
}
 

/**
 * Rollback a SQL transaction started on the connection. This closes the connection.
 *
 * @see #startTx(Handler)
 * @param trans the connection with an open transaction
 * @param done  success or failure
 */
//@Timer
public void rollbackTx(AsyncResult<SQLConnection> trans, Handler<AsyncResult<Void>> done) {
  try {
    if (trans.failed()) {
      done.handle(Future.failedFuture(trans.cause()));
      return;
    }
    trans.result().tx.rollback(res -> finalizeTx(res, trans.result().conn, done));
  } catch (Exception e) {
    done.handle(Future.failedFuture(e));
  }
}
 

/**
 * Create a parameterized/prepared INSERT, UPDATE or DELETE statement and
 * run it with a list of sets of parameters.
 *
 * <p>Example:
 * <pre>
 *  postgresClient.startTx(beginTx -> {
 *        try {
 *          postgresClient.execute(beginTx, sql, params, reply -> {...
 * </pre>
 * @param conn - connection - see {@link #startTx(Handler)}
 * @param sql - the sql to run
 * @param params - there is one list entry for each sql invocation containing the parameters for the placeholders.
 * @param replyHandler - reply handler with one UpdateResult for each list entry of params.
 */
public void execute(AsyncResult<SQLConnection> conn, String sql, List<Tuple> params,
                    Handler<AsyncResult<List<RowSet<Row>>>> replyHandler) {
  try {
    if (conn.failed()) {
      replyHandler.handle(Future.failedFuture(conn.cause()));
      return;
    }
    PgConnection sqlConnection = conn.result().conn;
    List<RowSet<Row>> results = new ArrayList<>(params.size());
    Iterator<Tuple> iterator = params.iterator();
    Runnable task = new Runnable() {
      @Override
      public void run() {
        if (! iterator.hasNext()) {
          replyHandler.handle(Future.succeededFuture(results));
          return;
        }
        Tuple params1 = iterator.next();
        sqlConnection.preparedQuery(sql).execute(params1, query -> {
          if (query.failed()) {
            replyHandler.handle(Future.failedFuture(query.cause()));
            return;
          }
          results.add(query.result());
          this.run();
        });
      }
    };
    task.run();
  } catch (Exception e) {
    log.error(e.getMessage(), e);
    replyHandler.handle(Future.failedFuture(e));
  }
}
 

static <T> AsyncResult<Void> withoutResult(AsyncResult<T> res) {
  if (res.failed()) {
    return Future.failedFuture(res.cause());
  } else {
    return Future.succeededFuture();
  }
}
 

private void expectSuccess(AsyncResult ar) {
  if (ar.failed()) {
    ar.cause().printStackTrace();
    fail("Failure unexpected: " + ar.cause().getMessage());
  }
  assertThat(ar.succeeded()).isTrue();
}
 

private void subscribeHandler(AsyncResult<Void> subscribeResult) {

        if (subscribeResult.failed()) {
            sendAmqpError(AmqpBridge.AMQP_ERROR_KAFKA_SUBSCRIBE,
                    "Error subscribing to topic " + this.topicSubscriptions,
                    subscribeResult);
        }
    }
 

private void handleResult(AsyncResult<TcfResponse<Integer>> asyncResult, RoutingContext context,
                          UidsCookie uidsCookie, String bidder) {
    if (asyncResult.failed()) {
        respondWithError(context, bidder, asyncResult.cause());
    } else {
        // allow cookie only if user is not in GDPR scope or vendor passed GDPR check
        final TcfResponse<Integer> tcfResponse = asyncResult.result();

        final boolean notInGdprScope = BooleanUtils.isFalse(tcfResponse.getUserInGdprScope());

        final Map<Integer, PrivacyEnforcementAction> vendorIdToAction = tcfResponse.getActions();
        final PrivacyEnforcementAction privacyEnforcementAction = vendorIdToAction != null
                ? vendorIdToAction.get(gdprHostVendorId)
                : null;
        final boolean blockPixelSync = privacyEnforcementAction == null
                || privacyEnforcementAction.isBlockPixelSync();

        final boolean allowedCookie = notInGdprScope || !blockPixelSync;

        if (allowedCookie) {
            respondWithCookie(context, bidder, uidsCookie);
        } else {
            respondWithoutCookie(context, HttpResponseStatus.OK.code(),
                    "The gdpr_consent param prevents cookies from being saved", bidder);
        }
    }
}
 

private void handleServiceRespond(AsyncResult<?> processResult, Promise<Boolean> promise) {
    if (processResult.failed()) {
        final Throwable cause = processResult.cause();
        cleanUp(saveFilePath).setHandler(removalResult -> handleCorruptedFileRemoval(removalResult, promise,
                cause));
    } else {
        promise.complete(false);
        logger.info("Existing file {0} was successfully reused for service", saveFilePath);
    }
}
 

private void partitionHandler(AsyncResult<Optional<PartitionInfo>> partitionResult) {

        if (partitionResult.failed()) {
            sendAmqpError(AmqpBridge.AMQP_ERROR_KAFKA_SUBSCRIBE,
                    "Error getting partition info for topic " + this.topicSubscriptions,
                    partitionResult);
        } else {

            Optional<PartitionInfo> requestedPartitionInfo = partitionResult.result();
            if (!requestedPartitionInfo.isPresent()) {
                sendAmqpError(AmqpBridge.newError(AmqpBridge.AMQP_ERROR_PARTITION_NOT_EXISTS,
                        "Specified partition doesn't exist"));
            }
        }
    }
 

public void update(AsyncResult<SQLConnection> conn, String table, Object entity, String jsonbField,
                   String whereClause, boolean returnUpdatedIds, Handler<AsyncResult<RowSet<Row>>> replyHandler) {
  if (conn.failed()) {
    replyHandler.handle(Future.failedFuture(conn.cause()));
    return;
  }
  long start = System.nanoTime();
  StringBuilder sb = new StringBuilder();
  sb.append(whereClause);
  StringBuilder returning = new StringBuilder();
  if (returnUpdatedIds) {
    returning.append(RETURNING_ID);
  }
  try {
    String q = UPDATE + schemaName + DOT + table + SET + jsonbField + " = $1::jsonb " + whereClause
        + SPACE + returning;
    log.debug("update query = " + q);
    conn.result().conn.preparedQuery(q).execute(Tuple.of(pojo2JsonObject(entity)), query -> {
      if (query.failed()) {
        log.error(query.cause().getMessage(), query.cause());
      }
      statsTracker(UPDATE_STAT_METHOD, table, start);
      replyHandler.handle(query);
    });
  } catch (Exception e) {
    log.error(e.getMessage(), e);
    replyHandler.handle(Future.failedFuture(e));
  }
}
 

private void handleDownload(AsyncResult<Void> downloadResult, Promise<Void> promise) {
    if (downloadResult.failed()) {
        retryDownload(promise, retryInterval, retryCount);
    } else {
        promise.complete();
    }
}
 

/**
 * Ends a SQL transaction (commit) started on the connection. This closes the connection.
 *
 * @see #startTx(Handler)
 * @param trans  the connection with an open transaction
 * @param done  success or failure
 */
//@Timer
public void endTx(AsyncResult<SQLConnection> trans, Handler<AsyncResult<Void>> done) {
  try {
    if (trans.failed()) {
      done.handle(Future.failedFuture(trans.cause()));
      return;
    }
    trans.result().tx.commit(res -> finalizeTx(res, trans.result().conn, done));
  } catch (Exception e) {
    done.handle(Future.failedFuture(e));
  }
}
 

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

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 static <T> void assertSuccess(TestContext context, AsyncResult<T> result) {
  if (result.failed()) {
    context.fail(result.cause());
  }
}
 

private static <T> void statelessExecution(
    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.failed() || (event.failed() && retryCount <= 0)) {
    executor.execute(
        methodId,
        vxmsShared,
        event.cause(),
        errorMethodHandler,
        requestMessage,
        objectConsumer,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        circuitBreakerTimeout);
  } else if (event.failed() && retryCount > 0) {
    retryFunction(
        methodId, targetId,
        message,
        objectFunction,
        requestDeliveryOptions,
        vxmsShared,
        event.cause(),
        errorMethodHandler,
        requestMessage,
        encoder,
        errorHandler,
        onFailureRespond,
        responseDeliveryOptions,
        retryCount,
        timeout,
        circuitBreakerTimeout,
        retry);
  }
}
 

private void handleConnectionAttemptResult(
        final AsyncResult<ProtonConnection> conAttempt,
        final String containerId,
        final Long connectionTimeoutTimerId,
        final AtomicBoolean connectionTimeoutReached,
        final ProtonClientOptions clientOptions,
        final Handler<AsyncResult<ProtonConnection>> closeHandler,
        final Handler<ProtonConnection> disconnectHandler,
        final Handler<AsyncResult<ProtonConnection>> connectionResultHandler) {

    if (connectionTimeoutReached.get()) {
        handleTimedOutConnectionAttemptResult(conAttempt, clientOptions);
        return;
    }

    if (conAttempt.failed()) {
        if (connectionTimeoutTimerId != null) {
            vertx.cancelTimer(connectionTimeoutTimerId);
        }
        failConnectionAttempt(clientOptions, connectionResultHandler, conAttempt.cause());
    } else {

        // at this point the SASL exchange has completed successfully
        logger.debug("connected to AMQP 1.0 container [{}://{}:{}, role: {}], opening connection ...",
                clientOptions.isSsl() ? "amqps" : "amqp",
                config.getHost(),
                config.getPort(),
                config.getServerRole());
        final ProtonConnection downstreamConnection = conAttempt.result();
        downstreamConnection
                .setContainer(containerId)
                .setHostname(config.getAmqpHostname())
                .openHandler(openCon -> {
                    if (connectionTimeoutTimerId != null) {
                        vertx.cancelTimer(connectionTimeoutTimerId);
                    }
                    if (connectionTimeoutReached.get()) {
                        handleTimedOutOpenHandlerResult(openCon, downstreamConnection, clientOptions);
                    } else {
                        if (openCon.succeeded()) {
                            logger.debug("connection to container [{}] at [{}://{}:{}, role: {}] open",
                                    downstreamConnection.getRemoteContainer(),
                                    clientOptions.isSsl() ? "amqps" : "amqp",
                                    config.getHost(),
                                    config.getPort(),
                                    config.getServerRole());
                            downstreamConnection.disconnectHandler(disconnectHandler);
                            downstreamConnection.closeHandler(closeHandler);
                            connectionResultHandler.handle(Future.succeededFuture(downstreamConnection));
                        } else {
                            final ErrorCondition error = downstreamConnection.getRemoteCondition();
                            if (error == null) {
                                logger.warn("can't open connection to container [{}] at [{}://{}:{}, role: {}]",
                                        downstreamConnection.getRemoteContainer(),
                                        clientOptions.isSsl() ? "amqps" : "amqp",
                                        config.getHost(),
                                        config.getPort(),
                                        config.getServerRole(),
                                        openCon.cause());
                            } else {
                                logger.warn("can't open connection to container [{}] at [{}://{}:{}, role: {}]: {} -{}",
                                        downstreamConnection.getRemoteContainer(),
                                        clientOptions.isSsl() ? "amqps" : "amqp",
                                        config.getHost(),
                                        config.getPort(),
                                        config.getServerRole(),
                                        error.getCondition(),
                                        error.getDescription());
                            }
                            connectionResultHandler.handle(Future.failedFuture(openCon.cause()));
                        }
                    }
                }).disconnectHandler(disconnectedCon -> {
                    if (connectionTimeoutTimerId != null) {
                        vertx.cancelTimer(connectionTimeoutTimerId);
                    }
                    if (connectionTimeoutReached.get()) {
                        logger.warn("ignoring error - connection attempt already timed out: can't open connection to container [{}] at [{}://{}:{}, role: {}]: {}",
                                downstreamConnection.getRemoteContainer(),
                                clientOptions.isSsl() ? "amqps" : "amqp",
                                config.getHost(),
                                config.getPort(),
                                config.getServerRole(),
                                "underlying connection was disconnected while opening AMQP connection");
                    } else {
                        logger.warn("can't open connection to container [{}] at [{}://{}:{}, role: {}]: {}",
                                downstreamConnection.getRemoteContainer(),
                                clientOptions.isSsl() ? "amqps" : "amqp",
                                config.getHost(),
                                config.getPort(),
                                config.getServerRole(),
                                "underlying connection was disconnected while opening AMQP connection");
                        connectionResultHandler.handle(Future
                                .failedFuture("underlying connection was disconnected while opening AMQP connection"));
                    }
                }).open();
    }
}