SpringWebFlux ve RestController

Giriş
Açıklaması şöyle

As Spring 5.x comes with Reactor implementation, if we want to build REST APIs using imperative style programming with Spring servlet stack, it still supports.

Örnek - FluxSink ile Kendi Flux Sınıfımız

Elimizde şöyle bir controller olsun. Bir tane Flux nesnesi Course nesnesini çektikçe bunu CourseDto olarak gönderiyor.

import dev.bluvolve.reactive.courseservice.course.mappers.CourseMapper;
import dev.bluvolve.reactive.courseservice.course.processors.CourseCreatedEventProcessor;

import reactor.core.publisher.Flux;

@RestController
public class CourseController {

  private final CourseMapper mapper;
  private final Flux<CourseCreated> events;

  public CourseController(CourseCreatedEventProcessor processor,
                          CourseMapper mapper) {

    this.mapper = mapper;
    this.events = Flux.create(processor).share();
  }

  @GetMapping(value = "/course/sse", produces = "text/event-stream;charset=UTF-8")
  public Flux<CourseDto> stream() {
    return this.events.map(event -> {
      CourseDto dto = this.mapper.entityToDto((Course) event.getSource());
      return dto;
    });
  }
  ... 
}

Flux.create() metodunun imzası şöyle. Yani bir tane FluxSink'ten kalıtan java.util.function.Consumer istiyor.

Flux.create(Consumer<? super FluxSink<T>> emitter)    

Consumer şöyledir. Kendisine CourseCreated SpringEvent'i geldikçe bunu kuyruğa yazar. Kuyruğu boşaltan thread ise FluxSink.next(event) çağrısı ile sink'e yazar. Sink'i dinleyen controller ise CourseCreatedEvent nesnesini mapper vasıtasıyla CourseDto nesnesine çevirir ve gönderir.

import dev.bluvolve.reactive.courseservice.course.events.CourseCreated;

import reactor.core.publisher.FluxSink;
import java.util.function.Consumer;

@Component
public class CourseCreatedEventProcessor
        implements ApplicationListener<CourseCreated>,
        Consumer<FluxSink<CourseCreated>> {

  private final Executor executor;
  private final BlockingQueue<CourseCreated> queue = new LinkedBlockingQueue<>();

  CourseCreatedEventProcessor(Executor executor) {
    this.executor = executor;
  }

  @Override
  public void onApplicationEvent(CourseCreated event) {
    this.queue.offer(event);
  }

  @Override
  public void accept(FluxSink<CourseCreated> sink) {
    this.executor.execute(() -> {
      while (true)
        try {
          CourseCreated event = queue.take();
          sink.next(event);
        }
        catch (InterruptedException e) {
          ...
        }
      });
  }
}

Örnek
Şöyle yaparız. Burada RestController Mono<...> veya Flux<...> tipleri dönüyor. çünkü service nesnesi Mono ve Flux döndürüyor.

@RestController
public class ProductController {
  @Autowired
  private ProductService productService;

  //This endpoint allows to create a product.
  @PostMapping("/product")
  @ResponseStatus(HttpStatus.CREATED)
  public Mono<Product> createProduct(@RequestBody Product product){
    return productService.save(product);
  }

  //This endpoint gives all the products
  @GetMapping("/products")
  public Flux<Product> getAllProducts(){
    return productService.getAllProducts();
  }

  //This endpoint allows to delete a product
  @DeleteMapping("/product/{id}")
  public Mono<Void> deleteProduct(@PathVariable int id){
    return productService.deleteProduct(id);
  }

  //This endpoint allows to update a product
  @PutMapping("product/{id}")
  public Mono<ResponseEntity<Product>> updateProduct(@RequestBody Product product){
    return productService.update(product);
  }
}

Örnek

Burada POST edilen parametreyi de Mono tipine çeviriyoruz

@PostMapping(path = "", produces = MediaType.APPLICATION_JSON_UTF8_VALUE,
             consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
public Mono<Reservation> createReservation(@RequestBody Mono<Reservation>
  reservationRequest) {
  return reservationService.createReservation(reservationRequest);
}

Örnek

Şöyle yaparız. Burada Repository Mono tipi döndürmüyor. Mono.just().xxx() şeklin de kendimiz kodluyoruz

@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)));
  }
  @PutMapping("/{id}")
  public Mono<Post> update(@PathVariable("id") Long id, @RequestBody Post post) {
    return this.posts.findById(id)
      .map(p -> {
        p.setTitle(post.getTitle());
        p.setContent(post.getContent());
        return p;
      })
      .flatMap(this.posts::save);
   }
   
}