Upgrade to Pro — share decks privately, control downloads, hide ads and more …

UPDATED: The Building Blocks of an Event Driven...

UPDATED: The Building Blocks of an Event Driven CQRS Application in Scala

Talk given at Amsterdam.scala. Content similar to what I presented at Scala.World unconference, but with much more diagrams and animations.

Renato Cavalcanti

October 04, 2015
Tweet

More Decks by Renato Cavalcanti

Other Decks in Technology

Transcript

  1. Agenda • DDD / CQRS / Event Sourcing (intro) •

    Aggregates / Commands / Events in Scala • Type members: type projections • Partial Functions, lifting and function composition • Akka and asynchrounous programming • Akka Persistence and Event Sourcing
  2. DDD - Aggregate • Aggregate is a central DDD concept.

    • It has a root and zero or more entities and value objects underneath.
  3. DDD - Aggregate • Aggregate is a central DDD concept.

    • It has a root and zero or more entities and value objects underneath. • Responsible for the consistency of underneath objects.
  4. DDD - Aggregate • Aggregate is a central DDD concept.

    • It has a root and zero or more entities and value objects underneath. • Responsible for the consistency of underneath objects. • You can only modify one Aggregate per transaction.
  5. CQRS - Event Sourcing • "It is simply the creation

    of two objects where there was previously only one" - Greg Young
  6. CQRS - Event Sourcing • "It is simply the creation

    of two objects where there was previously only one" - Greg Young • Write Model receives Commands and produces Events.
  7. CQRS - Event Sourcing • "It is simply the creation

    of two objects where there was previously only one" - Greg Young • Write Model receives Commands and produces Events. • Read Models are created from Events.
  8. CQRS - Event Sourcing • "It is simply the creation

    of two objects where there was previously only one" - Greg Young • Write Model receives Commands and produces Events. • Read Models are created from Events. • Events can be stored and replayed.
  9. Event Driven / Event Sourcing • CQRS is Event Driven,

    but not necessarily implements Event Sourcing
  10. Event Driven / Event Sourcing • CQRS is Event Driven,

    but not necessarily implements Event Sourcing • in synchronous CQRS:
  11. Event Driven / Event Sourcing • CQRS is Event Driven,

    but not necessarily implements Event Sourcing • in synchronous CQRS: •tx(Cmd 㱺 Write Model 㱺 Event 㱺 View)
  12. Event Driven / Event Sourcing • CQRS is Event Driven,

    but not necessarily implements Event Sourcing • in synchronous CQRS: •tx(Cmd 㱺 Write Model 㱺 Event 㱺 View) •tx(Cmd 㱺 Write Model 㱺 Event 㱺 
 View1, View2, View3, View4, …, ViewN)
  13. Event Driven / Event Sourcing • CQRS is Event Driven,

    but not necessarily implements Event Sourcing • in synchronous CQRS: •tx(Cmd 㱺 Write Model 㱺 Event 㱺 View) •tx(Cmd 㱺 Write Model 㱺 Event 㱺 
 View1, View2, View3, View4, …, ViewN) Won’t scale! Damm blocking!
  14. Event Driven / Event Sourcing • in asynchronous CQRS: •tx(Cmd

    㱺 Write Model 㱺 Event) •tx(Event 㱺 View1) •tx(Event 㱺 View2) •tx(Event 㱺 View3)
  15. Event Driven / Event Sourcing • in asynchronous CQRS: •tx(Cmd

    㱺 Write Model 㱺 Event) •tx(Event 㱺 View1) •tx(Event 㱺 View2) •tx(Event 㱺 View3) • Introduces Eventual Consistency
  16. • Aggregate responsible for consistency • You can only modify

    one Aggregate per transaction • Write Model receives Commands and produces Events • Read Models are created from Events • Events can be stored and replayed (when Event Sourcing) • Eventual Consistency (if async Write/Read models)
  17. Basic Operations (non-reactive) trait Cmd trait Event trait Aggregate //

    validate a Command and produces an Event def validate(cmd:Cmd): Event // apply creational Event and produces a new Aggregate def applyEvent(evt:Event): Aggregate // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // apply event on current Aggregate def applyEvent(agg:Aggregate, evt:Event): Aggregate
  18. Basic Operations (non-reactive) trait Cmd trait Event trait Aggregate //

    validate a Command and produces an Event def validate(cmd:Cmd): Event // apply creational Event and produces a new Aggregate def applyEvent(evt:Event): Aggregate // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // apply event on current Aggregate def applyEvent(agg:Aggregate, evt:Event): Aggregate
  19. Basic Operations (non-reactive) trait Cmd trait Event trait Aggregate //

    validate a Command and produces an Event def validate(cmd:Cmd): Event // apply creational Event and produces a new Aggregate def applyEvent(evt:Event): Aggregate // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // apply event on current Aggregate def applyEvent(agg:Aggregate, evt:Event): Aggregate
  20. Basic Operations (non-reactive) trait Cmd trait Event trait Aggregate //

    validate a Command and produces an Event def validate(cmd:Cmd): Event // apply creational Event and produces a new Aggregate def applyEvent(evt:Event): Aggregate // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // apply event on current Aggregate def applyEvent(agg:Aggregate, evt:Event): Aggregate
  21. E0 State of Aggregate Event Store // first command is

    validated def validate(cmd:Cmd): Event // CreateOrder => OrderCreated
  22. E0 State of Aggregate Event Store S0 // first command

    is validated def validate(cmd:Cmd): Event // CreateOrder => OrderCreated // apply creational Event and produces a new Aggregate def applyEvent(evt:Event): Aggregate // OrderCreated => Order
  23. E0 State of Aggregate Event Store E1 S0 // validate

    a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // (Order, AddProduct) => ProductAdded
  24. E0 State of Aggregate Event Store E1 S0 S1 //

    apply event on current Aggregate def applyEvent(agg:Aggregate, evt:Event): Aggregate // (Order, ProductAdded) => Order // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // (Order, AddProduct) => ProductAdded
  25. E0 State of Aggregate Event Store E1 E2 S0 S1

    // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // (Order, ExecuteOrder) => OrderExecute
  26. E0 State of Aggregate Event Store E1 E2 S0 S1

    S2 // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // (Order, ExecuteOrder) => OrderExecute // apply event on current Aggregate def applyEvent(agg:Aggregate, evt:Event): Aggregate // (Order, OrderExecuted) => Order
  27. E0 State of Aggregate Event Store E1 E2 S0 S1

    S2 // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // (Order, ExecuteOrder) => OrderExecute E3
  28. E0 State of Aggregate Event Store E1 E2 S0 S1

    S2 // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // (Order, ExecuteOrder) => OrderExecute E3 Invalid Command! Order is already executed!
  29. Watch out! We don’t throw exceptions in Scala. We’ll fix

    it soon! E0 State of Aggregate Event Store E1 E2 S0 S1 S2 // validate a Command against current Aggregate state def validate(agg:Aggregate, cmd:Cmd): Seq[Event] // (Order, ExecuteOrder) => OrderExecute E3 Invalid Command! Order is already executed!
  30. E0 State of Aggregate Event Store E1 E2 E0 S0

    // apply creational Event and produces a new Aggregate // OrderCreated => Order def applyEvent(evt:Event): Aggregate
  31. E0 State of Aggregate Event Store E1 E2 S0 E0

    S0 // apply creational Event and produces a new Aggregate // OrderCreated => Order def applyEvent(evt:Event): Aggregate
  32. E0 State of Aggregate Event Store E1 E2 S0 S0

    S1 // apply event on current Aggregate // (Order, ProductAdded) => Order def applyEvent(agg:Aggregate, evt:Event): Aggregate E1
  33. E0 State of Aggregate Event Store E1 E2 S0 S1

    S0 S1 // apply event on current Aggregate // (Order, ProductAdded) => Order def applyEvent(agg:Aggregate, evt:Event): Aggregate E1
  34. E0 State of Aggregate Event Store E1 E2 S0 S1

    E2 S1 S2 // apply event on current Aggregate // (Order, OrderExecuted) => Order def applyEvent(agg:Aggregate, evt:Event): Aggregate
  35. E0 State of Aggregate Event Store E1 E2 S0 S1

    S2 E2 S1 S2 // apply event on current Aggregate // (Order, OrderExecuted) => Order def applyEvent(agg:Aggregate, evt:Event): Aggregate
  36. CQRS using Scala std lib • DomainCommand and DomainEvent •

    A Protocol - message (cmd & evt) for a given Aggregate • A case class extending the Aggregate trait • A Behavior definition - Reactive
  37. import java.util.UUID trait DomainCommand { val id: CommandId = CommandId()

    } case class CommandId(value: UUID = UUID.randomUUID()) trait DomainEvent { def id: EventId } case class EventId(value: UUID = UUID.randomUUID())
  38. object ProductProtocol extends ProtocolDef { sealed trait ProductCommand extends ProtocolCommand

    sealed trait ProductEvent extends ProtocolEvent } trait ProtocolDef { trait ProtocolCommand extends DomainCommand trait ProtocolEvent extends DomainEvent }
  39. object ProductProtocol extends ProtocolDef { sealed trait ProductCommand extends ProtocolCommand

    sealed trait ProductEvent extends ProtocolEvent case class CreateProduct(name: String, description: String, price: Double) extends ProductCommand case class ProductCreated(name: String, description: String, price: Double, id: EventId = EventId()) extends ProductEvent } trait ProtocolDef { trait ProtocolCommand extends DomainCommand trait ProtocolEvent extends DomainEvent }
  40. trait ProtocolDef { trait ProtocolCommand extends DomainCommand trait ProtocolEvent extends

    DomainEvent } object ProductProtocol extends ProtocolDef { sealed trait ProductCommand extends ProtocolCommand sealed trait ProductEvent extends ProtocolEvent case class CreateProduct(name: String, description: String, price: Double) extends ProductCommand case class ProductCreated(name: String, description: String, price: Double, id: EventId = EventId()) extends ProductEvent case class ChangePrice(price: Double) extends ProductCommand case class PriceChanged(newPrice: Double, id: EventId = EventId()) extends ProductEvent }
  41. trait AggregateID { def value: String } trait Aggregate {

    type Id <: AggregateID type Protocol <: ProtocolDef def id: Id } case class ProductNumber(value: String) extends AggregateID case class Product(name: String, description: String, price: Double, id: ProductNumber) extends Aggregate { type Id = ProductNumber type Protocol = ProductProtocol.type }
  42. trait Behavior[A <: Aggregate] { type AggregateType = A type

    Command = A#Protocol#ProtocolCommand type Event = A#Protocol#ProtocolEvent type Events = Seq[Event] def applyEvent(event: Event): AggregateType def applyEvent(event: Event, aggregate: AggregateType): AggregateType // async behavior protected def validateAsync(cmd: Command) (implicit ec: ExecutionContext): Future[Event] protected def validateAsync(cmd: Command, aggregate: AggregateType) (implicit ec: ExecutionContext): Future[Events] }
  43. trait Behavior[A <: Aggregate] { type AggregateType = A type

    Command = A#Protocol#ProtocolCommand type Event = A#Protocol#ProtocolEvent type Events = Seq[Event] def applyEvent(event: Event): AggregateType def applyEvent(event: Event, aggregate: AggregateType): AggregateType // async behavior protected def validateAsync(cmd: Command) (implicit ec: ExecutionContext): Future[Event] protected def validateAsync(cmd: Command, aggregate: AggregateType) (implicit ec: ExecutionContext): Future[Events] }
  44. trait Behavior[A <: Aggregate] { type AggregateType = A type

    Command = A#Protocol#ProtocolCommand type Event = A#Protocol#ProtocolEvent type Events = Seq[Event] def applyEvent(event: Event): AggregateType def applyEvent(event: Event, aggregate: AggregateType): AggregateType // async behavior protected def validateAsync(cmd: Command) (implicit ec: ExecutionContext): Future[Event] protected def validateAsync(cmd: Command, aggregate: AggregateType) (implicit ec: ExecutionContext): Future[Events] }
  45. trait Behavior[A <: Aggregate] { type AggregateType = A type

    Command = A#Protocol#ProtocolCommand type Event = A#Protocol#ProtocolEvent type Events = Seq[Event] def applyEvent(event: Event): AggregateType def applyEvent(event: Event, aggregate: AggregateType): AggregateType // async behavior protected def validateAsync(cmd: Command) (implicit ec: ExecutionContext): Future[Event] protected def validateAsync(cmd: Command, aggregate: AggregateType) (implicit ec: ExecutionContext): Future[Events] } Inconvenient to always have to work with Futures and Seqs
  46. // on-create Command => Event Command => Future[Event] // post-create

    (Command, AggregateType) => Event (Command, AggregateType) => Future[Event] (Command, AggregateType) => Seq[Event] (Command, AggregateType) => Future[Seq[Event]] We need some variations, for instance…
  47. // on-create Command => Event Command => Future[Event] // post-create

    (Command, AggregateType) => Event (Command, AggregateType) => Future[Event] (Command, AggregateType) => Seq[Event] (Command, AggregateType) => Future[Seq[Event]] We need some variations, for instance…
  48. PartialFunction[A, Option[B]] a1 a3 a2 Some(b1) Some(b2) Some(b3) domain =

    a1, a2 and a3 a4 None def fallback[A, B]: PartialFunction[A, Option[B]] = { case any => None }
  49. def fallback[A, B]: PartialFunction[A, Future[B]] = { case any =>

    case any => Future.failed(new IllegalArgumentException(s"Can NOT do anything useful with $any")) } PartialFunction[A, Future[B]] a1 a3 a2 Success(b1) Success(b2) Success(b3) domain = a1, a2 and a3 a4 Failure(e)
  50. Behavior DSL • PartialFunctions for all results we want to

    produce
 (Event, Future[Event], Seq[Event] and Future[Seq[Event]])
  51. Behavior DSL • PartialFunctions for all results we want to

    produce
 (Event, Future[Event], Seq[Event] and Future[Seq[Event]]) • lift them all to Future[Event] or Future[Seq[Event]]
  52. Behavior DSL • PartialFunctions for all results we want to

    produce
 (Event, Future[Event], Seq[Event] and Future[Seq[Event]]) • lift them all to Future[Event] or Future[Seq[Event]] • fallback PartialFunction for unknown Commands
 returning a failed Future
  53. Behavior DSL • PartialFunctions for all results we want to

    produce
 (Event, Future[Event], Seq[Event] and Future[Seq[Event]]) • lift them all to Future[Event] or Future[Seq[Event]] • fallback PartialFunction for unknown Commands
 returning a failed Future • compose them all with ‘orElse’ (fallback as last)
  54. Behavior DSL • PartialFunctions for all results we want to

    produce
 (Event, Future[Event], Seq[Event] and Future[Seq[Event]]) • lift them all to Future[Event] or Future[Seq[Event]] • fallback PartialFunction for unknown Commands
 returning a failed Future • compose them all with ‘orElse’ (fallback as last) • Since Future has two possible outcomes (Success and Failure)
 we can build the total functions we need
  55. def behavior(num: ProductNumber): Behavior[Product] = { // DSL to build

    a Behavior using PF behaviorFor[Product].whenConstructing { it => it.emitsEvent { // PF (Command) => Event case cmd: CreateProduct if cmd.price > 0 => ProductCreated(cmd.name, cmd.description, cmd.price) } it.acceptsEvents { // PF (Event) => Aggregate case evt: ProductCreated => Product(evt.name, evt.description, evt.price, num) } }.whenUpdating { it => it.rejectsCommands { // PF (Aggregate, Command) => Throwable case (prod, cmd: ChangePrice) if cmd.price < prod.price => new CommandException("Can't decrease the price") } it.emitsSingleEvent { // PF (Aggregate, Command) => Event case (_, cmd: ChangePrice) if cmd.price > 0 => PriceChanged(cmd.price) }
  56. def behavior(num: ProductNumber): Behavior[Product] = { // DSL to build

    a Behavior using PF behaviorFor[Product].whenConstructing { it => it.emitsEvent { // PF (Command) => Event case cmd: CreateProduct if cmd.price > 0 => ProductCreated(cmd.name, cmd.description, cmd.price) } it.acceptsEvents { // PF (Event) => Aggregate case evt: ProductCreated => Product(evt.name, evt.description, evt.price, num) } }.whenUpdating { it => it.rejectsCommands { // PF (Aggregate, Command) => Throwable case (prod, cmd: ChangePrice) if cmd.price < prod.price => new CommandException("Can't decrease the price") } it.emitsSingleEvent { // PF (Aggregate, Command) => Event case (_, cmd: ChangePrice) if cmd.price > 0 => PriceChanged(cmd.price) }
  57. case evt: ProductCreated => Product(evt.name, evt.description, evt.price, num) } }.whenUpdating

    { it => it.rejectsCommands { // PF (Aggregate, Command) => Throwable case (prod, cmd: ChangePrice) if cmd.price < prod.price => new CommandException("Can't decrease the price") } it.emitsSingleEvent { // PF (Aggregate, Command) => Event case (_, cmd: ChangePrice) if cmd.price > 0 => PriceChanged(cmd.price) } it.acceptsEvents { // PF (Aggregate, Event) => Aggregate case (prod, evt: PriceChanged) => prod.copy(price = evt.newPrice) } } }
  58. Akka for Event Sourcing • Given an Aggregate, its Protocol

    and its Behavior… • …we can have an AggregateActor that understands the life-cycle of an Aggregate
  59. Akka for Event Sourcing • Given an Aggregate, its Protocol

    and its Behavior… • …we can have an AggregateActor that understands the life-cycle of an Aggregate • Events are stored and replayed via Akka Persistence
  60. Akka for Event Sourcing • Given an Aggregate, its Protocol

    and its Behavior… • …we can have an AggregateActor that understands the life-cycle of an Aggregate • Events are stored and replayed via Akka Persistence • One Actor per Aggregate Id
  61. Akka for Event Sourcing • Given an Aggregate, its Protocol

    and its Behavior… • …we can have an AggregateActor that understands the life-cycle of an Aggregate • Events are stored and replayed via Akka Persistence • One Actor per Aggregate Id • Actor MUST handle one single Command at time (become/unbecome)
  62. Akka for Event Sourcing • Given an Aggregate, its Protocol

    and its Behavior… • …we can have an AggregateActor that understands the life-cycle of an Aggregate • Events are stored and replayed via Akka Persistence • One Actor per Aggregate Id • Actor MUST handle one single Command at time (become/unbecome) • Guarantees consistency without Optimistic Locking
  63. AggregateActor Behavior // initialized with Aggregate’s Id and Behavior class

    AggregateActor[A <: Aggregate](identifier: A#Id, behavior: Behavior[A]) extends PersistentActor Event Store
  64. AggregateActor Behavior // initialized with Aggregate’s Id and Behavior class

    AggregateActor[A <: Aggregate](identifier: A#Id, behavior: Behavior[A]) extends PersistentActor Event Store CreateProduct receives first command
  65. AggregateActor Behavior // initialized with Aggregate’s Id and Behavior class

    AggregateActor[A <: Aggregate](identifier: A#Id, behavior: Behavior[A]) extends PersistentActor Event Store E0 Behavior emits first Event Actor pesists it ProductCreated
  66. AggregateActor Behavior // initialized with Aggregate’s Id and Behavior class

    AggregateActor[A <: Aggregate](identifier: A#Id, behavior: Behavior[A]) extends PersistentActor Event Store E0 Behavior applies Event and constructs Aggregate Aggregate ProductCreated
  67. AggregateActor Behavior // initialized with Aggregate’s Id and Behavior class

    AggregateActor[A <: Aggregate](identifier: A#Id, behavior: Behavior[A]) extends PersistentActor ChangePrice receives an update Event Store E0 Aggregate
  68. AggregateActor Behavior // initialized with Aggregate’s Id and Behavior class

    AggregateActor[A <: Aggregate](identifier: A#Id, behavior: Behavior[A]) extends PersistentActor Event Store E1 E0 Aggregate Behavior emits Event Actor pesists it PriceChanged
  69. AggregateActor Behavior // initialized with Aggregate’s Id and Behavior class

    AggregateActor[A <: Aggregate](identifier: A#Id, behavior: Behavior[A]) extends PersistentActor Event Store E1 E0 Aggregate PriceChanged Behavior applies Event to Aggregate
  70. View Projections • Akka Persistence Query (experimental in Akka 2.4.0)

    • Generate Read Model (Views) from Events • Reactive Stream of Events
  71. View Projections • Akka Persistence Query (experimental in Akka 2.4.0)

    • Generate Read Model (Views) from Events • Reactive Stream of Events • ProjectionActor is an ActorSubscriber (backpressure)