com.fasterxml.jackson.databind.exc.InvalidTypeIdException#io.vertx.core.AsyncResult源码实例Demo

/**
 * Mocks the api client telling that the next time there is remote call, an error should be
 * returned.
 *
 * @param statusCode the status code of the response (404 for example)
 * @param errorResponse the raw response, it may or may not be a json string.
 */
protected void mockErrorCodeRawResponse(int statusCode, String errorResponse) {
    String reasonPhrase = HttpStatus.valueOf(statusCode).getReasonPhrase();
    VertxHttpHeaders headers = new VertxHttpHeaders();
    ApiException exception = new ApiException(reasonPhrase, statusCode, headers,
        errorResponse);

    Mockito.doAnswer((Answer<Void>) invocationOnMock -> {

        Handler<AsyncResult<Object>> resultHandler = (Handler<AsyncResult<Object>>) invocationOnMock
            .getArguments()[invocationOnMock.getArguments().length - 1];
        resultHandler.handle(Future.failedFuture(exception));

        return null;
    }).when(apiClientMock)
        .invokeAPI(Mockito.anyString(), Mockito.anyString(), Mockito.anyList(), Mockito.any(),
            Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(),
            Mockito.any(), Mockito.any(), Mockito.any());
}
 

private void getContainerLog(Handler<AsyncResult<Void>> future) {
  final String url = "/containers/" + containerId
      + "/logs?stderr=1&stdout=1&follow=1";
  HttpClientRequest req = request(HttpMethod.GET, url, res -> {
    if (res.statusCode() == 200) {
      // stream OK. Continue other work but keep fetching!
      // remove 8 bytes of binary data and final newline
      res.handler(this::logHandler);
      tcpPortWaiting.waitReady(null, future);
    } else {
      String m = "getContainerLog HTTP error "
          + res.statusCode();
      logger.error(m);
      future.handle(Future.failedFuture(m));
    }
  });
  req.exceptionHandler(d -> future.handle(Future.failedFuture(d.getCause())));
  req.end();
}
 

@Override
public void start(Future<Void> startFuture) throws Exception {

    server = vertx.createHttpServer(options);

    server.requestHandler(request -> {
        VertxHttpExchange delegate = new VertxHttpExchange(request, allocator, blockingExecutor, null, null);
        rootHandler.handle(delegate);
    });

    server.listen(port, host, new Handler<AsyncResult<HttpServer>>() {
        @Override
        public void handle(AsyncResult<HttpServer> event) {
            if (event.failed()) {
                startFuture.fail(event.cause());
            } else {
                startFuture.complete();
            }
        }
    });
}
 

public ProductService retrieveProductPrice(String productId, Handler<AsyncResult<JsonObject>> resultHandler) {
  if (closed) {
    resultHandler.handle(Future.failedFuture(new IllegalStateException("Proxy is closed")));
    return this;
  }
  JsonObject _json = new JsonObject();
  _json.put("productId", productId);
  DeliveryOptions _deliveryOptions = (_options != null) ? new DeliveryOptions(_options) : new DeliveryOptions();
  _deliveryOptions.addHeader("action", "retrieveProductPrice");
  _vertx.eventBus().<JsonObject>send(_address, _json, _deliveryOptions, res -> {
    if (res.failed()) {
      resultHandler.handle(Future.failedFuture(res.cause()));
    } else {
      resultHandler.handle(Future.succeededFuture(res.result().body()));
    }
  });
  return this;
}
 

/**
 * Sets a handler on a link that is invoked when an AMQP <em>detach</em> frame
 * with its <em>close</em> property set to {@code false} is received from the peer.
 * <p>
 * The resources maintained for the link will be freed up after the given handler has
 * been invoked.
 *
 * @param <T> The type of link.
 * @param link The link to set the handler on.
 * @param handler The handler to invoke.
 * @return The wrapper that has been created around the given handler.
 * @throws NullPointerException if link or handler are {@code null}.
 */
