Event-Sourcing和CQRS实战案例

public final class BookingView {

  ...

  public static BookingView create(BookingManager mgr) {
    final AtomicBoolean working = new AtomicBoolean(false);
    final AtomicLong last  = new AtomicLong();
    final AtomicLong total = new AtomicLong();
        
    final String table = mgr.getTableIdentifier().getTableName();
    final String field = Booking.ID.identifier().getColumnName();

    final Timer timer = new Timer();
    final BookingView view = new BookingView(timer);
    final TimerTask task = ...;

    timer.scheduleAtFixedRate(task, 0, UPDATE_EVERY);
    return view;
  }
}

final TimerTask task = new TimerTask() {
  @Override
  public void run() {
    boolean first = true;

    // Make sure no previous task is already inside this block.
    if (working.compareAndSet(false, true)) {
      try {

        // Loop until no events was merged 
        // (the database is up to date).
        while (true) {

          // Get a list of up to 25 events that has not yet 
          // been merged into the materialized object view.
          final List added = unmodifiableList(
            mgr.stream()
              .filter(Booking.ID.greaterThan(last.get()))
              .sorted(Booking.ID.comparator())
              .limit(MAX_BATCH_SIZE)
              .collect(toList())
            );

          if (added.isEmpty()) {
            if (!first) {
              System.out.format(
                "%s: View is up to date. A total of " + 
                "%d rows have been loaded.%n",
                System.identityHashCode(last),
                total.get()
              );
            }

            break;
          } else {
            final Booking lastEntity = 
              added.get(added.size() - 1);

            last.set(lastEntity.getId());
            added.forEach(view::accept);
            total.addAndGet(added.size());

            System.out.format(
              "%s: Downloaded %d row(s) from %s. " + 
              "Latest %s: %d.%n", 
              System.identityHashCode(last),
              added.size(),
              table,
              field,
              Long.parseLong("" + last.get())
            );
          }

          first = false;
        }

        // Release this resource once we exit this block.
      } finally {
        working.set(false);
      }
    }
  }
};

private final static int MAX_BATCH_SIZE = 25;
private final static int UPDATE_EVERY   = 1_000; // Milliseconds

private final Timer timer;
private final Map<Long, Booking> bookings;

private BookingView(Timer timer) {
  this.timer    = requireNonNull(timer);
  this.bookings = new ConcurrentHashMap<>();
}

public Stream<Booking> stream() {
  return bookings.values().stream();
}

public void stop() {
  timer.cancel();
}

private boolean accept(Booking ev) {
    final String type = ev.getEventType();

    // If this was a creation event
    switch (type) {
        case "CREATE" :
            // Creation events must contain all information.
            if (!ev.getSauna().isPresent()
            ||  !ev.getTenant().isPresent()
            ||  !ev.getBookedFrom().isPresent()
            ||  !ev.getBookedTo().isPresent()
            ||  !checkIfAllowed(ev)) {
                return false;
            }

            // If something is already mapped to that key, refuse the 
            // event.
            return bookings.putIfAbsent(ev.getBookingId(), ev) == null;

        case "UPDATE" :
            // Create a copy of the current state
            final Booking existing = bookings.get(ev.getBookingId());

            // If the specified key did not exist, refuse the event.
            if (existing != null) {
                final Booking proposed = new BookingImpl();
                proposed.setId(existing.getId());

                // Update non-null values
                proposed.setSauna(ev.getSauna().orElse(
                    unwrap(existing.getSauna())
                ));
                proposed.setTenant(ev.getTenant().orElse(
                    unwrap(existing.getTenant())
                ));
                proposed.setBookedFrom(ev.getBookedFrom().orElse(
                    unwrap(existing.getBookedFrom())
                ));
                proposed.setBookedTo(ev.getBookedTo().orElse(
                    unwrap(existing.getBookedTo())
                ));

                // Make sure these changes are allowed.
                if (checkIfAllowed(proposed)) {
                    bookings.put(ev.getBookingId(), proposed);
                    return true;
                }
            }

            return false;


        case "DELETE" :
            // Remove the event if it exists, else refuse the event.
            return bookings.remove(ev.getBookingId()) != null;

        default :
            System.out.format(
                "Unexpected type '%s' was refused.%n", type);
            return false;
    }
}

public static void main(String... params) {
  final SaunaApplication app = new SaunaApplicationBuilder()
    .withPassword("password")
    .build();

  final BookingManager bookings = 
    app.getOrThrow(BookingManager.class);

  final SecureRandom rand = new SecureRandom();
  rand.setSeed(System.currentTimeMillis());

  // Insert three new bookings into the system.
  bookings.persist(
    new BookingImpl()
      .setBookingId(rand.nextLong())
      .setEventType("CREATE")
      .setSauna(1)
      .setTenant(1)
      .setBookedFrom(Date.valueOf(LocalDate.now().plus(3, DAYS)))
      .setBookedTo(Date.valueOf(LocalDate.now().plus(5, DAYS)))
  );

  bookings.persist(
    new BookingImpl()
      .setBookingId(rand.nextLong())
      .setEventType("CREATE")
      .setSauna(1)
      .setTenant(2)
      .setBookedFrom(Date.valueOf(LocalDate.now().plus(1, DAYS)))
      .setBookedTo(Date.valueOf(LocalDate.now().plus(2, DAYS)))
  );

  bookings.persist(
    new BookingImpl()
      .setBookingId(rand.nextLong())
      .setEventType("CREATE")
      .setSauna(1)
      .setTenant(3)
      .setBookedFrom(Date.valueOf(LocalDate.now().plus(2, DAYS)))
      .setBookedTo(Date.valueOf(LocalDate.now().plus(7, DAYS)))
  );

  final BookingView view = BookingView.create(bookings);

  // Wait until the view is up-to-date.
  try { Thread.sleep(5_000); }
  catch (final InterruptedException ex) {
    throw new RuntimeException(ex);
  }

  System.out.println("Current Bookings for Sauna 1:");
  final SimpleDateFormat dt = new SimpleDateFormat("yyyy-MM-dd");
  final Date now = Date.valueOf(LocalDate.now());
  view.stream()
    .filter(Booking.SAUNA.equal(1))
    .filter(Booking.BOOKED_TO.greaterOrEqual(now))
    .sorted(Booking.BOOKED_FROM.comparator())
    .map(b -> String.format(
      "Booked from %s to %s by Tenant %d.", 
      dt.format(b.getBookedFrom().get()),
      dt.format(b.getBookedTo().get()),
      b.getTenant().getAsInt()
    ))
    .forEachOrdered(System.out::println);

  System.out.println("No more bookings!");
  view.stop();
}