Error in webflux webclient with response 200

Question

I am creating an example usisng spring boot webflux that is super simple. Two services and one of them call reactively two times the service of the other. Bascially one controller of the flights "microservice" can register Flights with departure airport code and landing airport code and it will call a Mono endpoint of the airports service to know if the airport code exists.

All code can be found here: https://github.com/Gaboxondo/springbootwebfluxdemo/blob/main/error.log

Just for explaining it with some code that you can find on the github repo:

@Service
@AllArgsConstructor
@Slf4j
public class FlightsServiceImpl implements FlightsService {

    private final FlightRepository flightRepository;
    private final FlightFactory flightfactory;
    private final AirportsService airportsService;

    @Override
    public Mono<Flight> createFlight(String departureAirportCode, String landingAirportCode, Double price) {
        Mono<Boolean> departCodeMonoValid = airportsService.airportCodeIsValid( departureAirportCode );
        Mono<Boolean> landingCodeMonoValid = airportsService.airportCodeIsValid( landingAirportCode );
        return Mono.zip(departCodeMonoValid,landingCodeMonoValid )
            .flatMap( data-> {
                if(!(data.getT1() && data.getT2())){
                    throw new ControlledErrorException( "error.code.09","the airports codes are not valid" );
                }
            return flightRepository.saveFlight( flightfactory.createFlight( departureAirportCode,landingAirportCode,price ) );
        });
    }

The client to call the other service is the following:

@Component
@AllArgsConstructor
public class AirportsWebClient {

    //I know all is hardcoded and should be configurable with some @ConfigurationProperties, this is just for testing
    public Mono<Boolean> validateAirportCode(String airportCode){
        WebClient airportsWebClient = WebClient.builder().baseUrl( "http://localhost:8082" ).build();
        return airportsWebClient.get().uri( "/v1/airports/validate/" + airportCode ).accept(
                MediaType.valueOf( MediaType.TEXT_EVENT_STREAM_VALUE ) )
            .retrieve().bodyToMono( Boolean.class );
    }
}

and the endpoint in the airport service is this one:

@RestController
@RequestMapping("/v1/airports")
@AllArgsConstructor
public class AirportsController {

    private static final List<String> validCodes = List.of("MLG","ROM");

    @GetMapping(value = "/validate/{airportCode}",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @Operation(description = "Indicates if the airport code is valid or not")
    @ResponseStatus(HttpStatus.OK)
    public Mono<Boolean> validateAirportCode(@PathVariable String airportCode){
        if(validCodes.contains( airportCode )){
            return Mono.just( true ).log();
        }else {
            return Mono.just( false ).log();
        }
    }
}

When I invoque the service to create a flight that also calls the other "microservice" for validating the airpotrs code I get the following error:

2024-05-14 11:05:53,612 service=flights-service ERROR 664329712a5442f77e1ce9dc12f43b11 class=reactor.Mono.MapFuseable.1 
org.springframework.web.reactive.function.client.WebClientResponseException: 200 OK from GET http://localhost:8082/v1/airports/validate/ROM
Caused by: java.lang.UnsupportedOperationException: ServerSentEventHttpMessageReader only supports reading stream of events as a Flux
    at org.springframework.http.codec.ServerSentEventHttpMessageReader.readMono(ServerSentEventHttpMessageReader.java:218)
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    *__checkpoint ⇢ Body from GET http://localhost:8082/v1/airports/validate/ROM [DefaultClientResponse]

I also added the log in the repo:

https://github.com/Gaboxondo/springbootwebfluxdemo/blob/main/error.log

I do not really understand what I am doing wrong, because the two reactives calls are returning 200 with a Mono Boolean but the error say something about that can only be flux.

You can also can run the two services without any special configuration just running the docker compose to create the mongoDb server that already creates the two databases.

Then in postman just call the service:

or with curl

curl --location 'http://localhost:8081/v1/flights' \
--header 'Content-Type: application/json' \
--data '{
    "departureAirportCode" : "MLG",
    "landingAirportCode": "ROM",
    "price": 100
}'

Thanks in advance

Answer 1

Why are you using MediaType.TEXT_EVENT_STREAM_VALUE in both com.mashosoft.airportsService.interfaces.web.AirportsController#validateAirportCode and com.mashosoft.flightsService.infrastructure.airports.client.AirportsWebClient#validateAirportCode ?

Doing this makes your controller switch to SSE mode which indeed sends a Flux and not a Mono.

A quick fix could be to simply switch to the following code:

  @GetMapping(value = "/validate/{airportCode}", produces = MediaType.APPLICATION_JSON_VALUE)
  @Operation(description = "Indicates if the airport code is valid or not")
  @ResponseStatus(HttpStatus.OK)
  public Mono<Boolean> validateAirportCode(@PathVariable String airportCode) {
    if (validCodes.contains(airportCode)) {
      return Mono.just(true).log();
    } else {
      return Mono.just(false).log();
    }
  }

and

    public Mono<Boolean> validateAirportCode(String airportCode){
        WebClient airportsWebClient = WebClient.builder().baseUrl( "http://localhost:8082" ).build();
        return airportsWebClient.get().uri( "/v1/airports/validate/" + airportCode )
            .retrieve().bodyToMono( Boolean.class );
    }

Running the curl then gives:

➜  ~ curl -X POST --location "http://127.0.0.1:8081/v1/flights" \
    -H "Content-Type: application/json" \
    -H "Accept: application/x-ndjson" \
    -d '{
          "departureAirportCode": "ROM",
          "landingAirportCode": "MLG",
          "price": 100
        }'

{"id":"99465ee5-2879-4f03-923c-1acf825adf50","departureAirportCode":"ROM","landingAirportCode":"MLG","price":100.0}