public static <T extends ProtonLink<T>> Handler<AsyncResult<T>> setDetachHandler(
        final ProtonLink<T> link,
        final Handler<AsyncResult<T>> handler) {

    Objects.requireNonNull(link);
    Objects.requireNonNull(handler);

    final Handler<AsyncResult<T>> wrappedHandler = remoteDetach -> {
        try {
            handler.handle(remoteDetach);
        } catch (final Exception ex) {
            LOG.warn("error running detachHandler", ex);
        }
        link.free();
    };
    link.detachHandler(wrappedHandler);
    return wrappedHandler;
}
 

@Test
public void testAssertAsyncFailureHandlerSucceeded() throws Exception {
  BlockingQueue<Handler<AsyncResult<String>>> handlerQueue = new ArrayBlockingQueue<>(1);
  BlockingQueue<Throwable> resultQueue = new ArrayBlockingQueue<>(1);
  TestSuite suite = TestSuite.create("my_suite").test("my_test", context -> {
    handlerQueue.add(context.<String>asyncAssertFailure(resultQueue::add));
  });
  TestReporter reporter = new TestReporter();
  run(suite, reporter);
  Handler<AsyncResult<String>> handler = handlerQueue.poll(2, TimeUnit.SECONDS);
  Throwable expected = new Throwable();
  handler.handle(Future.failedFuture(expected));
  Throwable cause = resultQueue.poll(2, TimeUnit.SECONDS);
  assertSame(expected, cause);
  reporter.await();
  assertEquals(0, reporter.exceptions.size());
  assertEquals(1, reporter.results.size());
  assertEquals("my_test", reporter.results.get(0).name());
  assertFalse(reporter.results.get(0).failed());
}
 

private void relocateAsync(Marker marker, byte[] bytes, Handler<AsyncResult<byte[]>> callback) {
  logger.info(marker, "Relocating event. Total event byte size: {}", bytes.length);
  Service.vertx.executeBlocking(future -> {
    try {
      future.complete(relocationClient.relocate(marker.getName(), bytes));
    }
    catch (Exception e) {
      logger.error("An error occurred when trying to relocate the event.", e);
      future.fail(new HttpException(BAD_GATEWAY, "Unable to relocate event.", e));
    }
  }, ar -> {
    if (ar.failed())
      callback.handle(Future.failedFuture(ar.cause()));
    else
      callback.handle(Future.succeededFuture((byte[]) ar.result()));
  });
}
 

private void startServer(List<CommandResolver> resolvers, Handler<AsyncResult<Void>> startHandler) {
  TelnetTermOptions telnetOptions = options.getTelnetOptions();
  SSHTermOptions sshOptions = options.getSSHOptions();
  HttpTermOptions webOptions = options.getHttpOptions();
  if (telnetOptions != null) {
    server.registerTermServer(new TelnetTermServer(vertx, telnetOptions));
  }
  if (sshOptions != null) {
    server.registerTermServer(new SSHServer(vertx, sshOptions));
  }
  if (webOptions != null) {
    server.registerTermServer(new HttpTermServer(vertx, webOptions));
  }
  resolvers.forEach(server::registerCommandResolver);
  server.listen(startHandler);
}
 

private <T> Handler<AsyncResult<T>> createHandler(Message msg) {
  return res -> {
    if (res.failed()) {
      if (res.cause() instanceof ServiceException) {
        msg.reply(res.cause());
      } else {
        msg.reply(new ServiceException(-1, res.cause().getMessage()));
      }
    } else {
      if (res.result() != null  && res.result().getClass().isEnum()) {
        msg.reply(((Enum) res.result()).name());
      } else {
        msg.reply(res.result());
      }
    }
  };
}
 

/**
 * only one impl allowed
 * @param resultHandler
 * @throws Exception
 */
