SpringWebFlux Mono.timeout metodu

Örnek

Şöyle yaparız

webClient.get()
    .uri("/endpoint")
    .retrieve()
    .bodyToMono(String.class)
    .timeout(Duration.ofSeconds(10));

SpringWebFlux Mono.retryWhen metodu

Örnek

Şöyle yaparız

this.retryBackoffSpec = Retry.backoff(maxAttempts, Duration.ofSeconds(backoffSeconds))
  .doBeforeRetry(retrySignal -> 
    log.debug("Waiting {} seconds. Retry #{} of {} after exception: {}",
        backoffSeconds, (retrySignal.totalRetriesInARow()+1), maxAttempts,
        retrySignal.failure().getLocalizedMessage()
    ))
  .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> retrySignal.failure());

this.webClient.post().uri(uri)
  .bodyValue(emailText)
  .retrieve()
  …
  .bodyToMono(CtfdUserResponse.class)
  .retryWhen(retryBackoffSpec)
  .block();

Örnek

Şöyle yaparız

WebClient webClient = WebClient.builder()
  .baseUrl("http://example.com")
  .build();

Mono<String> response = webClient.get()
    .uri("/retry-endpoint")
    .retrieve()
    .bodyToMono(String.class)
    // number of retries and backoff configuration
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
      // maximum backoff time
      .maxBackoff(Duration.ofSeconds(10))) 
    // fallback if retries all fail
    .onErrorResume(e -> Mono.just("Fallback response")); 

response.subscribe(result -> System.out.println(result));

Örnek

Şöyle yaparız

Predicate<Throwable> exceptionFilter() {
  return throwable -> throwable instanceof RuntimeException
      || (throwable instanceof WebClientResponseException
          && (throwable.getStatusCode() == HttpStatus.GATEWAY_TIMEOUT
              || throwable.getStatusCode() == HttpStatus.SERVICE_UNAVAILABLE
              || throwable.getStatusCode() == HttpStatus.BAD_GATEWAY));
}

Retry retry = Retry.backoff(3, Duration.ofSeconds(2))
    .jitter(0.7)
    .filter(exceptionFilter())
    .onRetryExhaustedThrow((retrySpec, retrySignal) -> {
      log.error("Service at {} failed to respond, after max attempts of: {}", 
        uri, retrySignal.totalRetries());
      return retrySignal.failure();
    });

SpringWebFlux Mono.fromCallable metodu

Örnek

Şöyle yaparız

return Mono.fromCallable(() ->
    getHttpBlocking("https://www.google.com"))
        .subscribeOn(Schedulers.boundedElastic())

Log çıktısı şöyledir

{"@timestamp":"2021-06-07T07:42:49.709-04:00","@version":"1",

"message":"querying google",

"logger_name":"net.kamradtfamily.blockingnono.Controller",

"thread_name":"reactor-http-nio-3",

"level":"INFO","level_value":20000}

{"@timestamp":"2021-06-07T07:42:49.939-04:00","@version":"1",

"message":"found content length 12991",

"logger_name":"net.kamradtfamily.blockingnono.Controller",

"thread_name":"boundedElastic-1",

"level":"INFO","level_value":20000}

SpringWebFlux Mono.just metodu

Giriş

Çağıran thread üzerinde çalışır. Thread değiştirmek için Mono.fromCallable().subscribeOn() kullanılır.

Örnek

Elimizde şöyle bir kod olsun

@Slf4j
@RestController
@RequestMapping("/")
public class Controller {
  RestTemplate restTemplate = new RestTemplate();
  @GetMapping("/blocking")
  public Mono<String> getFiles() {
    log.info("querying google");
    return Mono.just(getHttpBlocking("https://www.google.com"))
      .doOnNext(s -> 
           log.info("found content length {}", s.length()));
  }

  String getHttpBlocking(String url) {
    return restTemplate.getForObject(url, String.class);
  }
}

Log şöyledir. Burada Mono.just() çağrısının thread değiştirmediği görülebilir.

{"@timestamp":"2021-06-07T07:05:57.225-04:00","@version":"1",

"message":"querying google",

"logger_name":"net.kamradtfamily.blockingnono.Controller",

"thread_name":"reactor-http-nio-3",

"level":"INFO","level_value":20000}

{"@timestamp":"2021-06-07T07:05:57.640-04:00","@version":"1",

"message":"found content length 12962",

"logger_name":"net.kamradtfamily.blockingnono.Controller",

"thread_name":"reactor-http-nio-3",

"level":"INFO","level_value":20000}

SpringWebFlux Mono.switchIfEmpty metodu

Giriş

Flux.switchIfEmpty metodu ile kardeştir.

Açıklaması şöyle. Mono boş ise switchIfEmpty() ile belirtilen sonucu döner

we can emit an error signal when our mono has no values to emit.

Örnek

Şöyle yaparız

