Java 类io.vertx.core.eventbus.ReplyFailure 实例源码

protected void send(Request request, String target) {
    DeliveryOptions options = new DeliveryOptions().setSendTimeout(request.timeout());

    core.bus().send(target, request.data(), options, send -> {
        if (send.succeeded()) {
            request.write(send.result().body());
        } else {
            Throwable exception = send.cause();

            if (exception instanceof ReplyException) {
                ReplyFailure status = ((ReplyException) exception).failureType();
                exceptionHandlers.get(status).accept(request);
            } else {
                request.error(send.cause());
            }
        }
    });
}

private static JsonObject replyJson(ReplyException ex) {
  JsonObject jsonObject = new JsonObject();
  DefaultErrorCode errorCode = DefaultErrorCode.getCode(ex.failureCode());
  if (errorCode != null) {
    jsonObject.put("code", errorCode.getNumber())
            .put("message", errorCode.getMessage());
  } else {
    ReplyFailure replyFailure = ex.failureType();
    if (replyFailure == ReplyFailure.NO_HANDLERS) {
      jsonObject.put("code", DefaultErrorCode.SERVICE_UNAVAILABLE.getNumber())
              .put("message", DefaultErrorCode.SERVICE_UNAVAILABLE.getMessage());
    } else if (replyFailure == ReplyFailure.TIMEOUT) {
      jsonObject.put("code", DefaultErrorCode.TIME_OUT.getNumber())
              .put("message", DefaultErrorCode.TIME_OUT.getMessage());
    } else if (replyFailure == ReplyFailure.RECIPIENT_FAILURE) {
      jsonObject.put("code", ex.failureCode())
              .put("message", ex.getMessage());
    } else {
      jsonObject.put("code", DefaultErrorCode.UNKOWN.getNumber())
              .put("message", DefaultErrorCode.UNKOWN.getMessage());
    }
  }
  return jsonObject;
}

private void failed(Future<RpcResponse> completed,
                    Throwable throwable) {
  if (throwable instanceof ReplyException) {
    ReplyException ex = (ReplyException) throwable;
    if (ex.failureType() == ReplyFailure.NO_HANDLERS) {
      SystemException resourceNotFoundEx =
              SystemException.create(DefaultErrorCode.SERVICE_UNAVAILABLE)
                      .set("details", "No handlers");
      completed.fail(resourceNotFoundEx);
    } else {
      ErrorCode errorCode = CustomErrorCode.create(ex.failureCode(), ex.getMessage(), 400);
      SystemException systemException
              = SystemException.create(errorCode);
      completed.fail(systemException);
    }
  } else {
    completed.fail(throwable);
  }
}

@Test
public void testCallWithMessageParamWrongType() {
  JsonObject message = new JsonObject();
  message.put("object", new JsonObject().put("foo", "bar"));
  message.put("str", 76523);
  message.put("i", 1234);
  message.put("char", (int)'X'); // chars are mapped to ints
  message.put("enum", SomeEnum.BAR.toString()); // enums are mapped to strings
  vertx.eventBus().send("someaddress", message, new DeliveryOptions().addHeader("action", "invokeWithMessage").setSendTimeout(500), onFailure(t -> {
    assertTrue(t instanceof ReplyException);
    ReplyException re = (ReplyException) t;
    // This will as operation will fail to be invoked
    assertEquals(ReplyFailure.RECIPIENT_FAILURE, re.failureType());
    testComplete();
  }));
  await();
}

@Override
public void replyFailure(String address, ReplyFailure failure) {
    switch (failure) {
        case TIMEOUT:
            counterService.increment("reply.failures.timeout");
            break;
        case NO_HANDLERS:
            counterService.increment("reply.failures.noHandlers");
            break;
        case RECIPIENT_FAILURE:
            counterService.increment("reply.failures.recipientFailure");
            break;
    }
}

/**
 *  Use to terminate the application using a HTTP Post
 *  It requires an AdminKey header to work
 */
private void shutdownHandler(final RoutingContext ctx) {
    // check for AdminKey header
    String adminKey = this.config().getString("AdminKey");
    if (adminKey == null || adminKey.equals(ctx.request().getHeader("AdminKey"))) {
        // TODO: check the body for the right credentials
        this.shutdownExecution(ctx.response());
    } else {
        ctx.fail(new ReplyException(ReplyFailure.RECIPIENT_FAILURE, 401, "Sucker nice try!"));
    }
}