private void runShutdownHook(Handler<AsyncResult<Void>> resultHandler) throws Exception {
  try {
    ArrayList<Class<?>> aClass = InterfaceToImpl.convert2Impl(RTFConsts.PACKAGE_OF_IMPLEMENTATIONS, RTFConsts.PACKAGE_OF_HOOK_INTERFACES + ".ShutdownAPI", false);
    for (int i = 0; i < aClass.size(); i++) {
      Class<?>[] paramArray = new Class[] { Vertx.class, Context.class, Handler.class };
      Method method = aClass.get(i).getMethod("shutdown", paramArray);
      method.invoke(aClass.get(i).newInstance(), vertx, context, resultHandler);
      LogUtil.formatLogMessage(getClass().getName(), "runShutdownHook",
        "shutdown hook called with implemented class " + "named " + aClass.get(i).getName());
    }
  } catch (ClassNotFoundException e) {
    // no hook implemented, this is fine, just startup normally then
    LogUtil.formatLogMessage(getClass().getName(), "runShutdownHook", "no shutdown hook implementation found, continuing with shutdown");
    resultHandler.handle(io.vertx.core.Future.succeededFuture());
  }
}
 

@Override
public void getAdminJstack(java.util.Map<String, String>okapiHeaders, Handler<AsyncResult<Response>> asyncResultHandler, Context vertxContext) {

  final StringBuilder dump = new StringBuilder();
  vertxContext.owner().executeBlocking(
    code -> {
      try {
        dump.append("<html><body>");
        final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
        final ThreadInfo[] threadInfos = threadMXBean.getThreadInfo(
          threadMXBean.getAllThreadIds(), 100);
        for (ThreadInfo threadInfo : threadInfos) {
          dump.append(threadInfo.getThreadName());
          final Thread.State state = threadInfo.getThreadState();
          dump.append("</br>   java.lang.Thread.State: ");
          dump.append(state);
          final StackTraceElement[] stackTraceElements = threadInfo.getStackTrace();
          for (final StackTraceElement stackTraceElement : stackTraceElements) {
            dump.append("</br>        at ");
            dump.append(stackTraceElement);
          }
          dump.append("</br></br>");
        }
        dump.append("</body></html>");
        code.complete(dump);
      } catch (Exception e) {
        log.error(e.getMessage(), e);
        asyncResultHandler.handle(io.vertx.core.Future.succeededFuture(GetAdminJstackResponse.respond500WithTextPlain("ERROR"
            + e.getMessage())));
      }
    },
    result -> {
      asyncResultHandler.handle(io.vertx.core.Future.succeededFuture(GetAdminJstackResponse.respond200WithTextHtml(result.result().toString())));
    });
}
 

public MockServer(Vertx vertx, Handler<ProtonConnection> connectionHandler) throws ExecutionException, InterruptedException {
  if(connectionHandler == null) {
    connectionHandler = (connection) -> processConnection(vertx, connection);
  }

  ProtonServerOptions protonServerOptions = new ProtonServerOptions();
  protonServerOptions.setReuseAddress(reuseAddress);
  server = ProtonServer.create(vertx, protonServerOptions);
  server.connectHandler(connectionHandler);
  FutureHandler<ProtonServer, AsyncResult<ProtonServer>> handler = FutureHandler.asyncResult();
  server.listen(bindPort, handler);
  handler.get();
}
 

@Override
public void parseCredentials(RoutingContext context, Handler<AsyncResult<JsonObject>> handler) {
  final List<String> access_token = Query.queryParam(Query.ACCESS_TOKEN, context);
  if (access_token != null && access_token.size() > 0) {
    handler.handle(Future.succeededFuture(new JsonObject().put("jwt", access_token.get(0)).put("options", options)));
    return;
  }
  handler.handle(Future.failedFuture(new HttpStatusException(UNAUTHORIZED.code(), "Missing auth credentials.")));
}
 

