Psuedo-code around a ComposableEventRepository

ComposableEventRepository.java

public interface ComposableEventRepository<V, K extends Serializable> extends ComposableMessagingRepository<Event<V>, K> {

  Stream<Event<V>> receive(Selector sel);

}

ComposableMessagingRepository.java

public interface ComposableMessagingRepository<V, K extends Serializable> extends ComposableRepository<V, K> {

  /**
   * Send the given message using the given routing key.
   *
   * @param key
   *    The routing key to use.
   * @param message
   *    The message to send.
   *
   * @return a {@link reactor.core.composable.Promise} that will be fulfilled when the messages has been transmitted.
   */
  Promise<Void> send(K key, V message);

  /**
   * Receive messages sent to the given routing key.
   *
   * @param key
   *    The routing key to use.
   *
   * @return a {@link reactor.core.composable.Stream} that will be populated by messages sent to the given routing key.
   */
  Stream<V> receive(K key);

}

PersonReactor.java

public interface PersonReactor extends ComposableEventRepository<Person, String> {}

ServiceBean.java

class ServiceBean {
  
  @Autowired
  PersonReactor personReactor;

  public void afterPropertiesSet() throws Exception {
    // Create new Persons
    personReactor.receive("create")
                 .consume(new Consumer<Event<Person>>() {
                   public void accept(Event<Person> ev) {}
                 });

    // Update existing Persons
    personReactor.receive("update")
                 .consume(new Consumer<Event<Person>>() {
                   public void accept(Event<Person> ev) {}
                 });

    // Audit all actions on Persons
    personReactor.receive(R("(.*)"))
                 .consume(new Consumer<Event<Person>>() {
                   public void accept(Event<Person> ev) {}
                 });
  }

  public void update(Person p) {
    // Update Person object
    mutate(p);

    personReactor.send("update", Event.wrap(p));
  }

}