r2dbc ItemController listen

ItemController.java

@RestController
@RequestMapping(value = "/items")
public class ItemController {

    private final ItemService itemService;

    ...
 
    @GetMapping("/events")
    public Flux<ServerSentEvent<Event>> listenToEvents() {

        final Flux<Event> itemSavedFlux =
                this.itemService.listenToSavedItems()
                        .map(itemMapper::toResource)
                        .map(ItemSaved::new);

        final Flux<Event> itemDeletedFlux =
                this.itemService.listenToDeletedItems()
                        .map(ItemDeleted::new);

        return Flux.merge(itemSavedFlux, itemDeletedFlux)
                .map(event -> ServerSentEvent.<Event>builder()
                        .retry(Duration.ofSeconds(4L))
                        .event(event.getClass().getSimpleName())
                        .data(event).build());
    }

}