@Validate
@Override
public void getAdminLoglevel(java.util.Map<String, String>okapiHeaders, Handler<AsyncResult<Response>> asyncResultHandler, Context vertxContext) {

  try {
    JsonObject loggers = LogUtil.getLogConfiguration();
    OutStream os = new OutStream();
    os.setData(loggers);
    asyncResultHandler.handle(io.vertx.core.Future.succeededFuture(PutAdminLoglevelResponse.respond200WithApplicationJson(os)));
  } catch (Exception e) {
    asyncResultHandler.handle(io.vertx.core.Future.succeededFuture(PutAdminLoglevelResponse.respond500WithTextPlain("ERROR"
        + e.getMessage())));
    log.error(e.getMessage(), e);
  }
}
 

private void checkAccountStatus(io.gravitee.am.model.User user, Handler<AsyncResult<Void>> handler) {
    userManager.get(user.getId())
            .subscribe(
                    user1 -> {
                        // if user is disabled, throw exception
                        if (!user1.isEnabled()) {
                            handler.handle(Future.failedFuture(new AccountDisabledException(user1.getId())));
                            return;
                        }
                        handler.handle(Future.succeededFuture());
                    },
                    error -> handler.handle(Future.failedFuture(error)),
                    () -> handler.handle(Future.succeededFuture()));
}
 

/**
 * streamGet with existing transaction/connection
 * @param <T>
 * @param connResult
 * @param table
 * @param clazz
 * @param fieldName
 * @param wrapper
 * @param returnIdField
 * @param distinctOn
 * @param facets
 * @param replyHandler
 */
@SuppressWarnings({"squid:S00107"})    // Method has >7 parameters
<T> void streamGet(AsyncResult<SQLConnection> connResult,
  String table, Class<T> clazz, String fieldName, CQLWrapper wrapper,
  boolean returnIdField, String distinctOn, List<FacetField> facets,
  Handler<AsyncResult<PostgresClientStreamResult<T>>> replyHandler) {

  if (connResult.failed()) {
    log.error(connResult.cause().getMessage(), connResult.cause());
    replyHandler.handle(Future.failedFuture(connResult.cause()));
    return;
  }
  doStreamGetCount(connResult.result(), table, clazz, fieldName, wrapper, returnIdField,
      distinctOn, facets, replyHandler);
}
 

@Override
public void close(final Handler<AsyncResult<Void>> closeHandler) {

    closeLinks(ok -> {
        if (localCloseHandler != null) {
            localCloseHandler.handle(receiver.getSource().getAddress());
        }
        if (closeHandler != null) {
            closeHandler.handle(Future.succeededFuture());
        }
    });
}
 

@Override
public void basicAck(long deliveryTag, boolean multiple, Handler<AsyncResult<Void>> resultHandler) {
  Future<Void> fut = basicAck(deliveryTag, multiple);
  if (resultHandler != null) {
    fut.onComplete(resultHandler);
  }
}
 

@Override
public void delete(String id, Handler<AsyncResult<Void>> resultHandler) {
    clusteredSessionStore.delete(id, res -> {
        if (res.succeeded()) {
            localMap.remove(id);
            resultHandler.handle(Future.succeededFuture());
        } else {
            resultHandler.handle(Future.failedFuture(res.cause()));
        }
    });
}
 

private void load(Map<io.gravitee.am.model.Factor, FactorProvider> providers, User user, Handler<AsyncResult<List<Factor>>> handler) {
    Observable.fromIterable(providers.entrySet())
            .flatMapSingle(entry -> entry.getValue().enroll(user.getUsername())
                    .map(enrollment -> new Factor(entry.getKey(), enrollment))
            )
            .toList()
            .subscribe(
                    factors -> handler.handle(Future.succeededFuture(factors)),
                    error -> handler.handle(Future.failedFuture(error)));
}
 