@Before
public void createServer(TestContext testContext) {
  vertx = Vertx.vertx();
  Router router = Router.router(vertx);
  router.route().handler(BodyHandler.create());
  router.route().handler(BaseHandler.create());
  router.get("/ex/sys").handler(rc -> {
    throw SystemException.create(DefaultErrorCode.MISSING_ARGS);
  });
  router.get("/ex/validate").handler(rc -> {
    Multimap<String, String> error = ArrayListMultimap.create();
    throw new ValidationException(error);
  });
  router.get("/ex/unkown").handler(rc -> {
    throw new RuntimeException("xxxx");
  });
  router.get("/ex/reply").handler(rc -> {
    throw new ReplyException(ReplyFailure.TIMEOUT, DefaultErrorCode.INVALID_REQ.getNumber(),
                             DefaultErrorCode.INVALID_REQ.getMessage());
  });

  router.get("/ex/reply2").handler(rc -> {
    throw new ReplyException(ReplyFailure.RECIPIENT_FAILURE);
  });
  router.route().failureHandler(FailureHandler.create());
  vertx.createHttpServer().requestHandler(router::accept)
          .listen(port, testContext.asyncAssertSuccess());
}

public void consumerSendTimeout(@Observes @VertxConsumer(TEST_BUS_TIMEOUT) VertxEvent event) {
    assertEquals(TEST_BUS_TIMEOUT, event.getAddress());
    event.messageTo(TEST_SLOW_HANDLER).setDeliveryOptions(new DeliveryOptions().setSendTimeout(10)).send("foo", (r) -> {
        if (r.failed()) {
            ReplyException exception = (ReplyException) r.cause();
            if (exception.failureType().equals(ReplyFailure.TIMEOUT)) {
                SYNCHRONIZER.add("timeout");
            }
        }
    });
}

/**
 * Test code of {@link ReplyException}
 */
@Test
public void testThrowableToCodeReplyException() {
  int expectedCode = 505;

  Throwable throwable = new ReplyException(ReplyFailure.NO_HANDLERS, expectedCode, "Message");

  int statusCode = ThrowableHelper.throwableToCode(throwable);
  assertEquals(expectedCode, statusCode);
}

@Override
public void replyFailure(String address, ReplyFailure failure) {
  Map<String, String> labels = new HashMap<>();
  labels.putAll(defaultLabels);
  labels.put(SENSISION_CLASS_EB_LABEL_ADDRESS, address);
  incrementMetric(SENSISION_CLASS_EB_MESSAGE_REPLY_ERROR_COUNT, labels);
}

@Test
public void testEventBusMetricsReplyNoHandlers() {
  vertx.eventBus().send("foo", "bar", new DeliveryOptions().setSendTimeout(300), ar -> {
    assertTrue(ar.failed());
    testComplete();
  });

  await();

  JsonObject metrics = metricsService.getMetricsSnapshot(vertx.eventBus());
  assertCount(metrics.getJsonObject("messages.reply-failures"), 1L);
  assertCount(metrics.getJsonObject("messages.reply-failures." + ReplyFailure.NO_HANDLERS), 1L);
}

@Test
public void testEventBusMetricsReplyTimeout() {
  vertx.eventBus().consumer("foo").handler(msg -> {});

  vertx.eventBus().send("foo", "bar", new DeliveryOptions().setSendTimeout(300), ar -> {
    assertTrue(ar.failed());
    testComplete();
  });

  await();

  JsonObject metrics = metricsService.getMetricsSnapshot(vertx.eventBus());
  assertCount(metrics.getJsonObject("messages.reply-failures"), 1L);
  assertCount(metrics.getJsonObject("messages.reply-failures." + ReplyFailure.TIMEOUT), 1L);
}

@Test
public void testEventBusMetricsReplyRecipientFailure() {
  vertx.eventBus().consumer("foo").handler(msg -> msg.fail(1, "blah"));

  vertx.eventBus().send("foo", "bar", new DeliveryOptions(), ar -> {
    assertTrue(ar.failed());
    testComplete();
  });

  await();

  JsonObject metrics = metricsService.getMetricsSnapshot(vertx.eventBus());
  assertCount(metrics.getJsonObject("messages.reply-failures"), 1L);
  assertCount(metrics.getJsonObject("messages.reply-failures." + ReplyFailure.RECIPIENT_FAILURE), 1L);
}

@Test
public void testFailingMethod() {
  proxy.failingMethod(onFailure(t -> {
    assertTrue(t instanceof ReplyException);
    ReplyException re = (ReplyException) t;
    assertEquals(ReplyFailure.RECIPIENT_FAILURE, re.failureType());
    assertEquals("wibble", re.getMessage());
    testComplete();
  }));
  await();
}

@Test
public void testCallWithMessageInvalidAction() {
  JsonObject message = new JsonObject();
  message.put("object", new JsonObject().put("foo", "bar"));
  message.put("str", "blah");
  message.put("i", 1234);
  vertx.eventBus().send("someaddress", message, new DeliveryOptions().addHeader("action", "yourmum").setSendTimeout(500), onFailure(t -> {
    assertTrue(t instanceof ReplyException);
    ReplyException re = (ReplyException) t;
    // This will as operation will fail to be invoked
    assertEquals(ReplyFailure.RECIPIENT_FAILURE, re.failureType());
    testComplete();
  }));
  await();
}

