Field guide to DDD/CQRS using the Scala Type System and Akka

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Fun.CQRS Field guide to
DDD/CQRS using the Scala Type System and Akka

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Who am I? Renato
Cavalcanti   Independent consultant (mainly Scala)  Founder of BeScala @renatocaval  [email protected]  www.strongtyped.io github.com/rcavalcanti  github.com/strongtyped

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Credits Christophe Vanﬂeteren Filip
Mortier Jeroen Vuerinckx (@jeroenvu) Karel Maesen (@karelmaesen) Peter Mortier (@kwarkk) Peter Rigole (@peterrigole) Philip Van Bauwel (@vanbauph) Natal Vande Casteele Dan Brooke (@kiequoo) - GitHub

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Agenda • DDD /
CQRS / Event Sourcing (intro) • Akka and asynchrounous programming • Akka Persistence and Event Sourcing • View Projections with Akka Persistence Query • Aggregates / Commands / Events in Scala • Type members: type projections • Partial Functions, lifting and function composition • Magnet Pattern - implicits (bonus)

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka DDD - Aggregate •
Aggregate is a central DDD concept.

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

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.

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.

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka CQRS - Event Sourcing
• "It is simply the creation of two objects where there was previously only one" - Greg Young

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

• "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.

• "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.

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Event Driven / Event
Sourcing • CQRS is Event Driven, but not necessarily implements Event Sourcing

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

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

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)

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! Damn blocking!

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

Sourcing • in asynchronous CQRS: •tx(Cmd 㱺 Write Model 㱺 Event) •tx(Event 㱺 View1) •tx(Event 㱺 View2) •tx(Event 㱺 View3) • Introduces Eventual Consistency

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka • Aggregate is responsible
for consistency (DDD)

for consistency (DDD) • You can only modify one Aggregate per transaction (DDD)

for consistency (DDD) • You can only modify one Aggregate per transaction (DDD) • Write Model receives Commands and produces Events (CQRS)

for consistency (DDD) • You can only modify one Aggregate per transaction (DDD) • Write Model receives Commands and produces Events (CQRS) • Strict Consistency on Write Model side (DDD / CQRS) 

for consistency (DDD) • You can only modify one Aggregate per transaction (DDD) • Write Model receives Commands and produces Events (CQRS) • Strict Consistency on Write Model side (DDD / CQRS)  • Read Models are created from Events (CQRS)

for consistency (DDD) • You can only modify one Aggregate per transaction (DDD) • Write Model receives Commands and produces Events (CQRS) • Strict Consistency on Write Model side (DDD / CQRS)  • Read Models are created from Events (CQRS) • Eventual Consistency if asynchronous (CQRS / Event Sourcing)

for consistency (DDD) • You can only modify one Aggregate per transaction (DDD) • Write Model receives Commands and produces Events (CQRS) • Strict Consistency on Write Model side (DDD / CQRS)  • Read Models are created from Events (CQRS) • Eventual Consistency if asynchronous (CQRS / Event Sourcing) • Events can be stored and replayed (Event Sourcing)

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka trait Behavior { }
Basic Operations (non-reactive) trait DomainCommand trait DomainEvent trait Aggregate

Basic Operations (non-reactive) trait DomainCommand trait DomainEvent trait Aggregate // validates a Command and produces an Event def validate(cmd: DomainCommand): DomainEvent

Basic Operations (non-reactive) trait DomainCommand trait DomainEvent trait Aggregate // validates a Command and produces an Event def validate(cmd: DomainCommand): DomainEvent // applies an Event and produces a new Aggregate def applyEvent(evt: DomainEvent): Aggregate

Basic Operations (non-reactive) trait DomainCommand trait DomainEvent trait Aggregate // validates a Command and produces an Event def validate(cmd: DomainCommand): DomainEvent // validates a Command against current Aggregate state def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent // applies an Event and produces a new Aggregate def applyEvent(evt: DomainEvent): Aggregate

Basic Operations (non-reactive) trait DomainCommand trait DomainEvent trait Aggregate // validates a Command and produces an Event def validate(cmd: DomainCommand): DomainEvent // validates a Command against current Aggregate state def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent // applies an Event and produces a new Aggregate def applyEvent(evt: DomainEvent): Aggregate // applies an Event on current Aggregate def applyEvent(evt: DomainEvent, agg: Aggregate): Aggregate

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka State of Aggregate Event
Store

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka E1 // first command
is validated def validate(cmd: DomainCommand): DomainEvent // CreateLottery => LotteryCreated State of Aggregate Event Store

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka E1 S1 // first
command is validated def validate(cmd: DomainCommand): DomainEvent // CreateLottery => LotteryCreated // applies an Event and produces a new Aggregate def applyEvent(evt: DomainEvent): Aggregate // LotteryCreated => Lottery State of Aggregate Event Store

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka E1 E2 S1 S2
// applies an Event on current Aggregate def applyEvent(evt: DomainEvent, agg: Aggregate): Aggregate // (ParticipantAdded, Lottery) => Lottery // validates a Command against current Aggregate state def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent // (AddParticipant, Lottery) => ParticipantAdded State of Aggregate Event Store

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka E1 E2 E3 S2
S3 // applies an Event on current Aggregate def applyEvent(evt: DomainEvent, agg: Aggregate): Aggregate // (WinnerSelected, Lottery) => Lottery // validates a Command against current Aggregate state def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent // (Run, Lottery) => WinnerSelected State of Aggregate Event Store

