�ݺ�ߣ

NOT YOUR USUAL CRUD!
A DIFFERENT APPROACH TO ARCHITECTURE AND PERSISTENCE

WHAT IS WRONG WITH CRUD?
In place updates can lead to degraded performance
Scalability is limited
History is lost
Does not support different read and write semantics
Con icts are hard to manage

WHAT CAN YOU EXPECT FROM
TODAY?
Introduction to Event Sourcing
Introduction to CQRS
Proof that it is not complicated just different

WHAT ARE EVENTS?
Immutable facts from the past
Contain all information describing the change

WHAT IS EVENT SOURCING
ABOUT?
Do not store and mutate the current state
Record sequence of changes as events in a journal
Consider the journal to be the single source of truth

HOW DOES A JOURNAL LOOK LIKE?
create table if not exists journal(
    id BIGSERIAL PRIMARY KEY, 
    aggregate_id VARCHAR(255) NOT NULL,
    version BIGINT NOT NULL,
    type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    tags JSONB NOT NULL,
    timestamp TIMESTAMP NOT NULL
);

WHAT DO WE STORE
Payload
{
  "product": "123",
  "amount": 5
}
Tags
{
  "aggregate" -> "cart",
  "tenant" -> "123"
}

LIMITATIONS
Data model does not qualify for advanced queries
Tooling is not as advanced

CQRS
Command Query Responsibility Segregation
Separate data model for processing updates
Separate data model for performing queries

LIMITATIONS
Complexity
Eventual Consistency

WORKSHOP
Clone the code from the git repository
JDK8+, SBT, PostgreSQL required
Please ask for assistence
https://github.com/dpfeiffer/eventsourcing-postgres-jdbc contains
example (master, solutions)
init.sql contains DDL

JOURNAL FOR STORING EVENTS
trait Journal {
  def store(events: List[EventEnvelope]): Unit
  def stream(aggregateId: String): LazyList[EventEnvelope]
}

AGGREGATES AS COMMAND
HANDLERS
abstract class Aggregate[C <: Command, E <: Event, S](journal: Journal, eventStream: EventStream) {
  protected def initialState: S
  protected def encode(event: E): String
  protected def decode(`type`: String, payload: String): E
  protected def handleCommand(s: S, c: C): (List[E], Any)
  protected def handleEvent(s: S, e: E): S
  def processCommand(c: C): Any = ???
  ...
}

EVENT STREAM AS MESSAGING
MECHANISM
class EventStream {
  def publish[E](event: E): Unit = ???
  def subscribe[E: ClassTag](handler: E => Unit): Unit = ???
}

SHOPPING CART
Adding products: quantity, productId
Removing products
Chaning quantity

SHOPPING CART - ENCODING
import eventsourcing._
import example.ShoppingCart._
import play.api.libs.json._
object ShoppingCart {
  trait ShoppingCartCommand extends Command
  case class AddItem(id: String, productId: String, quantity: Int) extends ShoppingCartCommand
  abstract class ShoppingCartEvent(`type`: String) extends Event(`type`)
  case class ItemAdded(productId: String, quantity: Int) extends ShoppingCartEvent("item.added")
  private val addedFormat = Json.format[ItemAdded]
}
class ShoppingCart(j: Journal, s: EventStream) 
  extends Aggregate[ShoppingCartCommand, ShoppingCartEvent, Map[String, Int]](j, s){
  override def decode(`type`: String, payload: String): ShoppingCart.ShoppingCartEvent = `type` match {
    case "item.added" => addedFormat.reads(Json.parse(payload)).get
  }
  override def encode(event: ShoppingCart.ShoppingCartEvent): String = event match {
    case e: ItemAdded => addedFormat.writes(e).toString()
  }
}

SHOPPING CART - COMMAND
HANDLING
class ShoppingCart(j: Journal, s: EventStream) 
  extends Aggregate[ShoppingCartCommand, ShoppingCartEvent, Map[String, Int]](j, s){
    override def handleCommand(s: ShoppingCartState, c: ShoppingCart.ShoppingCartCommand): (List[ShoppingCa
      case c: AddItem => ???
    }
    override def handleEvent(s: ShoppingCartState, e: ShoppingCart.ShoppingCartEvent): ShoppingCartState = 
      e match {
        case i: ItemAdded => ???
      }
    }
}

SHOPPING CART - USAGE
package example._
import eventsourcing._
object Main {
  def main(args: Array[String]]): Unit = {
    val carts = new ShoppingCart(journal, eventStream)
    carts.processCommand(ShoppingCart.AddItem("customer-1", "jeans"))
    carts.processCommand(ShoppingCart.AddItem("customer-1", "blue jeans"))
  }
}
select * from journal;

SHOPPING CART ICON
Query view for a sum of items in the shopping cart
Use the event stream to subscribe to shopping carts updates

SHOPPING CART ICON -
ENCODING
abstract class View[E <: Event: ClassTag, R](journal: Journal, stream: EventStream) {
  protected def decode(`type`: String, payload: String): E
  protected def handleEvent(result: R, e: E): R
  protected def initialState: R
  protected def initialStream: LazyList[EventEnvelope]
  def get: R = ???
}

PRODUCT
Reserve quantity if a user puts it into the cart
Release qunatity if a user removes it from the cart

PRODUCT - ENCODING
class Product(journal: Journal, stream: EventStream) extends Aggregate[...](journal, stream){
  stream.subscribe[ShoppingCart.ItemAdded]{ event =>
    
  }
}

REFERENCES
https://docs.microsoft.com/en-us/azure/architecture/patterns/cqrs for
useful  gures

�ݺ�ߣ

Event Sourcing without any Framework

More Related Content

Event Sourcing without any Framework