Mono<String> publisher = Mono.just("My first publisher");
publisher
  .filter(value -> value.equals("This is not my first publisher"))
  .switchIfEmpty(Mono.error(new RuntimeException("Something went wrong")))
  .subscribe(new MySubscriber());

Örnek
Şöyle yaparız.

Mono.just(new Employee().setName("Kill"))
    .switchIfEmpty(Mono.defer(() -> Mono.just(new Employee("Bill"))))
    .block()
    .getName();

Örnek

Şöyle yaparız. 

@RestController
@RequestMapping(value = "/posts")
class PostController {
  private final PostRepository posts;

  @GetMapping("/{id}")
  public Mono<Post> get(@PathVariable("id") Long id) {
    return Mono.just(id)
      .flatMap(posts::findById)
      .switchIfEmpty(Mono.error(new PostNotFoundException(id)));
  }
}

SpringWebFlux Mono Sınıfı

Giriş
Şu satırı dahil ederiz

import reactor.core.publisher.Mono;

Bu sınıf aslında Spring'e ait değil. Project Reactor'a ait. Soyut bir sınıf. [0-1) arası nesne üretir. Project Reactor Java 8+ ile kullanılır. Açıklaması şöyle.

Reactor on the other hand is Java 8+ only, so it can make full use of the new Java 8 native classes. Since Spring 5.0 is also Java 8+ only, that means Reactor has the edge in this regard.

Bu Sınıf Niçin Lazım ?
Açıklaması şöyle. Reactive Stream tiplerinin yetersiz kaldığı düşünülerek tasarlanmış. Reactive Stream'deki Publisher arayüzünü gerçekleştirir.

The Reactive Streams types are not enough; you’ll need higher order implementations to support operations like filtering and transformation. The Reactor project is a good choice here; it builds on top of the Reactive Streams specification. It provides two specializations of  Publisher<T>.

The first, Flux<T>, is a Publisher that produces zero or more values. It’s unbounded. The second, Mono<T>, is a Publisher<T> that produces zero or one value. They’re both publishers and you can treat them that way, but they go much further than the Reactive Streams specification. They both provide operators, ways to process a stream of values. Reactor types compose nicely - the output of one thing can be the input to another and if a type needs to work with other streams of data, they rely upon Publisher<T> instances.

Both Mono<T> and Flux<T> implement Publisher<T>; our recommendation is that your methods accept Publisher<T> instances but return Flux<T> or Mono<T> to help the client distinguish the kind of data its being given.

block metodu
Açıklaması şöyle

to retrieve object from mono you have to block

Örnek
Şöyle yaparız.

Mono.just(new Employee().setName("Kill"))
    .switchIfEmpty(Mono.defer(() -> Mono.just(new Employee("Bill"))))
    .block()
    .getName();

doOnError metodu - Consumer
Şöyle yaparız.

Mono<AuthorizeResponse> result = ...
response = result.doOnError(e -> {throw new RuntimeException(e);})
                 .block();

error metodu

Şöyle yaparız

@Transactional
public Mono<Store> addStore(Store store) {
  if (store.isValid()) {
    final Mono<Store> result = storePort.addStore(store);
    return result;
  }
  return Mono.error(InvalidAttributesException::new);
}

flatMap metodu

Mono.flatMap metodu yazısına taşıdım

flatMapMany metodu

Mono nesnesini Flux nesnesine çevirir. Yani 1 nesneyi OneToMany haline getirir.

fromCallable metodu

Mono.fromCallable metodu yazısına taşıdım

fromRunnable metodu

Örnek

Şöyle yaparız

Mono.fromRunnable(() -> System.out.println("Hello"))
        .subscribe(
                i -> System.out.println("Received :: " + i),
                err -> System.out.println("Error :: " + err),
                () -> System.out.println("Successfully completed"));

//Output
Hello
Successfully completed  

fromSupplier metodu
Şöyle yaparız. Supplier verilen veriyi işler ve bir sonuç döner. Bu döndürülen Mono nesnesinin sonucunu almak için ya block() metodu ya da subscribe() metodu çağrılır.

public Mono<Integer> nonBlockingSum(Integer arr[])
  throws InterruptedException {

  Mono<Integer> m = Mono.fromSupplier(() ->this.computationService.getSum(arr))
                        .subscribeOn(this.scheduler);
  return m;
}

justmetodu

Mono.just metodu yazısına taşıdım

justOrEmpty metodu

Örnek

Şöyle yaparız

Mono<String> publisher = Mono.justOrEmpy("My first publisher");

onErrorResume metodu

Mono.onErrorResume metodu yazısına taşıdım

then metodu

Mono.then metodu yazısına taşıdım

thenReturn metodu

Mono.thenReturn metodu yazısına taşıdım

switchIfEmpty metodu

Mono.switchIfEmpty metodu yazısına taşıdım