How to merge multiple vertx web client responses

Question

I am new to vertx and async programming.

I have 2 verticles communicating via an event bus as follows:

//API Verticle

 public class SearchAPIVerticle extends AbstractVerticle {
    
    
        public static final String GET_USEARCH_DOCS = "get.usearch.docs";
    
        @Autowired
        private Integer defaultPort;
    
        private void sendSearchRequest(RoutingContext routingContext) {
    
            final JsonObject requestMessage = routingContext.getBodyAsJson();
    
            final EventBus eventBus = vertx.eventBus();
            eventBus.request(GET_USEARCH_DOCS, requestMessage, reply -> {
                if (reply.succeeded()) {
                    Logger.info("Search Result = " + reply.result().body());
                    routingContext.response()
                            .putHeader("content-type", "application/json")
                            .setStatusCode(200)
                            .end((String) reply.result().body());
                } else {
                    Logger.info("Document Search Request cannot be processed");
                    routingContext.response()
                            .setStatusCode(500)
                            .end();
                }
            });
        }
    
         
        @Override
        public void start() throws Exception {
            Logger.info("Starting the Gateway service (Event Sender) verticle");
            // Create a Router
            Router router = Router.router(vertx);
    
            //Added bodyhandler so we can process json messages via the event bus
            router.route().handler(BodyHandler.create());
    
            //    Mount the handler for incoming requests
            // Find documents
            router.post("/api/search/docs/*").handler(this::sendSearchRequest);
            
            // Create an HTTP Server using default options
            HttpServer server = vertx.createHttpServer();
            // Handle every request using the router
            server.requestHandler(router)
                    //start listening on port 8083
                    .listen(config().getInteger("http.port", 8083)).onSuccess(msg -> {
                        Logger.info("*************** Search Gateway Server started on "
                                + server.actualPort() + " *************");
                    });
        }
    
        @Override
        public void stop(){
           //house keeping
        }
    
    }

//Below is the target verticle should be making the multiple web client call and merging the responses .

  @Component
        public class SolrCloudVerticle extends AbstractVerticle {
        
            public static final String GET_USEARCH_DOCS = "get.usearch.docs";
        
            @Autowired
            private SearchRepository searchRepositoryService;
        
            @Override
            public void start() throws Exception {
                Logger.info("Starting the Solr Cloud Search Service (Event Consumer) verticle");
        
                super.start();
        
                ConfigStoreOptions fileStore = new ConfigStoreOptions().setType("file")
                        .setConfig(new JsonObject().put("path", "conf/config.json"));
                ConfigRetrieverOptions configRetrieverOptions = new ConfigRetrieverOptions()
                        .addStore(fileStore);
                ConfigRetriever configRetriever = ConfigRetriever.create(vertx, configRetrieverOptions);
                configRetriever.getConfig(ar -> {
                    if (ar.succeeded()) {
                        JsonObject configJson = ar.result();
                        EventBus eventBus = vertx.eventBus();
        
                        eventBus.<JsonObject>consumer(GET_USEARCH_DOCS).handler(getDocumentService(searchRepositoryService, configJson));
               
                        Logger.info("Completed search service event processing");
        
                    } else {
                        Logger.error("Failed to retrieve the config");
                    }
                });
        
            }
        
      





      private Handler<Message<JsonObject>> getDocumentService(SearchRepository searchRepositoryService, JsonObject configJson) {
                           
                            return requestMessage -> vertx.<String>executeBlocking(future -> {
                               
                    
                                try {
            
            //I need to incorporate the logic here that adds futures to list and composes the compositefuture

/*
//Below is my logic to populate the future list
WebClient client = WebClient.create(vertx);
                List<Future> futureList = new ArrayList<>();
                for (Object collection : searchRepositoryService.findAllCollections(configJson).getJsonArray(SOLR_CLOUD_COLLECTION).getList()) {
                    Future<String> future1 = client.post(8983, "127.0.0.1", "/solr/" + collection + "/query")
                            .expect(ResponsePredicate.SC_OK)
                            .sendJsonObject(requestMessage.body())
                            .map(HttpResponse::bodyAsString).recover(error -> {
                                System.out.println(error.getMessage());
                                return Future.succeededFuture();
                            });
                    futureList.add(future1);
                }
//Below is the CompositeFuture logic, but the logic and construct does not make sense to me. What goes as first and second argument of executeBlocking method

 /*CompositeFuture.join(futureList)
                        .onSuccess(result -> {
                             result.list().forEach( x -> {
                                 if(x != null){
                                     requestMessage.reply(result.result());
                                 }
                             }
                             );
                        })
                        .onFailure(error -> {
                            System.out.println("We should not fail");
                        })

*/
        
                     future.complete("DAO returns a Json String");
                                 
             
                                } catch (Exception e) {
                                    future.fail(e);
                                }
                            }, result -> {
                                if (result.succeeded()) {
                                    requestMessage.reply(result.result());
                                } else {
                                    requestMessage.reply(result.cause()
                                            .toString());
                                }
                            });
                        }
                    
                   }

