Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Reactive Spring

Reactive Spring

This session will start with a Reactive Streams and Reactive APIs introduction, then will show concretely how you can build a reactive application using Spring 5.0, Spring Boot 2.0 and Java 8 with code and demos including a JSON REST API, Mustache templates based HTML rendering with Server-Sent Events push for live updates. We will also cover how to use Spring Data Reactive and the new ‘WebClient’ for processing data coming from remote services (REST or streaming APIs). Netty will be used as the underlying non-blocking HTTP engine for both client and server. I will also give you an overview of the Spring WebFlux functional Web API based on lambdas instead of annotation and of Spring 5.0 Kotlin native support.

Sébastien Deleuze

April 24, 2017
Tweet

More Decks by Sébastien Deleuze

Other Decks in Programming

Transcript

  1. 2

  2. 9 public interface Publisher<T> { void subscribe(Subscriber<? super T> s);

    } public interface Subscriber<T> { void onSubscribe(Subscription s); void onNext(T t); void onError(Throwable t); void onComplete(); } public interface Subscription { void request(long n); void cancel(); } public interface Processor<T, R> extends Subscriber<T>, Publisher<R> { } Reactive Streams API
  3. 16 String location = "Lyon, France"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex ->

    logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> String.format("Weather in %s is %s", w.getLocation(), w.getDescription())) .subscribe(message -> logger.info(message));
  4. 16 String location = "Lyon, France"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex ->

    logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> String.format("Weather in %s is %s", w.getLocation(), w.getDescription())) .subscribe(message -> logger.info(message)); Mono<Weather> fetchWeather(String city);
  5. 16 String location = "Lyon, France"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex ->

    logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> String.format("Weather in %s is %s", w.getLocation(), w.getDescription())) .subscribe(message -> logger.info(message)); times out and emits an error after 2 sec
  6. 16 String location = "Lyon, France"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex ->

    logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> String.format("Weather in %s is %s", w.getLocation(), w.getDescription())) .subscribe(message -> logger.info(message)); logs a message in case of errors
  7. 16 String location = "Lyon, France"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex ->

    logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> String.format("Weather in %s is %s", w.getLocation(), w.getDescription())) .subscribe(message -> logger.info(message)); switches to a different service in case of error
  8. 16 String location = "Lyon, France"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex ->

    logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> String.format("Weather in %s is %s", w.getLocation(), w.getDescription())) .subscribe(message -> logger.info(message)); transforms a weather instance into a String message
  9. 16 String location = "Lyon, France"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex ->

    logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> String.format("Weather in %s is %s", w.getLocation(), w.getDescription())) .subscribe(message -> logger.info(message)); triggers the processing of the chain
  10. 18 interface ReactiveUserRepository { Mono<User> findOne(String id); Flux<User> findAll(); Mono<Void>

    save(Mono<User> user); } interface UserRepository { User findOne(String id); List<User> findAll(); void save(User user); } Blocking API Reactive API
  11. 19 interface UserRepository { User findOne(String id); List<User> findAll(); void

    save(User user); } Blocking API Method returns when whole list is received
  12. 19 interface UserRepository { User findOne(String id); List<User> findAll(); void

    save(User user); } Blocking API Throws exception if an error happens
  13. 20 interface ReactiveUserRepository { Mono<User> findOne(String id); Flux<User> findAll(); Mono<Void>

    save(Mono<User> user); } interface UserRepository { User findOne(String id); List<User> findAll(); void save(User user); } Blocking API Reactive API
  14. 21 interface ReactiveUserRepository { Mono<User> findOne(String id); Flux<User> findAll(); Mono<Void>

    save(Mono<User> user); } Reactive API Method returns immediately!
  15. 21 interface ReactiveUserRepository { Mono<User> findOne(String id); Flux<User> findAll(); Mono<Void>

    save(Mono<User> user); } Reactive API Elements are received as soon as they are available, ending with a success "onComplete" event.
  16. 21 interface ReactiveUserRepository { Mono<User> findOne(String id); Flux<User> findAll(); Mono<Void>

    save(Mono<User> user); } Reactive API Mono<Void> means error or success,
 no data will be received
  17. 24 GET /users … 200 OK
 Content-Type: application/json [ {"a":

    "foo", "b": "bar"}, {"a": "baz", "b": "boo"} ] Flux<User> as a JSON array
  18. 25 GET /users … 200 OK
 Content-Type: text/event-stream data: {"a":

    "foo", "b": "bar"} data: {"a": "baz", "b": "boo"} data: {"a": "dev", "b": "oxx"} data: {"a": "spr", "b": "ing"} data: {"a": "lov", "b": "es"} data: {"a": "and", "b": "more"} data: {"a": "data", "b": "coming"} Flux<User> as Server Sent Events
  19. 26 GET /users … 200 OK
 Content-Type: application/stream+json {"a": "foo",

    "b": "bar"} {"a": "baz", "b": "boo"} {"a": "dev", "b": "oxx"} {"a": "spr", "b": "ing"} {"a": "lov", "b": "es"} {"a": "and", "b": "more"} {"a": "data", "b": "coming"} Flux<User> as a JSON stream
  20. 29

  21. 29

  22. 30 With Reactive Spring, you can use your reactive library

    of choice for your application: RxJava Reactor Akka Streams
  23. 32

  24. 32 @Controller, @RequestMapping Spring MVC Servlet API Servlet Container Servlet

    3.1, Netty, Undertow Spring WebFlux HTTP / Reactive Streams
  25. 32 @Controller, @RequestMapping Spring MVC Servlet API Servlet Container Router

    functions Servlet 3.1, Netty, Undertow Spring WebFlux HTTP / Reactive Streams
  26. 36 WebClient Web handler WebFlux ✔ Shared resources (event loop,

    buffers) ✔ Built-in mocking capabilities
  27. 43 // Annotation-based Java @RequestMapping("/quotes/feed", produces = TEXT_EVENT_STREAM_VALUE) public Flux<Quote>

    fetchQuotesStream() { ... } route( // Functional Java without static imports RequestPredicates. RequestPredicates.accept(TEXT_EVENT_STREAM)), path( ) { ... } ) .and( "/quotes/feed" RouterFunctions.
  28. 44 // Annotation-based Java @RequestMapping("/quotes/feed", produces = TEXT_EVENT_STREAM_VALUE) public Flux<Quote>

    fetchQuotesStream() { ... } accept(TEXT_EVENT_STREAM)), path( ) { ... } ) .and( "/quotes/feed" // Functional Java with static imports route(
  29. 45 // Annotation-based Java @RequestMapping("/quotes/feed", produces = TEXT_EVENT_STREAM_VALUE) public Flux<Quote>

    fetchQuotesStream() { ... } accept(TEXT_EVENT_STREAM) { ... } and "/quotes/feed" // Functional Kotlin router { }
  30. 46 @RestController public class UserController { @GetMapping( ) public Flux<User>

    () { // ... } @GetMapping( ) public Mono<User> (@PathVariable String userId) { // ... } } fetchAll "/users" "/users/{id}" fetch
  31. 47 @Configuration class ApplicationRoutes(val userHandler: UserHandler, val blogHandler: BlogHandler, val

    shopRepository: ShopRepository) { @Bean fun appRouter() = router { GET( , userHandler:: ) GET( , userHandler:: ) } } fetchAll "/users" "/users/{id}" fetch @Bean fun nestedRouter() = router { @Bean fun dynamicRouter() = router { ... ... } }
  32. 48 @Bean fun nestedRouter() = router { @Bean fun dynamicRouter()

    = router {... } } ("/blog" and accept(TEXT_HTML)).nest { GET("/", blogHandler::findAllView) GET("/{slug}", blogHandler::findOneView) } ("/api/blog" and accept(APPLICATION_JSON)).nest { GET("/", blogHandler::findAll) GET("/{id}", blogHandler::findOne) POST("/", blogHandler::create) }
  33. 49 @Bean fun dynamicRouter() = router { } shopRepository.findAll() .toIterable()

    .forEach { shop -> GET("/${shop.id}") { req -> shopHandler.homepage(shop, req) } }
  34. 51

  35. 52 inline fun <reified T : Any> ServerResponse.BodyBuilder.body(publisher: Publisher<T>) =

    body(publisher, T::class.java) ServerResponse extensions provided in spring-webflux.jar
  36. 52 inline fun <reified T : Any> ServerResponse.BodyBuilder.body(publisher: Publisher<T>) =

    body(publisher, T::class.java) ServerResponse extensions provided in spring-webflux.jar Extend existing Spring Java type
  37. 52 inline fun <reified T : Any> ServerResponse.BodyBuilder.body(publisher: Publisher<T>) =

    body(publisher, T::class.java) ServerResponse extensions provided in spring-webflux.jar Extend existing Spring Java type Mono<ServerResponse> findAll(ServerRequest request) { return ServerResponse.ok().body(repository.findAll(), User.class); } Need to specify the type because of type erasure
  38. 52 fun findAll(req: ServerRequest) = ok().body(repository.findAll()) No need to specify

    explicitly types thanks
 to Kotlin reified type parameters inline fun <reified T : Any> ServerResponse.BodyBuilder.body(publisher: Publisher<T>) = body(publisher, T::class.java) ServerResponse extensions provided in spring-webflux.jar Extend existing Spring Java type
  39. 53 Kotlin extensions bundled with Spring 5 Reactor (via reactor-kotlin-extensions)

    ApplicationContext WebFlux client WebFlux server Spring MVC RestTemplate JDBC Spring Data extensions are coming