@Test
public void testConnectionTimeout() {

  consumer.unregister();
  long timeoutSeconds = 2;
  consumer = ProxyHelper.registerService(TestService.class, vertx, service, SERVICE_ADDRESS, timeoutSeconds);

  proxy.createConnection("foo", onSuccess(conn -> {
    long start = System.nanoTime();

    conn.startTransaction(onSuccess(res -> {
      assertEquals("foo", res);

      vertx.eventBus().consumer("closeCalled").handler(msg -> {
        assertEquals("blah", msg.body());

        long duration = System.nanoTime() - start;
        assertTrue(String.valueOf(duration), duration >= SECONDS.toNanos(timeoutSeconds));

        // Should be closed now
        conn.startTransaction(onFailure(cause -> {
          assertNotNull(cause);
          assertTrue(cause instanceof ReplyException);
          assertFalse(cause instanceof ServiceException);
          ReplyException re = (ReplyException) cause;
          assertEquals(ReplyFailure.NO_HANDLERS, re.failureType());
          testComplete();
        }));


      });

    }));
  }));

  await();
}

@Test
public void testConnectionWithCloseFutureTimeout() {

  consumer.unregister();
  long timeoutSeconds = 2;
  consumer = ProxyHelper.registerService(TestService.class, vertx, service, SERVICE_ADDRESS, timeoutSeconds);

  long start = System.currentTimeMillis();
  proxy.createConnectionWithCloseFuture(onSuccess(conn -> {

    vertx.eventBus().consumer("closeCalled").handler(msg -> {
      assertEquals("blah", msg.body());

      long now = System.currentTimeMillis();
      assertTrue(now - start > timeoutSeconds * 1000);

      // Should be closed now
      conn.someMethod(onFailure(cause -> {
        assertNotNull(cause);
        assertTrue(cause instanceof ReplyException);
        assertFalse(cause instanceof ServiceException);
        ReplyException re = (ReplyException) cause;
        assertEquals(ReplyFailure.NO_HANDLERS, re.failureType());
        testComplete();
      }));
    });
  }));

  await();
}

@Test
public void testLongDelivery2() {
  TestService proxyLong = TestService.createProxyLongDelivery(vertx, SERVICE_ADDRESS);
  proxyLong.longDeliveryFailed(onFailure(t -> {
    assertNotNull(t);
    assertTrue(t instanceof ReplyException);
    assertFalse(t instanceof ServiceException);
    ReplyException re = (ReplyException) t;
    assertEquals(ReplyFailure.TIMEOUT, re.failureType());
    testComplete();
  }));
  await();
}

@Override
public void replyFailure(String address, ReplyFailure failure) {
    dispatch(m -> m.replyFailure(address, failure));
}

@Override
public void replyFailure(@NotNull String address, @NotNull ReplyFailure failure) {
  failures.labels(AddressResolver.instance.apply(address), "reply", failure.name()).inc();
}

@Override
public void replyFailure(String address, ReplyFailure failure) {
  replyFailures.increment();
}

@Override
public void replyFailure(String address, ReplyFailure failure) {
  replyFailures.mark();
  meter("messages", "reply-failures", failure.name()).mark();
}

@Override
public Future<Message<JsonObject>> apply(Message<JsonObject> msg) {
  final String authorization = msg.headers().get("auth-token");

  if (authorization == null) {
    return Future.failedFuture(new ReplyException(ReplyFailure.RECIPIENT_FAILURE, 401, "Unauthorized"));
  }

  Future<Message<JsonObject>> fut = Future.future();

  jwtAuth.authenticate(new JsonObject().put("jwt", authorization), authenticate -> {
    if (authenticate.failed()) {
      fut.fail(new ReplyException(ReplyFailure.RECIPIENT_FAILURE, 500, authenticate.cause().getMessage()));
      return;
    }

    final User user = authenticate.result();

    if (user == null) {
      fut.fail(new ReplyException(ReplyFailure.RECIPIENT_FAILURE, 403, "Forbidden"));
      return;
    }

    final int requiredcount = authorities == null ? 0 : authorities.size();

    if (requiredcount > 0) {

      AtomicInteger count = new AtomicInteger();
      AtomicBoolean sentFailure = new AtomicBoolean();

      Handler<AsyncResult<Boolean>> authHandler = res -> {
        if (res.succeeded()) {
          if (res.result()) {
            if (count.incrementAndGet() == requiredcount) {
              // Has all required authorities
              fut.complete(msg);
            }
          } else {
            if (sentFailure.compareAndSet(false, true)) {
              fut.fail(new ReplyException(ReplyFailure.RECIPIENT_FAILURE, 403, "Forbidden"));
            }
          }
        } else {
          fut.fail(new ReplyException(ReplyFailure.RECIPIENT_FAILURE, 500, res.cause().getMessage()));
        }
      };
      for (String authority : authorities) {
        if (!sentFailure.get()) {
          user.isAuthorised(authority, authHandler);
        }
      }
    } else {
      // No auth required
      fut.complete(msg);
    }
  });

  return fut;
}

public ServiceException(int failureCode, String message, JsonObject debugInfo) {
  super(ReplyFailure.RECIPIENT_FAILURE, failureCode, message);
  this.debugInfo = debugInfo;

}