ReactiveRestApi.java

  @Bean
  public Action<Chain> handlers(PersonRepository persons,
                                HotStream<Person> personStream,
                                ObjectMapper jsonMapper,
                                ModelMapper beanMapper) {
    return (chain) -> {
      chain.handler("person", ctx ->
          ctx.byMethod(spec ->
             spec
                 .get(c -> {
                   c.render(Streams.just("")
                                   .dispatchOn(get())
                                   .map(o -> persons.findAll()));
                 })
                 .put(c -> {
                   Person p = c.parse(fromJson(Person.class));

                   c.render(Streams.just(p.getId())
                                   .dispatchOn(get())
                                   .<Person>map(persons::findOne)
                                   .observe(pers -> beanMapper.map(p, pers))
                                   .map(persons::save));
                 })
                 .post(c -> {
                   Person p = c.parse(fromJson(Person.class));

                   c.render(Streams.just(p)
                                   .dispatchOn(get())
                                   .map(persons::save)
                                   .observe(personStream::broadcastNext));
                 })
          ));

      chain.handler("person/updates", ctx -> websocketBroadcast(ctx, personStream.map(p -> p.toJson(jsonMapper))));
    };
  }