Перехватчик шины событий Vertx блокирует вызов запроса - PullRequest
Новая
       98

Перехватчик шины событий Vertx блокирует вызов запроса

1 голос
/ 13 июля 2020

Я пытаюсь отправить данные по шине событий, когда получил звонок от клиента и ответ клиенту.

Все работает нормально, пока я не добавлю перехватчик на шину событий ..

Вот код: - 

public class TestVerticle extends AbstractVerticle {

    @Override
    public void start(Promise<Void> promise) {

        Router router = Router.router(vertx);

        vertx.exceptionHandler(globalExceptionHandler -> {
            System.out.println("Exception not handled in application  : " + globalExceptionHandler.getCause());
        });

        vertx.eventBus().consumer("test", handler -> {
            System.out.println("Message receive form event bus : " + handler.body().toString());
            handler.reply(handler.body().toString());
        });

        router.get().handler(this::rootHandler);
        vertx.createHttpServer().requestHandler(router).listen(8080, resultHandler -> {
            if (resultHandler.succeeded()) {
                promise.complete();
            } else {
                promise.fail(resultHandler.cause());
            }
        });

        vertx.eventBus().addOutboundInterceptor(handler -> {
            System.out.println("Outbound data : "+handler.body().toString());
        });

//        vertx.eventBus().addInboundInterceptor(handler -> {
//            System.out.println("Inbound data : " + handler.body().toString());
//        });
    }

    private void rootHandler(RoutingContext routingContext) {
        JsonObject msg = new JsonObject().put("path", routingContext.request().path());
        vertx.eventBus().request("test","Hello from Application",reply -> this.replyHandler(routingContext,reply));
    }

    private void replyHandler(RoutingContext ctx, AsyncResult<Message<Object>> reply) {
        HttpServerResponse response = ctx.response()
                .putHeader("Content-Type", "application/json");
        System.out.println("Reply from event bus : " +reply.result().body().toString());
        if (reply.succeeded()) {
            response.setStatusCode(200)
                    .setStatusMessage("OK")
                    .end(reply.result().body().toString());
        } else {
            response.setStatusCode(500)
                    .setStatusMessage("Server Error")
                    .end(new JsonObject().put("error", reply.cause().getLocalizedMessage()).encodePrettily());
        }
    }

    public static void main(String[] args) {
        Vertx.vertx().deployVerticle(new TestVerticle(), deploymenHandler -> {
            if (deploymenHandler.succeeded()) {
                System.out.println("verticle deplyed");
            } else {
                System.out.println("failed");
            }
        });

    }

    @Override
    public void stop(Promise<Void> promise) {
        System.out.println("Exiting verticle");
        promise.complete()
    }
}

Еще одно сомнение в том, что когда я останавливаю приложение из IDE, метод Stop не вызывается, но если я отменяю развертывание этой вертикали из другой, которая работает нормально.

1 Ответ

1 голос
/ 13 июля 2020

Когда вы используете перехватчики шины событий, вам придется вызвать метод next() в обработчике, иначе вы застрянете там. 

vertx.eventBus().addOutboundInterceptor(handler -> {
  System.out.println("Outbound data : "+handler.body().toString());
  handler.next();
});

vertx.eventBus().addInboundInterceptor(handler -> {
  System.out.println("Inbound data : "+handler.body().toString());
  handler.next();
});
...