@Override
public void sell(int amount, JsonObject quote, Handler<AsyncResult<Portfolio>> resultHandler) {
    if (amount <= 0) {
        resultHandler.handle(Future.failedFuture("Cannot sell " + quote.getString("name") + " - the amount must be " +
            "greater than 0"));
        return;
    }

    double price = amount * quote.getDouble("bid");
    String name = quote.getString("name");
    int current = portfolio.getAmount(name);
    // 1) do we have enough stocks
    if (current >= amount) {
        // Yes, sell it
        int newAmount = current - amount;
        if (newAmount == 0) {
            portfolio.getShares().remove(name);
        } else {
            portfolio.getShares().put(name, newAmount);
        }
        portfolio.setCash(portfolio.getCash() + price);
        sendActionOnTheEventBus("SELL", amount, quote, newAmount);
        resultHandler.handle(Future.succeededFuture(portfolio));
    } else {
        resultHandler.handle(Future.failedFuture("Cannot sell " + amount + " of " + name + " - " + "not enough stocks " +
            "in portfolio"));
    }

}
 

/**
 * Verifies that the attempt to create a sender succeeds when sender never gets credits.
 *
 * @param ctx The vert.x test context.
 */
@Test
public void testCreateSenderThatGetsNoCredits(final VertxTestContext ctx) {
    final ProtonSender sender = mock(ProtonSender.class);
    when(sender.isOpen()).thenReturn(Boolean.TRUE);
    when(con.createSender(anyString())).thenReturn(sender);
    final Target target = new Target();
    target.setAddress("someAddress");
    when(sender.getRemoteTarget()).thenReturn(target);
    when(sender.getCredit()).thenReturn(0);
    // just invoke openHandler with succeeded future
    doAnswer(AdditionalAnswers.answerVoid(
            (final Handler<AsyncResult<ProtonSender>> handler) -> handler.handle(Future.succeededFuture(sender))))
                    .when(sender).openHandler(VertxMockSupport.anyHandler());
    final Handler<String> remoteCloseHook = VertxMockSupport.mockHandler();

    // GIVEN an established connection
    honoConnection.connect()
        .compose(c -> honoConnection.createSender(
            "target", ProtonQoS.AT_LEAST_ONCE, remoteCloseHook))
        .onComplete(ctx.succeeding(s -> {
                ctx.verify(() -> {
                    assertThat(s).isEqualTo(sender);
                    // sendQueueDrainHandler gets unset
                    verify(sender).sendQueueDrainHandler(null);
                });
                ctx.completeNow();
            }));
}
 

private void resolveClient(String clientId, Handler<AsyncResult<Client>> handler) {
    clientSyncService.findByDomainAndClientId(domain, clientId)
            .subscribe(
                    client -> handler.handle(Future.succeededFuture(client)),
                    error -> handler.handle(Future.failedFuture(error)),
                    () -> handler.handle(Future.failedFuture(new ClientNotFoundException(clientId)))
            );
}
 

/**
 * Verifies username/password credentials with a remote authentication server using SASL PLAIN.
 *
 * @param authzid The identity to act as.
 * @param authcid The username.
 * @param password The password.
 * @param authenticationResultHandler The handler to invoke with the authentication result. On successful authentication,
 *                                    the result contains a JWT with the authenticated user's claims.
 */
public void verifyPlain(final String authzid, final String authcid, final String password,
        final Handler<AsyncResult<HonoUser>> authenticationResultHandler) {

    final ProtonClientOptions options = new ProtonClientOptions();
    options.setReconnectAttempts(3).setReconnectInterval(50);
    options.addEnabledSaslMechanism(AuthenticationConstants.MECHANISM_PLAIN);

    final Promise<ProtonConnection> connectAttempt = Promise.promise();
    factory.connect(options, authcid, password, null, null, connectAttempt);

    connectAttempt.future()
    .compose(openCon -> getToken(openCon))
    .onComplete(s -> {
        if (s.succeeded()) {
            authenticationResultHandler.handle(Future.succeededFuture(s.result()));
        } else {
            authenticationResultHandler
            .handle(Future.failedFuture(mapConnectionFailureToServiceInvocationException(s.cause())));
        }
        Optional.ofNullable(connectAttempt.future().result())
        .ifPresent(con -> {
            LOG.debug("closing connection to Authentication service");
            con.close();
        });
    });
}
 

