Search code examples
javamicroservicesspring-webfluxproject-reactor

Spring WebFlux: WebClient combines 2 Reactive RESTful Web Service


I'm working on a Microservices application with Reactive support using Spring WebFlux. Let see, I have a list of questions belong to a category and list of options for each question. I separate the question and the option into services with Reactive support and I want to have another service to combine them together using WebClient of Spring WebFlux. Of course, it needs to support Reactive also.

QuestionServiceImpl:

public Flux<Question> getQuestions(String categoryId) {
    WebClient client = WebClient
        .builder()
        .baseUrl(getServiceUrl())
        .build();

    WebClient.ResponseSpec responseSpec = client
        .get()
        .uri("/questions/" + categoryId)
        .retrieve();

    return responseSpec.bodyToFlux(Question.class);
}

OptionServiceImpl:

public Flux<Option> getOptions(String questionId) {
    WebClient client = WebClient
            .builder()
            .baseUrl(getServiceUrl())
            .build();

        WebClient.ResponseSpec responseSpec = client
            .get()
            .uri("/options/" + questionId)
            .retrieve();

        return responseSpec.bodyToFlux(Option.class);
}

But I don't know how to combine a question with its options in the Reactive way. Can anyone suggest some ideas?

Updated solution:

I added a new class named CompositeQuestion

@Data
@AllArgsConstructor 
public class CompositeQuestion {

    private String id;

    private String description;

    private String categoryId;

    private List<Option> options;

}

and now to get list options for a question, my code is as below:

Flux<CompositeQuestion> compositQuestion = questionsFromCoreQuestionService.flatMap(question ->
        optionService.getOptions(question.getId())
            .collectList()
            .map(options -> new CompositeQuestion(question.getId(), question.getDescription(), question.getCategoryId(), options)))
        .subscribeOn(Schedulers.elastic());

Solution

  • Let's assume that you have a class like the following:

    @Value
    public class QuestionOptions {
         private Question question;
         private List<Option> options;
    }
    

    (the @Value annotation is from Lombok)

    You can retrieve a question with its options like this:

    Flux<String> categoryIds = Flux.just("1", "2", "3");
    Flux<QuestionOptions> questionOptions = 
        categoryIds.flatMap(categoryId -> 
             // retrieve questions for each category
             questionService.getQuestions(categoryId)
                  // get options for each question 
                  .flatMap(question -> optionService.getOptions(question.getId())
                  .collectList()
                  .map(optionList -> new QuestionOptions(question, optionList))
             ))
        .subscribeOn(Schedulers.elastic()); // retrieve each question on a different thread.
    

    Note that, if the order of categories might be different than the order you requested. If that's a deal breaker for you, you might consider using concatMap() instead of flatMap(), though each request will run sequentially then.