README.mkd

Shopping cart, Event Sourcing style

This is an example of how to interact with a shopping cart by emitting actions or events.

A processor object dispatches incoming events to susbcribers, who are in charge of mutating data (an order in this case).

Run a simple user session that interacts with the cart with

ruby user_session.rb

This will update the order and print it to the screen.

It will also log all events to ./events.log.

You can re-play the events log against a new blank order with

ruby buid_from_log.rb

This should re-play the event history and recreate the exact same state that you got to by running the user session.

So what?

Modelling all actions in the system as a flat list of actions or events could have some benefits:

• you can recreate the state of the data just by re-playing the event log, in order.
• you can re-play parts of the history to help you debug user or other actions.
• you can re-play events against new code, for example to generate analytics or store data in different places
• backups and replication are easy, just append events to a new log somewhere
• event subscribers can be in your own code, or they can subscribe over the network. You can build a plugin or addon system. A new addon just needs to subscribe to the events it's interested in.

These are the ideas behind the Event Sourcing pattern, Apache Kafka and others.

build_from_log.rb

require "json"
require_relative "./processor"

order = Order.new

File.readlines("./events.log").each do |l|
  event = JSON.parse(l, symbolize_names: true)
  order = PROCESSOR.dispatch(event[:type], event[:payload], order)
end

puts order.inspect

processor.rb

require "ostruct"
require "json"

class Order
  attr_accessor :id, :net_total, :total, :items
  attr_reader :tax

  def initialize
    @net_total = 0
    @tax = 0
    @total = 0
    @items = []
  end

  def tax=(amount)
    @tax = amount
    recalculate_totals
  end

  def add_item(item)
    items << item

    recalculate_totals
  end

  def remove_item(id)
    it = items.find{|i| i.id == id }
    idx = items.index(it)
    items.delete_at idx

    recalculate_totals
  end

  def recalculate_totals
    self.net_total = items.reduce(0){|sum, item| sum + item[:units] * item[:price] }
    self.total = net_total + tax
    self
  end
end

class Processor
  def initialize
    @subscribers = Hash.new([])
    @events = []
  end

  def subscribe(event_type, handler)
    subscribers[event_type] += [handler]
  end

  def dispatch(event_type, payload, state)
    events << {type: event_type, payload: payload}

    subscribers[event_type].reduce(state) do |st, handler|
      handler.call st, payload, self
    end
  end

  def log_to(file, &block)
    yield

    File.open(file, "w+") do |f|
      f.write events.map{|e| JSON.dump(e)}.join("\n")
    end
  end

  private
  attr_reader :subscribers, :events
end

PROCESSOR = Processor.new

# These are the system's core events.
PROCESSOR.subscribe "orders.created", ->(order, payload, processor) {
  order.id = payload[:id]
  order
}

PROCESSOR.subscribe "orders.items.added", ->(order, payload, processor) {
  order.add_item OpenStruct.new(payload)

  order
}

PROCESSOR.subscribe "orders.items.removed", ->(order, payload, processor) {
  order.remove_item payload[:id]

  order
}

PROCESSOR.subscribe "orders.tax", ->(order, payload, processor) {
  order.tax = payload[:tax]
  order
}

user_session.rb

require_relative "./processor"

# Define some custom addons that respond to events and dispatch other events
PROCESSOR.subscribe "orders.items.added", ->(order, payload, processor) {
  if order.net_total > 20
    processor.dispatch("orders.tax", {tax: 10}, order)
  end

  order
}


PROCESSOR.subscribe "orders.items.removed", ->(order, payload, processor) {
  if order.net_total < 20
    processor.dispatch("orders.tax", {tax: -10}, order)
  end

  order
}

# Initial state of an order
order = Order.new

PROCESSOR.log_to "./events.log" do
  order = PROCESSOR.dispatch("orders.created", {id: "abc"}, order)
  order = PROCESSOR.dispatch("orders.items.added", {id: 11, name: "iPhone 6", units: 2, price: 100}, order)
  order = PROCESSOR.dispatch("orders.items.added", {id: 22, name: "Samsung Galaxy", units: 1, price: 50}, order)
  order = PROCESSOR.dispatch("orders.items.added", {id: 33, name: "Nokia 123", units: 1, price: 30}, order)
  order = PROCESSOR.dispatch("orders.items.removed", {id: 33}, order)
end

puts order.inspect