Sketch of ReactiveStream interface-based API

ReactiveStreamApi.java

public interface PersonStream extends ReactiveStream<Person> {

  @MapAction("@personRepo.merge($1, $2)")
  ReactivePromise<Person> merge(String id, Person p);

}

@ReactiveRestController
public class MyController {
  
  private final EventBus events;
  private final MyService service;
  
  @Inject
  public MyController(EventBus events, MyService service) {
    this.events = events;
    this.service = service;
  }

  @RequestMapping(value="/person/{id}", method=HttpMethod.PATCH)
  public void update(HttpExchange xchng, @RequestBody Person p, @PathParam("id") String id) {
    ReactiveStreams.just(PersonStream.class, body)
        .observe(service::firePreMergeAction)
        .merge(id, p)
        .onSuccess(p -> {
          events.notify("some.topic", Event.wrap(p));
          xchng.redirect(xchng.getRequestUri());
        })
        .onError(xchng::internalServerError);
    
  }

}

Unsure what ReactiveStream is here, but could you not use RS's Publisher or something like Source?