Added sample of stream transformations and README for events versioning samples

Author: oskardudycz

README.md

@@ -25,6 +25,7 @@ Tutorial, practical samples and other resources about Event Sourcing in JVM. See
       - [Main assumptions](#main-assumptions)
       - [Prerequisites](#prerequisites)
       - [Tools used](#tools-used)
+    - [Event Versioning](#event-versioning)
   - [Articles](#articles)
 
 ## Event Sourcing
@@ -267,6 +268,21 @@ For running the Event Store examples you need to have:
 2. [PostgreSQL](https://www.postgresql.org/) - Read Models
 3. [Spring Boot](https://spring.io/projects/spring-boot) - Web Application framework
 
+### [Event Versioning](./samples/events-versioning)
+Shows how to handle basic event schema versioning scenarios using event and stream transformations (e.g. upcasting):
+- [Simple mapping](./samples/events-versioning/#simple-mapping)
+  - [New not required property](./samples/events-versioning/#new-not-required-property)
+  - [New required property](./samples/events-versioning/#new-required-property)
+  - [Renamed property](./samples/events-versioning/#renamed-property)
+- [Upcasting](./samples/events-versioning/#upcasting)
+  - [Changed Structure](./samples/events-versioning/#changed-structure)
+  - [New required property](./samples/events-versioning/#new-required-property-1)
+- [Downcasters](./samples/events-versioning/#downcasters)
+- [Events Transformations](./samples/events-versioning/#events-transformations)
+- [Stream Transformation](./samples/events-versioning/#stream-transformation)
+- [Summary](./samples/events-versioning/#summary)
+- 📝 [Simple patterns for events schema versioning](https://event-driven.io/en/simple_events_versioning_patterns/?utm_source=event_sourcing_jvm_) 
+
 ## Articles
 
 Read also more on the **Event Sourcing** and **CQRS** topics in my [blog](https://event-driven.io/?utm_source=event_sourcing_jvm) posts:

samples/events-versioning/README.md

@@ -3,4 +3,4 @@
 public record PricedProductItem(
   ProductItem productItem,
   double unitPrice
-) {};
+) {}

samples/events-versioning/assets/migration.png

@@ -15,7 +15,7 @@ public record ShoppingCartOpened(
     UUID clientId,
     // Adding new not required property
     LocalDateTime initializedAt
-  ) { };
+  ) { }
 
   @Test
   public void Should_BeForwardCompatible()

samples/events-versioning/assets/stream-transformation.png

@@ -12,8 +12,7 @@
 import java.util.function.Function;
 
 import static java.util.stream.Collectors.toMap;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
 
 public class MultipleTransformationsWithDifferentEventTypesTests {
   public record Client(

samples/events-versioning/src/main/java/io/eventdriven/eventsversioning/v1/productitems/PricedProductItem.java

@@ -0,0 +1,298 @@
+package io.eventdriven.eventsversioning.transformations;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.eventdriven.eventsversioning.serialization.Serializer;
+import io.eventdriven.eventsversioning.v1.ShoppingCartEvent;
+import io.eventdriven.eventsversioning.v1.productitems.PricedProductItem;
+import io.eventdriven.eventsversioning.v1.productitems.ProductItem;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+public class StreamTransformationsTests {
+  public record ShoppingCartInitializedWithProducts(
+    UUID shoppingCartId,
+    UUID clientId,
+    List<PricedProductItem> productItems
+  ) {
+  }
+
+  public record EventMetadata(
+    UUID correlationId
+  ) {
+  }
+
+  public record EventData(
+    String eventType,
+    byte[] data,
+    byte[] metaData
+  ) {
+  }
+
+  public static List<EventData> flattenInitializedEventsWithProductItemsAdded(List<EventData> events) {
+    var cartOpened = events.get(0);
+    var cartInitializedCorrelationId =
+      Serializer.deserialize(EventMetadata.class, cartOpened.metaData)
+        .orElseThrow(() -> new RuntimeException("Error deserializing metadata"))
+        .correlationId;
+
+    var i = 1;
+    var productItemsAdded = new ArrayList<EventData>();
+
+    while (i < events.size()) {
+      var eventData = events.get(i);
+
+      if (!eventData.eventType().equals("product_item_added_v1"))
+        break;
+
+      var correlationId =
+        Serializer.deserialize(EventMetadata.class, eventData.metaData)
+          .orElseThrow(() -> new RuntimeException("Error deserializing metadata"))
+          .correlationId;
+
+      if (!correlationId.equals(cartInitializedCorrelationId))
+        break;
+
+      productItemsAdded.add(eventData);
+      i++;
+    }
+
+    var mergedEvent = toShoppingCartInitializedWithProducts(
+      cartOpened,
+      productItemsAdded
+    );
+
+    return Stream.concat(
+      Stream.of(mergedEvent),
+      events.stream().skip(i)
+    ).toList();
+  }
+
+  private static EventData toShoppingCartInitializedWithProducts(
+    EventData shoppingCartInitialized,
+    List<EventData> productItemsAdded
+  ) {
+    var shoppingCartInitializedJson =
+      Serializer.deserialize(shoppingCartInitialized.data);
+
+    var newEvent = new ShoppingCartInitializedWithProducts(
+      UUID.fromString(shoppingCartInitializedJson.at("/shoppingCartId").asText()),
+      UUID.fromString(shoppingCartInitializedJson.at("/clientId").asText()),
+      productItemsAdded.stream()
+        .map(pi -> {
+          var pricedProductItem = Serializer.deserialize(pi.data);
+
+          return new PricedProductItem(
+            new ProductItem(
+              UUID.fromString(pricedProductItem.at("/productItem/productItem/productId").asText()),
+              pricedProductItem.at("/productItem/productItem/quantity").asInt()
+            ),
+            pricedProductItem.at("/productItem/unitPrice").asDouble()
+          );
+        }).toList()
+    );
+
+    return new EventData(
+      "shopping_cart_initialized_v2",
+      Serializer.serialize(newEvent),
+      shoppingCartInitialized.metaData
+    );
+  }
+
+  public class StreamTransformations {
+    private final List<Function<List<EventData>, List<EventData>>> jsonTransformations = new ArrayList<>();
+
+    public List<EventData> transform(List<EventData> events) {
+      if (jsonTransformations.isEmpty())
+        return events;
+
+      var result = events;
+
+      for (var transform : jsonTransformations) {
+        result = transform.apply(result);
+      }
+      return result;
+    }
+
+    public StreamTransformations register(Function<List<EventData>, List<EventData>> transformJson) {
+      jsonTransformations.add(transformJson);
+      return this;
+    }
+  }
+
+  public class EventTransformations {
+    private final Map<String, Function<byte[], Object>> jsonTransformations = new HashMap<>();
+
+    public Optional<Object> tryTransform(String eventTypeName, byte[] json) {
+      if (!jsonTransformations.containsKey(eventTypeName)) {
+        return Optional.empty();
+      }
+
+      var transformJson = jsonTransformations.get(eventTypeName);
+
+      return Optional.of(transformJson.apply(json));
+    }
+
+    public <Event> EventTransformations register(
+      String eventTypeName,
+      Function<JsonNode, Event> transformJson
+    ) {
+      jsonTransformations.put(
+        eventTypeName,
+        json -> transformJson.apply(Serializer.deserialize(json))
+      );
+      return this;
+    }
+
+    public <OldEvent, Event> EventTransformations register(
+      Class<OldEvent> oldEventClass,
+      String eventTypeName,
+      Function<OldEvent, Event> transformEvent
+    ) {
+      jsonTransformations.put(
+        eventTypeName,
+        json -> transformEvent.apply(Serializer.deserialize(oldEventClass, json)
+          .orElseThrow(() -> new RuntimeException("Error deserializing")))
+      );
+      return this;
+    }
+  }
+
+  public class EventTypeMapping {
+    private final Map<String, Class> mappings = new HashMap<>();
+
+    public <Event> EventTypeMapping register(Class<Event> eventClass, String... typeNames) {
+      for (var typeName : typeNames) {
+        mappings.put(typeName, eventClass);
+      }
+
+      return this;
+    }
+
+    public Class map(String eventType) {
+      return mappings.get(eventType);
+    }
+  }
+
+  public record EventSerializer(
+    EventTypeMapping mapping,
+    StreamTransformations streamTransformations,
+    EventTransformations transformations
+  ) {
+    public Optional<Object> deserialize(String eventTypeName, byte[] json) {
+      return transformations.tryTransform(eventTypeName, json)
+        .or(() -> Serializer.deserialize(mapping().map(eventTypeName), json));
+    }
+
+    public List<Optional<Object>> deserialize(List<EventData> events) {
+      return streamTransformations.transform(events).stream()
+        .map(event -> deserialize(event.eventType, event.data))
+        .toList();
+    }
+  }
+
+  public record EventEnvelope(
+    String type,
+    Object data,
+    EventMetadata metadata
+  ) {
+  }
+
+  @Test
+  public void UpcastObjects_Should_BeForwardCompatible() {
+    // Given
+    var mapping = new EventTypeMapping()
+      .register(ShoppingCartInitializedWithProducts.class, "shopping_cart_initialized_v2")
+      .register(ShoppingCartEvent.ProductItemAddedToShoppingCart.class, "product_item_added_v1")
+      .register(ShoppingCartEvent.ShoppingCartConfirmed.class, "shopping_card_confirmed_v1");
+
+    var streamTransformations =
+      new StreamTransformations()
+        .register(StreamTransformationsTests::flattenInitializedEventsWithProductItemsAdded);
+
+    var serializer = new EventSerializer(
+      mapping,
+      streamTransformations,
+      new EventTransformations()
+    );
+
+    var shoppingCardId = UUID.randomUUID();
+    var clientId = UUID.randomUUID();
+    var theSameCorrelationId = UUID.randomUUID();
+    var productItem = new PricedProductItem(new ProductItem(UUID.randomUUID(), 1), 23.22);
+
+    var events = new ArrayList<>(List.of(
+      new EventEnvelope(
+        "shopping_cart_initialized_v1",
+        new ShoppingCartEvent.ShoppingCartOpened(shoppingCardId, clientId),
+        new EventMetadata(theSameCorrelationId)
+      ),
+      new EventEnvelope(
+        "product_item_added_v1",
+        new ShoppingCartEvent.ProductItemAddedToShoppingCart(shoppingCardId, productItem),
+        new EventMetadata(theSameCorrelationId)
+      ),
+      new EventEnvelope(
+        "product_item_added_v1",
+        new ShoppingCartEvent.ProductItemAddedToShoppingCart(shoppingCardId, productItem),
+        new EventMetadata(theSameCorrelationId)
+      ),
+      new EventEnvelope(
+        "product_item_added_v1",
+        new ShoppingCartEvent.ProductItemAddedToShoppingCart(shoppingCardId, productItem),
+        new EventMetadata(UUID.randomUUID())
+      ),
+      new EventEnvelope(
+        "shopping_card_confirmed_v1",
+        new ShoppingCartEvent.ShoppingCartConfirmed(shoppingCardId, LocalDateTime.now()),
+        new EventMetadata(UUID.randomUUID())
+      )
+    ));
+
+    var serialisedEvents = events.stream()
+      .map(e ->
+        new EventData(
+          e.type,
+          Serializer.serialize(e.data()),
+          Serializer.serialize(e.metadata)
+        )
+      ).toList();
+
+    // When
+    var deserializedEvents = serializer.deserialize(serialisedEvents).stream()
+      .filter(Optional::isPresent)
+      .map(Optional::get)
+      .toList();
+
+    // Then
+    assertEquals(3, deserializedEvents.size());
+
+    assertInstanceOf(ShoppingCartInitializedWithProducts.class, deserializedEvents.get(0));
+    var shoppingCartInitializedWithProducts = (ShoppingCartInitializedWithProducts)deserializedEvents.get(0);
+
+    assertEquals(shoppingCardId, shoppingCartInitializedWithProducts.shoppingCartId());
+    assertEquals(clientId, shoppingCartInitializedWithProducts.clientId());
+    assertEquals(2, shoppingCartInitializedWithProducts.productItems().size());
+    assertEquals(productItem, shoppingCartInitializedWithProducts.productItems().get(0));
+    assertEquals(productItem, shoppingCartInitializedWithProducts.productItems().get(1));
+
+
+    assertInstanceOf(ShoppingCartEvent.ProductItemAddedToShoppingCart.class, deserializedEvents.get(1));
+    var productItemAddedToShoppingCart = (ShoppingCartEvent.ProductItemAddedToShoppingCart)deserializedEvents.get(1);
+
+    assertEquals(shoppingCardId, productItemAddedToShoppingCart.shoppingCartId());
+    assertEquals(productItem, productItemAddedToShoppingCart.productItem());
+
+
+    assertInstanceOf(ShoppingCartEvent.ShoppingCartConfirmed.class, deserializedEvents.get(2));
+    var shoppingCartConfirmed = (ShoppingCartEvent.ShoppingCartConfirmed)deserializedEvents.get(2);
+
+    assertEquals(shoppingCardId, shoppingCartConfirmed.shoppingCartId());
+  }
+}