private <T> Handler<AsyncResult<Set<T>>> createSetHandler(Message msg) {
  return res -> {
    if (res.failed()) {
      if (res.cause() instanceof ServiceException) {
        msg.reply(res.cause());
      } else {
        msg.reply(new ServiceException(-1, res.cause().getMessage()));
      }
    } else {
      msg.reply(new JsonArray(new ArrayList<>(res.result())));
    }
  };
}
 

/**
 * create execution chain for event-bus request and reply to rest
 *
 * @param _methodId the method identifier
 * @param _targetId the event-bus target id
 * @param _message the message to send
 * @param _objectFunction the function to process the result message
 * @param _options the event-bus delivery serverOptions
 * @param _vxmsShared the vxmsShared instance, containing the Vertx instance and other shared
 *     objects per instance
 * @param _failure the failure thrown while task execution
 * @param _errorMethodHandler the error-method handler
 * @param _context the vertx routing context
 * @param _encoder the encoder to encode your objects
 * @return the execution chain {@link ExecuteRSObjectResponse}
 */
public static ExecuteRSObjectResponse mapToObjectResponse(
    String _methodId,
    String _targetId,
    Object _message,
    ThrowableFutureBiConsumer<AsyncResult<Message<Object>>, Serializable> _objectFunction,
    DeliveryOptions _options,
    VxmsShared _vxmsShared,
    Throwable _failure,
    Consumer<Throwable> _errorMethodHandler,
    RoutingContext _context,
    Encoder _encoder) {

  return mapToObjectResponse(
      _methodId,
      _targetId,
      _message,
      _objectFunction,
      _options,
      _vxmsShared,
      _failure,
      _errorMethodHandler,
      _context,
      null,
      null,
      _encoder,
      null,
      null,
      0,
      0,
      0,
      0,
      0);
}
 

void getSQLConnection(int queryTimeout, Handler<AsyncResult<SQLConnection>> handler) {
  getConnection(res -> {
    if (res.failed()) {
      handler.handle(Future.failedFuture(res.cause()));
      return;
    }

    PgConnection pgConnection = res.result();

    if (queryTimeout == 0) {
      handler.handle(Future.succeededFuture(new SQLConnection(pgConnection, null, null)));
      return;
    }

    long timerId = vertx.setTimer(queryTimeout, id -> pgConnection.cancelRequest(ar -> {
      if (ar.succeeded()) {
        log.warn(
            String.format("Cancelling request due to timeout after : %d ms",
                queryTimeout));
      } else {
        log.warn("Failed to send cancelling request", ar.cause());
      }
    }));

    SQLConnection sqlConnection = new SQLConnection(pgConnection, null, timerId);
    handler.handle(Future.succeededFuture(sqlConnection));
  });
}
 

private void intsAsString(RowStream<Row> sqlRowStream, Handler<AsyncResult<String>> replyHandler) {
  StringBuilder s = new StringBuilder();
  sqlRowStream.handler(row -> {
    if (s.length() > 0) {
      s.append(", ");
    }
    s.append(row.getInteger(0));
  }).exceptionHandler(e -> {
    replyHandler.handle(Future.failedFuture(e));
  }).endHandler(end -> {
    replyHandler.handle(Future.succeededFuture(s.toString()));
  });
}
 

static void close(TestContext ctx, Consumer<Handler<AsyncResult<Void>>> producer) {
  if (producer != null) {
    Async closeAsync = ctx.async();
    producer.accept(v -> {
      closeAsync.complete();
    });
    closeAsync.awaitSuccess(10000);
  }
}
 

@Override

    public HttpServerResponse push(HttpMethod method, String host, String path,
            Handler<AsyncResult<HttpServerResponse>> handler) {
        delegate.push(method, host, path, handler);
        return this;
    }