SpringWebFlux WebClient Flux Dönüşümü

1. bodyToFlux metodu

Örnek

Şöyle yaparız

WebClient webClient = WebClient.create();
webClient.get()
    .uri("http://example.com/stream")
    .accept(MediaType.TEXT_EVENT_STREAM) // for Server-Sent Events (SSE)
    .retrieve()
    .bodyToFlux(String.class) // convert the response body to a Flux
    .subscribe(data -> System.out.println("Received: " + data));

2. toEntityFlux metodu

Örnek

Şöyle yaparız

WebClient webClient = WebClient.builder().build();
ResponseEntity<Flux<Part>> request = webClient
  .get()
  .uri("...")
  .accept(MediaType.MULTIPART_MIXED)
  .retrieve()
  .toEntityFlux((inputMessage, context) ->
    partReader
      .read(ResolvableType.forType(byte[].class), inputMessage, Map.of()))
  .block();

Örnek
Şöyle yaparız.

@GetMapping(path = "/streaming", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
@ResponseBody
public Flux<Something> streamSomething() {
  return WebClient.create()
    .get().uri("http://example.org/resource")
    .retrieve().bodyToFlux(Something.class)
    .delaySubscription(Duration.ofSeconds(5))
    .repeat();
}

Örnek

Şöyle yaparız.

WebClient webClient = WebClient.builder().baseUrl(baseUrl).build();

webClient.post().uri(uri)
  .contentType(MediaType.APPLICATION_JSON_UTF8)
  .accept(MediaType.APPLICATION_JSON_UTF8)
  .header(HttpHeaders.AUTHORIZATION, "Basic " + Base64Utils
  .encodeToString((plainCreds)
  .getBytes(Charset.defaultCharset())))
  .body(BodyInserters.fromObject(body)).retrieve()
  .bodyToFlux(EmployeeInfo.class)
  .doOnError(throwable -> {
    ...
  }).subscribe(new Consumer<EmployeeInfo>() {
    @Override
    public void accept(EmployeeInfo employeeInfo) {
      ...
    }
}); 

3. exchange + flatMap Yöntemi

Örnek

Elimizde şöyle bir kod olsun. Bu kod hem Mono hem de Flux dönebiliyor. Hem get() hem de put() işlemi için exchange() çağrısı yapıyor. Kod bir  spring bean. İskeleti şöyle

import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;

@Component
public class WebClientHelper {

  private WebClient webClient;

  public WebClientHelper() {
    webClient = WebClient.create();
  }
}

Flux dönen metodlar şöyle. Burada uri() metodu önemli. Gönderilecek uri'ye verilecek parametreler burada belirtiliyor.

public <T> Flux<T> performGetToFlux(URI uri, MultiValueMap<String, String> params,
  Class<? extends T> clazzResponse){
  return webClient.get()
        .uri(uriBuilder -> uriBuilder
            .scheme(uri.getScheme())
            .host(uri.getHost())
            .port(uri.getPort())
            .path(uri.getPath())
            .queryParams(params)
            .build()
        )
        .exchange()
        .flatMapMany(clientResponse -> clientResponse.bodyToFlux(clazzResponse));
}

Bu kodu kullanmak için şöyle yaparız

@RestController
@RequestMapping("/employeeClient")
public class EmployeeClientController {

  @Value("${employee.server.host}")
  private String employeeHost;

  @Autowired
  private WebClientHelper webClientHelper;

  @GetMapping
  public Flux<EmployeeModel> getEmployeeList(){
    return webClientHelper.performGetToFlux(URI.create(...), null, EmployeeModel.class);
  }
}