// validates a Command against current Aggregate state def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent // (Run, Lottery) => ??? E4 State of Aggregate Event Store

// validates a Command against current Aggregate state def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent // (Run, Lottery) => ??? E4 Invalid Command! Lottery has already a winner! State of Aggregate Event Store

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Watch out! We don’t
throw exceptions in Scala. We’ll ﬁx it soon! E1 E2 E3 S3 // validates a Command against current Aggregate state def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent // (Run, Lottery) => ??? E4 Invalid Command! Lottery has already a winner! State of Aggregate Event Store

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka E1 State of Aggregate
Event Store E2 E3

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka E1 S1 // applies
an Event and produces a new Aggregate // LotteryCreated => Lottery def applyEvent(evt: DomainEvent): Aggregate E1 State of Aggregate Event Store E2 E3 S1

Event Store E2 E3 S1 S2 S1 S2 // applies an Event on current Aggregate // (ParticipantAdded, Lottery) => Lottery def applyEvent(evt: DomainEvent, agg: Aggregate): Aggregate E2

Event Store E2 E3 S2 S3 E3 S2 S3 // applies an Event on current Aggregate // (WinnerSelected, Lottery) => Lottery def applyEvent(evt: DomainEvent, agg: Aggregate): Aggregate

Event Store E2 E3 S3 E3 S2 S3 // applies an Event on current Aggregate // (WinnerSelected, Lottery) => Lottery def applyEvent(evt: DomainEvent, agg: Aggregate): Aggregate

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Akka Basics type Receive
= PartialFunction[Any, Unit]

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Akka Basics • We
send messages to an Actor via its ActorRef. No direct access

send messages to an Actor via its ActorRef. No direct access • Mailbox - Actor processes one message at a time

send messages to an Actor via its ActorRef. No direct access • Mailbox - Actor processes one message at a time • May have mutable state, but shielded from external world

send messages to an Actor via its ActorRef. No direct access • Mailbox - Actor processes one message at a time • May have mutable state, but shielded from external world • An Actor stays in memory until it dies or it’s terminated

send messages to an Actor via its ActorRef. No direct access • Mailbox - Actor processes one message at a time • May have mutable state, but shielded from external world • An Actor stays in memory until it dies or it’s terminated • Akka Persistence allows us to save messages and recover state

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka ActorRef // Actor's receive
method def receive: Receive = { case msg: SomeMsg => println(s"got a message: $msg") case msg: AnotherMsg => println(s"got another one: $msg") } ActorSystem Mailbox Actor SomeMsg AnotherMsg

method def receive: Receive = { case msg: SomeMsg => println(s"got a message: $msg") case msg: AnotherMsg => println(s"got another one: $msg") } ActorSystem Mailbox Actor Thread 1 - application SomeMsg AnotherMsg

method def receive: Receive = { case msg: SomeMsg => println(s"got a message: $msg") case msg: AnotherMsg => println(s"got another one: $msg") } ActorSystem Mailbox Actor Thread 2 - akka pool SomeMsg AnotherMsg

method def receive: Receive = { case msg: SomeMsg => println(s"got a message: $msg") case msg: AnotherMsg => println(s"got another one: $msg") } ActorSystem Mailbox Actor SomeMsg AnotherMsg Thread 1 - application

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka ActorSystem AggregateActor ActorRef Actor’s
Mailbox Event Store Behavior

Mailbox CreateLottery (message) Event Store Behavior

Mailbox CreateLottery Event Store Behavior

Mailbox CreateLottery Event Store Behavior (validate)

Mailbox Event Store LotteryCreated Behavior

Mailbox Event Store LotteryCreated LotteryCreated E1 Behavior

Mailbox Event Store LotteryCreated LotteryCreated E1 Lottery S1 Behavior (apply)

Mailbox Event Store LotteryCreated E1 Lottery S1 Behavior

Mailbox Event Store LotteryCreated E1 Lottery S1 AddParticipant (message) Behavior

Mailbox Event Store LotteryCreated E1 Lottery S1 AddParticipant Behavior

Mailbox Event Store LotteryCreated E1 Lottery S1 AddParticipant Behavior (validate)

Mailbox Event Store LotteryCreated E1 Lottery S1 ParticipantAdded Behavior

Mailbox Event Store LotteryCreated E1 Lottery S1 ParticipantAdded ParticipantAdded E2 Behavior

Mailbox Event Store LotteryCreated E1 Lottery S1 ParticipantAdded ParticipantAdded E2 S2 Behavior (apply)

Mailbox Event Store LotteryCreated E1 Lottery S1 ParticipantAdded E2 S2 Behavior

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka AggregateManager [Lottery]

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka AggregateManager [Lottery] AggregateActor Lottery.id
= 1

