What is the right way to consolidate Flux results from different micro-services?

user_3002896 0
Viewed 126

I've got an edge service which consolidates results from 3 different micro-services.

  1. Returns a Flux of customers
  2. Returns Mono of profile for customer
  3. Returns a Flux of roles for customer

What is the correct way to build a Flux of CustomerProfile objects which will include information about customer, profile and roles?

@RequiredArgsConstructor
@RestController
class CrmRestController {
    private final CrmClient crmClient;

    @GetMapping("/customer-profiles")
    public Flux<CustomerProfile> getCustomerProfiles() {
        return crmClient.getCustomerProfiles();
    }
}

@RequiredArgsConstructor
@Component
class CrmClient {
    private final WebClient http;
    private final RSocketRequester rsocket;

    Flux<Customer> getCustomers() {
        return this.http.get().uri("http://localhost:8080/customers").retrieve().bodyToFlux(Customer.class);
    }

    Flux<Role> getRolesFor(Integer customerId) {
        return this.http.get().uri("http://localhost:8183/roles").retrieve().bodyToFlux(Role.class);
    }

    Mono<Profile> getProfileFor(Integer customerId) {
        return this.rsocket.route("profiles.{customerId}", customerId).retrieveMono(Profile.class);
    }

    Flux<CustomerProfile> getCustomerProfiles() {
        //TODO: What is the right way to create CustomerProfile objects and fill in values?
        //my best guess does not compile
        return getCustomers()
                .flatMap(customer -> getProfileFor(customer.getId())
                        .map(profile -> getRolesFor(customer.getId())
                                .collectList().map(roles -> new CustomerProfile(customer, profile, roles))));
    }
}

@Data
@AllArgsConstructor
class Customer {
    private Integer id;
    private String name;
}

@Data
@AllArgsConstructor
class Profile {
    Integer id;
    Date registered;
}

@Data
@AllArgsConstructor
class Role {
    private Integer id;
    private String title;
}

@Data
@AllArgsConstructor
class CustomerProfile {
    private Customer customer;
    private Profile profile;
    private List<Role> roles;
}
project-reactor spring-webflux
1 Answers

So, like I said above, your syntax error is easy to fix, but your question demands an explanation of reactive practices.

Remember, reactive is about "non blocking" during I/O operations. Every time you call one of your dependent services you are "blocking". Calls to getRolesFor and getProfileFor are blocking when they do database lookups. Therefore they return reactive responses as Mono or Flux. This is not to say they don't block, they do, but the WebFlux/Reactive framework allows the server to do work on other requests on the same thread. In the traditional model the server would have to create new threads for other requests while the current requests thread is blocked waiting on dependent services.

In your request you can pass the customerId to both getRolesFor and getProfileFor so you want to make both service calls at the same time and combine the results.

                    |--getProfilesFor()->|
  getCustomers() -->|------------------->|-->return new CustomerProfile()
                    |--getRolesFor()---->|

This is the most "efficient", IMHO.

The code for this is:

@RequiredArgsConstructor
@Component
class CrmClient {
    private final WebClient http;

    Flux<Customer> getCustomers() {
        return this.http.get().uri("http://localhost:8080/customers").retrieve().bodyToFlux(Customer.class);
    }

    Mono<List<Role>> getRolesFor(Integer customerId) {
        ParameterizedTypeReference<List<Role>> rolesType = new ParameterizedTypeReference() {};
        return this.http.get().uri("http://localhost:8183/roles").retrieve().bodyToMono(rolesType);
    }

    Mono<Profile> getProfileFor(Integer customerId) {
        return this.http.get().uri("http://localhost:8183/roles").retrieve().bodyToMono(Profile.class);
    }

    Flux<CustomerProfile> getCustomerProfiles() {
        return getCustomers()
                .flatMap(c -> Mono.zip(getRolesFor(c.getId()), getProfileFor(c.getId()))
                        .map(t -> new CustomerProfile(c, t.getT2(), t.getT1())));
    }
}

But notice that I have changed the getRolesFor signature. This is because the a Flux signature is incorrect. You're not getting a Flux from the database but rather a Mono<List<Role>>.

The database does not return a stream of results based on some internal non determinate events for getRoles() because there will only be a small number of Roles for a single customer. Instead the database server will create a multiple row result and return it all at once.

Your complaint can be heard. But the SpringDataJpa getRolesByCustomer (or whatever it is) returns a Flux. This is the default implementation but not appropriate for this use case . You can address it by making a custom method that returns a Mono<List<Role>> which is better.

So, assuming you change the getProfilesFor() to the correct signature, the resulting code above will

  • Get the customers
  • For each customer it will
    • pass the customerId to both getProfilesFor and getRolesFor at the same time and
    • when both results are returned it will pass them to the map function which
    • will instance and return a CustomerProfile object.

But wait, there's more. The Reactive framework first builds the flow and then executes it. The creation of the flow is always blocking so the framework can initialize. In this case getCustomers() creates the flow so the build process won't start until getCustomers() is complete, which will be blocking.

To fix this you need

return Flux.from((getCustomers())...

This way the flow is created from the Flux.from() publisher and getCustomers() is not executed until the build process has completed and the flow is subscribed to.

Finally, note the all the same discussion applies to the getCustomers call as well but since there is likely no significant performance benefit to be obtained from getting a Mono<List<Customer>> as opposed to a Flux<Customer> it isn't worth the effort worrying about it.

However, if you want to know:

Mono<List<Customer>> getCustomers() {
    return this.http.get().uri("http://localhost:8080/customers").retrieve().bodyToMono(new ParameterizedTypeReference<>() {});
}
Mono<List<CustomerProfile>> getCustomerProfiles() {
    return Mono.from(getCustomers())
            .flatMap(cs -> Flux.fromIterable(cs).flatMap(c -> Mono.zip(getRolesFor(c.getId()), getProfileFor(c.getId()))
                            .map(t -> new CustomerProfile(c, t.getT2(), t.getT1())))
                    .collect(Collectors.toList()));
}

I think this is the most efficient, but, I also think that the Flux.fromIterable(cs) will make multiple async calls to the getRolesFor and getProfileFor which could be a considerable load if there are lots of customers. Really, if you are getting all the customers then you can also get all the roles and all the profiles and parallelize the construction process in the client. At that point, however, you are doing what the database server is optimzed to do so you are much, much better off creating a single query to join the entire database to a result set and create the CustomerProfiles directly from that.

Breaking down the Customer, Role, and Profile entities into separate services should be considered a distributed systems anti-pattern. Build services around business processes instead.

user_37950360
Related