fix: align the event bus contract with the code and extend the demo

publish rejects null, exact-class dispatch is documented, the sequence counters are plain longs, and the demo shows a credit limit change flowing through the replica. The class diagram is a rendered PNG.

Author: ylcn91

event-carried-state-transfer/README.md

@@ -57,6 +57,8 @@ sequenceDiagram
     OS->>R: apply (version 1 <= 2, ignored)
 ```
 
+![Event-Carried State Transfer class diagram](./etc/event-carried-state-transfer.urm.png)
+
 ## Programmatic Example of Event-Carried State Transfer Pattern in Java
 
 The example has a producer, a channel and a consumer. The producer is the customer service, the channel is a tiny in-memory event bus and the consumer is the order service with its local customer replica.
@@ -116,7 +118,7 @@ public CustomerState changeShippingAddress(String customerId, String newAddress)
 
 private CustomerState store(CustomerState state) {
   customers.put(state.customerId(), state);
-  var event = new CustomerUpdatedEvent(eventSequence.incrementAndGet(), Instant.now(), state);
+  var event = new CustomerUpdatedEvent(++eventSequence, Instant.now(), state);
   bus.publish(event);
   return state;
 }
@@ -166,13 +168,13 @@ public Order placeOrder(String customerId, BigDecimal amount) {
         "Amount " + amount + " exceeds credit limit " + customer.creditLimit() + " of " + customerId);
   }
   return new Order(
-      "ORD-" + orderSequence.incrementAndGet(), customerId, customer.shippingAddress(), amount);
+      "ORD-" + ++orderSequence, customerId, customer.shippingAddress(), amount);
 }
 ```
 
 6. **The demo**
 
-`App` registers a customer and changes the address, takes the customer service offline and places an order from the replica, publishes a stale event that the replica ignores, and finally shows the replica enforcing the credit limit.
+`App` registers a customer and changes the address, takes the customer service offline and places an order from the replica, publishes a stale event that the replica ignores, and shows the replica enforcing the credit limit. It then brings the customer service back and raises that limit: the new limit travels inside the event, and the order that was just rejected is accepted from the replica alone, without a single call back to the producer.
 
 ```java
 var bus = new EventBus();
@@ -188,6 +190,10 @@ var order = orderService.placeOrder("C-1", new BigDecimal("120.00")); // succeed
 
 bus.publish(new CustomerUpdatedEvent(99, Instant.now(), staleVersionOne)); // ignored
 tryToOrder(orderService, "C-1", new BigDecimal("900.00")); // rejected, above the replicated limit
+
+customerService.restart();
+customerService.changeCreditLimit("C-1", new BigDecimal("1500.00")); // the event carries the new limit
+tryToOrder(orderService, "C-1", new BigDecimal("900.00")); // accepted now, still only from the replica
 ```
 
 Program output:
@@ -217,12 +223,17 @@ INFO App -- Order service replica: C-1 version 2 at '42 Ocean Avenue, Porto' wit
 INFO App -- --- Step 4: the replica is enough to enforce business rules ---
 WARN App -- Order rejected: Amount 900.00 exceeds credit limit 500.00 of C-1
 WARN App -- Order rejected: Unknown customer C-2 in replica