= 1 • Entry point for a given Aggregate type

= 1 • Entry point for a given Aggregate type • Manage it’s children

= 2 AggregateActor Lottery.id = 1 • Entry point for a given Aggregate type • Manage it’s children

= 2 AggregateActor Lottery.id = 1 • Entry point for a given Aggregate type • Manage it’s children • Forward message to the right Aggregate base on id

= 3 AggregateActor Lottery.id = 2 AggregateActor Lottery.id = 1 • Entry point for a given Aggregate type • Manage it’s children • Forward message to the right Aggregate base on id

= 3 AggregateActor Lottery.id = 2 • Entry point for a given Aggregate type • Manage it’s children • Forward message to the right Aggregate base on id • Stop children according to conﬁgured passivation strategy

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Akka Persistence - Snapshots
E1 State of Aggregate Event Store E2 E3

E1 State of Aggregate Event Store E2 E3 S1

E1 State of Aggregate Event Store E2 E3 S1 S2

E1 State of Aggregate Event Store E2 E3 S2 S3 snapshot

E1 State of Aggregate Event Store E2 E3 S3 snapshot E4 S4

E1 State of Aggregate Event Store E2 E3 E4 Crashed!!

E1 State of Aggregate Event Store E2 E3 E4 snapshot S3 On restart!

E1 State of Aggregate Event Store E2 E3 E4 snapshot S3 S4 On restart!

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Coding the C in
CQRS • A Protocol - message (cmd & evt) for a given Aggregate • A case class extending the Aggregate trait • A Behavior • Behavior DSL - convenient way for coding behavior base on commands and events

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka a1 a3 a2 b1
b2 b3 domain = a1, a2 and a3 a4 PartialFunction[A, B]

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka type Receive = PartialFunction[Any,
Unit] a1 b1 x1 Unit domain = Any a4

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Typed Protocol  (demo)

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Basic Operations - Revisited
trait Behavior { } def validate(cmd: DomainCommand): Future[DomainEvent] def validate(cmd: DomainCommand, agg: Aggregate): Future[Seq[DomainEvent]] def applyEvent(evt: DomainEvent): Aggregate def applyEvent(evt: DomainEvent, agg: Aggregate): Aggregate def validate(cmd: DomainCommand): DomainEvent def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Lifting… // behavior def
validate(cmd: DomainCommand): Future[DomainEvent] def validate(cmd: DomainCommand, agg: Aggregate): Future[Seq[DomainEvent]]

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Lifting… // behavior def
validate(cmd: DomainCommand): Future[DomainEvent] def validate(cmd: DomainCommand, agg: Aggregate): Future[Seq[DomainEvent]] // on-create def validate(cmd: DomainCommand): DomainEvent // post-create def validate(cmd: DomainCommand, agg: Aggregate): DomainEvent def validate(cmd: DomainCommand, agg: Aggregate): Future[DomainEvent] def validate(cmd: DomainCommand, agg: Aggregate): Seq[DomainEvent]

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka BehaviorDSL  (demo)

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Coding the Q in
CQRS • Akka Persistence Query (experimental in Akka 2.4.0) • Generate Read Model (Views) from Events • Akka Streams Source with selected Events • ProjectionActor to consume the source and save the pointer (offset) to the last processed event

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka ProjectionActor E1 E2 E3
Event Stream Projection E1 E2 E3 • Subscriber to an Event Stream  via Akka Persistence Query • Forward each event to a Projection deﬁnition • Projection is responsible for generating and updating the Read Model • ProjectionActor does NOT accept new events while Projection is busy

Event Stream Projection E2 E3 lastProcessedEvent = 1 • Subscriber to an Event Stream  via Akka Persistence Query • Forward each event to a Projection deﬁnition • Projection is responsible for generating and updating the Read Model • ProjectionActor does NOT accept new events while Projection is busy

Event Stream Projection E3 lastProcessedEvent = 2 • Subscriber to an Event Stream  via Akka Persistence Query • Forward each event to a Projection deﬁnition • Projection is responsible for generating and updating the Read Model • ProjectionActor does NOT accept new events while Projection is busy

Event Stream Projection lastProcessedEvent = 3 • Subscriber to an Event Stream  via Akka Persistence Query • Forward each event to a Projection deﬁnition • Projection is responsible for generating and updating the Read Model • ProjectionActor does NOT accept new events while Projection is busy

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Projection (demo)

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Akka for CQRS /
Event Sourcing • Given an Aggregate, its Protocol and its Behavior…

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

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

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

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)

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

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 • Projections backed by Akka Persistence Query

@renatocaval #Devoxx #funcqrs #scala #cqrs #akka Questions? http://bit.ly/fun-cqrs   (GitHub)
@renatocaval  [email protected]  www.strongtyped.io github.com/rcavalcanti  github.com/strongtyped

Field guide to DDD/CQRS using the Scala Type Sy...

Field guide to DDD/CQRS using the Scala Type System and Akka

More Decks by Renato Cavalcanti

Other Decks in Programming

Featured

Transcript