Vertx EventBus Ответить 'Определенное' сообщение - PullRequest
Новая
       22

Vertx EventBus Ответить 'Определенное' сообщение

0 голосов
/ 02 октября 2018

У нас есть случай, как показано ниже:

enter image description here

Проблема заключается в том, что координатор отправляет сообщение из контекста метода и получает ответот другого: 

private void forwardToVWClient(Message msg) {

        vertx.eventBus().send(RESTClient.ADDRESS, msg.body(), deliveryOptions, res -> {
            if (res.succeeded()) {
                log.info("forwardToVWClient. VW got result : success.");
                // do not reply ok until we get an OK from the Listener verticle

            } else {
                log.error("forwardToVWClient VW got result : failure.");
                msg.fail(500, res.cause().getMessage());
            }
        });
    }

тогда у меня есть другой метод использования шины событий, где я получаю ответ: 

vertx.eventBus().consumer(ADDRESS_RESPONSE, this::handleResponseMessage);


private void handleResponseMessage(Message msg) {
        // how to reply the message received in the context of forwardToVWClient ?? 
}

Итак, как я могу ответить на сообщение в контексте forwardToVWClient, когда я получу ответ в handleResponseMessage?

Пока пара идей:

  1. Поместить сообщение в контекст vertx?
  2. Объект сообщения имеет поле: .replyAddress(), которое возвращает int, я сохраняю его в статическом ConcurrentHashMap и использую его для ответа на конкретное сообщение.Я опубликую более подробную информацию как ответ.

Есть ли лучший способ?

1 Ответ

0 голосов
/ 03 октября 2018

Один из способов достижения этого - сохранить поле replyAddress сообщения и использовать его для отправки сообщения отправителю.

Ниже приведен упрощенный код, который показывает, как: 

public class VehicleStateCoordinatorVerticle extends AbstractVerticle {


    final static String ADDRESS_REQUEST = "CoordinatorRequest";
    final static String ADDRESS_RESPONSE = "CoordinatorResponse";

    static ConcurrentHashMap<String, VWApiRequest> pendingCommands = new ConcurrentHashMap<>();


    public void start() {
        vertx.eventBus().consumer(ADDRESS_REQUEST, this::handleRequestMessage);
        vertx.eventBus().consumer(ADDRESS_RESPONSE, this::handleResponseMessage);
        log.info("===== VehicleStateCoordinatorVerticle - bus consumer ready =====");
    }

    private void handleRequestMessage(Message msg) {

            // .... omitted for brevity
            // save the replyAddress and the command for later/callback
            cmd.setReplyAddress(msg.replyAddress());
            pendingCommands.put(cmd.getVwReference(), cmd);


            forwardToVWClient(msg);
    }


    private void forwardToVWClient(Message msg) {

        vertx.eventBus().send(AbstractOEMClientVerticle.ADDRESS, msg.body(), deliveryOptions, res -> {
            if (res.succeeded()) {
                log.info("forwardToVWClient. VW got result : success.");
                // do not reply ok until we get an OK from the VWAPIServer verticle

            } else {
                log.error("forwardToVWClient VW got result : failure.");
                msg.fail(500, res.cause().getMessage());
            }
        });
    }



    private void handleResponseMessage(Message msg) {

        //..
        VWApiRequest vwApiRequest = pendingCommands.get(vwReference);
        if(vwApiRequest == null){
            log.error("No pending vwApiRequest could be found!");
            return;
        }

        /**
         * Instead of targeting the RESTApi address,
         * we use the replyAddress to target the specific message that is pending response.
         */
        vertx.eventBus().send(vwApiRequest.getReplyAddress(), body, deliveryOptions, res -> {
            if (res.succeeded()) {
             // cheers!
            }
            else{
                log.error("Error in handleResponseMessage {}", res.cause().getMessage());
            }

        });
    }
...