Vertx how to make Eventbus request to wait for consumer message reply

user_3458271 0
Viewed 2381

As I am new to Vertx, I was trying request- response using Eventbus, But While trying that I am stuck at once place, what I had done:

EventBus bus = vertx.eventBus();
//Here I call the request
bus.<JsonObject>request("previewdata", m ,this::handle);

public  void handle(AsyncResult<Message<JsonObject>> result) {
    //request get fails before consumer don't send reply within 30 seconds
    if(result.succeeded()){
        System.out.println("Answer: "+Thread.currentThread().getName());
        System.out.println(result.result().body());
    } else{
        result.cause().printStackTrace();
    }
}

IDEng ie = new IDEng();
//Consumer of request
bus.<JsonObject>consumer("previewdata", this::getPreviewData);

public void getPreviewData(Message<JsonObject> message) { 
     JsonObject json = message.body();
     for (int i=0; i<10000; i++) {
         json.put("flag"+i, "got"+i);
     }
     try {
         //Only for example I had put wait of 40 seconds (so it fail as greate than 30 seconds) it may take more time
         Thread.sleep(40000);
     }catch (Exception e) {
        // TODO: handle exception
     }
     message.reply(json);
}

I had created one vertx example using Worker Verticle and I have one request and consumer, But request is not waiting for consumer to reply the response and it's get failed in 30 seconds as it's Default timeout is 30 seconds, But per our real time scenario sometime it is not sure that how much time will request take. So please help me how can I wait for response.

I know we can set Timeout using DeliveryOptions but that's not a proper way as we never estimate some tasks time if it depends on third server. E.g

new DeliveryOptions().setSendTimeout(50000)

Please let me know if someone can help me out is this or am I doing this in wrong way?

vert.x java
1 Answers

how about decoupling the response to a different bus-address e.g "previewdataresponse":

Verticle A.)

bus.<JsonObject>request("previewdata", m ,this::handle);
// show error if reply is not success

bus.<JsonObject>consumer("previewdataresponse", this::handleResponse);

public void handleResponse(Message<JsonObject> message) { 
    // do something with the reponse
}

Verticle B.)

bus.<JsonObject>consumer("previewdata", this::getPreviewData);

public void getPreviewData(Message<JsonObject> message) { 
     JsonObject json = message.body();
     //...
     message.reply(json); // acknowledge that you received the message
     try {
         Thread.sleep(40000);
         bus.<JsonObject>send("previewdataresponse", json);
     }catch (Exception e) {
        // TODO: handle exception
     }
}
user_72630270
Related