I was able to use the org.springframework.web.reactive.function.client.WebClient calls to compose my search result from multiple web client calls, as against using Future<io.vertx.ext.web.client.WebClient> with CompositeFuture. I was trying to avoid mixing Springboot and Vertx, but unfortunately Vertx CompositeFuture did not work here:

//This method supplies the parameter for the future.complete(..) line in getDocumentService(SearchRepository,JsonObject) 
 private List<JsonObject> findByQueryParamsAndDataSources(SearchRepository searchRepositoryService,
                                                   JsonObject configJson,
                                                   JsonObject requestMessage)
            throws SolrServerException, IOException {
        List<JsonObject> searchResultList = new ArrayList<>();
        for (Object collection : searchRepositoryService.findAllCollections(configJson).getJsonArray(SOLR_CLOUD_COLLECTION).getList()) {
           
            searchResultList.add(new JsonObject(doSearchPerCollection(collection.toString(), requestMessage.toString())));
        }
        return aggregateMultiCollectionSearchResults(searchResultList);
    }

public String doSearchPerCollection(String collection, String message) {

        org.springframework.web.reactive.function.client.WebClient client =
                org.springframework.web.reactive.function.client.WebClient.create();

        return client.post()
                .uri("http://127.0.0.1:8983/solr/" + collection + "/query")
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.APPLICATION_JSON)
                .body(BodyInserters.fromValue(message.toString()))
                .retrieve()
                .bodyToMono(String.class)
                .block();
    }

    private List<JsonObject> aggregateMultiCollectionSearchResults(List<JsonObject> searchList){
        //TODO: Search result aggregation
        return searchList;
    }

My use case is the second verticle should make multiple vertx web client calls and should combine the responses. If an API call falls, I want to log the error and still continue processing and merging responses from other calls. Please, any help on how my code above could be adaptable to handle the use case?

I am looking at vertx CompositeFuture, but no headway or useful example seen yet!

Answer 1

What you are looking for can done with Future coordination with a little bit of additional handling:

CompositeFuture.join(future1, future2, future3).onComplete(ar -> {
    if (ar.succeeded()) {
        // All succeeded
    } else {
        // All completed and at least one failed
    }
});

The join composition waits until all futures are completed, either with a success or a failure. CompositeFuture.join takes several futures arguments (up to 6) and returns a future that is succeeded when all the futures are succeeded, and failed when all the futures are completed and at least one of them is failed

Using join you will wait for all Futures to complete, the issue is that if one of them fails you will not be able to obtain response from others as CompositeFuture will be failed. To avoid this you should add Future<T> recover(Function<Throwable, Future<T>> mapper) on each of your Futures in which you should log the error and pass an empty response so that the future does not fail.

Here is short example:

Future<String> response1 = client.post(8887, "localhost", "work").expect(ResponsePredicate.SC_OK).send()
    .map(HttpResponse::bodyAsString).recover(error -> {
        System.out.println(error.getMessage());
        return Future.succeededFuture();
    });
Future<String> response2 = client.post(8887, "localhost", "error").expect(ResponsePredicate.SC_OK).send()
    map(HttpResponse::bodyAsString).recover(error -> {
        System.out.println(error.getMessage());
        return Future.succeededFuture();
    });

CompositeFuture.join(response2, response1)
    .onSuccess(result -> {
        result.list().forEach(x -> {
            if(x != null) {
                System.out.println(x);
            }
        });
    })
    .onFailure(error -> {
        System.out.println("We should not fail");
    });

Edit 1:

Limit for CompositeFuture.join(Future...) is 6 Futures, in the case you need more you can use: CompositeFuture.join(Arrays.asList(future1, future2, future3)); where you can pass unlimited number of futures.