+INFO App -- --- Step 5: a new credit limit travels in the event and unblocks the rejected order ---
+INFO CustomerService -- Customer service is back online
+INFO CustomerService -- Customer C-1 gets a credit limit of 1500.00
+INFO CustomerService -- Publishing event 3 with the full state of C-1 (version 3)
+INFO EventBus -- Publishing CustomerUpdatedEvent to 1 subscriber(s)
+INFO CustomerReplica -- Replica updated from event 3: C-1 is now at version 3 with address '42 Ocean Avenue, Porto' and limit 1500.00
+INFO App -- Order service replica: C-1 version 3 at '42 Ocean Avenue, Porto' with limit 1500.00
+INFO OrderService -- Accepted ORD-2 for C-1 (900.00) shipping to '42 Ocean Avenue, Porto' using replica version 3
+INFO App -- Order ORD-2 accepted
 ```
 
-## Class diagram
-
-See [event-carried-state-transfer.urm.puml](./etc/event-carried-state-transfer.urm.puml) for the PlantUML class diagram.
-
 ## When to Use the Event-Carried State Transfer Pattern in Java
 
 * Consumers need data owned by another service on every request and a synchronous call would add latency, load, or a hard availability dependency.

event-carried-state-transfer/etc/event-carried-state-transfer.urm.png

@@ -36,14 +36,15 @@ package com.iluwatar.eventcarriedstatetransfer {
   class CustomerService {
     - customers : Map<String, CustomerState>
     - bus : EventBus
-    - eventSequence : AtomicLong
+    - eventSequence : long
     - online : boolean
     + CustomerService(bus : EventBus)
     + register(customerId : String, name : String, shippingAddress : String, creditLimit : BigDecimal) : CustomerState
     + changeShippingAddress(customerId : String, newAddress : String) : CustomerState
     + changeCreditLimit(customerId : String, newLimit : BigDecimal) : CustomerState
     + findCustomer(customerId : String) : Optional<CustomerState>
     + shutdown() : void
+    + restart() : void
     + isOnline() : boolean
   }
   class CustomerReplica {
@@ -55,7 +56,7 @@ package com.iluwatar.eventcarriedstatetransfer {
   }
   class OrderService {
     - replica : CustomerReplica
-    - orderSequence : AtomicLong
+    - orderSequence : long
     + OrderService(bus : EventBus)
     + placeOrder(customerId : String, amount : BigDecimal) : Order
     + replica() : CustomerReplica
@@ -86,8 +87,8 @@ CustomerService --> "*" CustomerState
 CustomerService ..> CustomerUpdatedEvent : publishes
 CustomerReplica --> "*" CustomerState
 CustomerReplica ..> CustomerUpdatedEvent : applies
-CustomerReplica ..|> EventListener
 OrderService --> CustomerReplica
+OrderService ..> EventListener : adapts the replica
 OrderService ..> EventBus : subscribes
 OrderService ..> Order : creates
 OrderService ..> OrderRejectedException : throws

event-carried-state-transfer/etc/event-carried-state-transfer.urm.puml

@@ -41,8 +41,10 @@
  *
  * <p>The demo registers a customer and changes the address, showing the replica following each
  * event. It then takes the customer service offline and places an order anyway, purely from the
- * replica. A stale event is published to show that the replica ignores it, and finally an order
- * above the replicated credit limit is rejected.
+ * replica. A stale event is published to show that the replica ignores it, and an order above the
+ * replicated credit limit is rejected. Finally the customer service comes back and raises that
+ * credit limit: the new state travels in the event, and the order that was just rejected is
+ * accepted from the replica alone.
  */
 @Slf4j
 public class App {
@@ -86,6 +88,13 @@ public static void main(String[] args) {
     LOGGER.info("--- Step 4: the replica is enough to enforce business rules ---");
     tryToOrder(orderService, "C-1", new BigDecimal("900.00"));
     tryToOrder(orderService, "C-2", new BigDecimal("10.00"));
+
+    LOGGER.info(
+        "--- Step 5: a new credit limit travels in the event and unblocks the rejected order ---");
+    customerService.restart();
+    customerService.changeCreditLimit("C-1", new BigDecimal("1500.00"));
+    logReplica(orderService, "C-1");
+    tryToOrder(orderService, "C-1", new BigDecimal("900.00"));
   }
 
   /**

event-carried-state-transfer/src/main/java/com/iluwatar/eventcarriedstatetransfer/App.java

@@ -24,7 +24,7 @@
  */
 package com.iluwatar.eventcarriedstatetransfer;
 
-import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
@@ -35,11 +35,13 @@
  * <p>Because each event carries the full state, applying one is a simple upsert. The version
  * carried by the state guards against events that arrive late or twice: an event is ignored unless
  * its version is newer than what the replica already holds.
+ *
+ * <p>Not thread-safe; the demo drives it from a single thread.
  */
 @Slf4j
 public class CustomerReplica {
 
-  private final Map<String, CustomerState> customers = new HashMap<>();
+  private final Map<String, CustomerState> customers = new LinkedHashMap<>();
 
   /**
    * Applies an event to the replica.

event-carried-state-transfer/src/main/java/com/iluwatar/eventcarriedstatetransfer/CustomerReplica.java

@@ -29,7 +29,6 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.atomic.AtomicLong;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -39,13 +38,15 @@
  * {@link CustomerUpdatedEvent} that carries the customer's complete new state. Consumers never need
  * to query this service to act on the change, which is demonstrated by taking it offline in the
  * demo while orders keep flowing.
+ *
+ * <p>Not thread-safe; the demo drives it from a single thread.
  */
 @Slf4j
 public class CustomerService {
 
   private final Map<String, CustomerState> customers = new LinkedHashMap<>();
   private final EventBus bus;
-  private final AtomicLong eventSequence = new AtomicLong();
+  private long eventSequence;
   private boolean online = true;
 
   /**
@@ -118,6 +119,12 @@ public void shutdown() {
     LOGGER.warn("Customer service is going offline");
   }
 
+  /** Ends the outage: direct queries are answered again. */
+  public void restart() {
+    online = true;
+    LOGGER.info("Customer service is back online");
+  }
+
   /** Whether direct queries are currently answered. */
   public boolean isOnline() {
     return online;
@@ -133,7 +140,7 @@ private CustomerState existing(String customerId) {
 
   private CustomerState store(CustomerState state) {
     customers.put(state.customerId(), state);
-    var event = new CustomerUpdatedEvent(eventSequence.incrementAndGet(), Instant.now(), state);
+    var event = new CustomerUpdatedEvent(++eventSequence, Instant.now(), state);
     LOGGER.info(
         "Publishing event {} with the full state of {} (version {})",
         event.eventId(),

event-carried-state-transfer/src/main/java/com/iluwatar/eventcarriedstatetransfer/CustomerService.java

@@ -28,6 +28,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -45,6 +46,9 @@ public class EventBus {
   /**
    * Registers a listener for events of the given class.
    *
+   * <p>Dispatch is by exact runtime class, so the listener receives only events whose class is
+   * exactly {@code eventType}, never those of a subclass.
+   *
    * @param eventType the class of events to receive
    * @param listener the listener to notify
    * @param <E> the event type
@@ -58,9 +62,14 @@ public <E> void subscribe(Class<E> eventType, EventListener<? super E> listener)
   /**
    * Delivers the event to every listener subscribed to its class.
    *
+   * <p>Only listeners subscribed to the event's exact runtime class are notified; a listener
+   * subscribed to a supertype of the event does not receive it.
+   *
    * @param event the event to publish
+   * @throws NullPointerException if the event is {@code null}
    */
   public void publish(Object event) {
+    Objects.requireNonNull(event, "event");
     var subscribers = listeners.getOrDefault(event.getClass(), List.of());
     if (subscribers.isEmpty()) {
       LOGGER.warn("No subscribers for {}", event.getClass().getSimpleName());

event-carried-state-transfer/src/main/java/com/iluwatar/eventcarriedstatetransfer/EventBus.java

@@ -25,7 +25,6 @@
 package com.iluwatar.eventcarriedstatetransfer;
 
 import java.math.BigDecimal;
-import java.util.concurrent.atomic.AtomicLong;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -34,12 +33,14 @@
  * <p>The order service subscribes its {@link CustomerReplica} to customer events and afterwards
  * answers every order using only that replica. It has no reference to the customer service at all,
  * so it keeps working while the customer service is down and it never adds load to it.
+ *
+ * <p>Not thread-safe; the demo drives it from a single thread.
  */
 @Slf4j
 public class OrderService {
 
   private final CustomerReplica replica = new CustomerReplica();
-  private final AtomicLong orderSequence = new AtomicLong();
+  private long orderSequence;
 
   /**
    * Creates the service and subscribes its replica to customer events.
@@ -74,12 +75,7 @@ public Order placeOrder(String customerId, BigDecimal amount) {
               + " of "
               + customerId);
     }
-    var order =
-        new Order(
-            "ORD-" + orderSequence.incrementAndGet(),
-            customerId,
-            customer.shippingAddress(),
-            amount);
+    var order = new Order("ORD-" + ++orderSequence, customerId, customer.shippingAddress(), amount);
     LOGGER.info(
         "Accepted {} for {} ({}) shipping to '{}' using replica version {}",
         order.orderId(),

event-carried-state-transfer/src/main/java/com/iluwatar/eventcarriedstatetransfer/OrderService.java

@@ -100,4 +100,15 @@ void refusesDirectLookupsWhenOffline() {
     var thrown = assertThrows(IllegalStateException.class, () -> service.findCustomer("C-1"));
     assertEquals("customer service is offline", thrown.getMessage());
   }
+
+  @Test
+  void answersDirectLookupsAgainAfterARestart() {
+    service.register("C-1", "Alice", "Lisbon", new BigDecimal("500.00"));
+    service.shutdown();
+
+    service.restart();
+
+    assertTrue(service.isOnline());
+    assertEquals("Alice", service.findCustomer("C-1").orElseThrow().name());
+  }
 }

event-carried-state-transfer/src/test/java/com/iluwatar/eventcarriedstatetransfer/CustomerServiceTest.java

@@ -26,6 +26,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.ArrayList;
@@ -62,6 +63,13 @@ void publishingWithoutSubscribersIsHarmless() {
     assertDoesNotThrow(() -> bus.publish("nobody listens"));
   }
 
+  @Test
+  void rejectsNullEvents() {
+    var exception = assertThrows(NullPointerException.class, () -> bus.publish(null));
+
+    assertEquals("event", exception.getMessage());
+  }
+
   @Test
   void deliversInSubscriptionOrder() {
     var order = new ArrayList<String>();