实践中的事件源和CQRS

1
2
3
4
5
6
7
8
9
10
11
CREATE DATABASE `sauna`;
CREATE TABLE `sauna`.`booking` (
`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
`booking_id` BIGINT NOT NULL,
`event_type` ENUM('CREATE', 'UPDATE', 'DELETE') NOT NULL,
`tenant` INT NULL,
`sauna` INT NULL,
`booked_from` DATE NULL,
`booked_to` DATE NULL,
PRIMARY KEY (`id`)
);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<speedment.version>3.0.0-EA2</speedment.version>
<mysql.version>5.1.39</mysql.version>
</properties>
<build>
<plugins>
<plugin>
<groupId>com.speedment</groupId>
<artifactId>speedment-maven-plugin</artifactId>
<version>${speedment.version}</version>
<dependencies>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>${mysql.version}</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>${mysql.version}</version>
</dependency>
<dependency>
<groupId>com.speedment</groupId>
<artifactId>runtime</artifactId>
<version>${speedment.version}</version>
<type>pom</type>
</dependency>
</dependencies>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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;
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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);
}
}
}
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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;
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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();
}

1
2
3
4
5
6
677772350: Downloaded 3 row(s) from booking. Latest id: 3.
677772350: View is up to date. A total of 3 rows have been loaded.
Current Bookings for Sauna 1:
Booked from 2016-10-11 to 2016-10-12 by Tenant 2.
Booked from 2016-10-13 to 2016-10-15 by Tenant 1.
No more bookings!