Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
(definterface ES
(putIfAbsent [k v])
(rep [k old-val new-val])
(get [k]))
(defn ^ES jes []
(let [m (atom {})]
(reify ES
(rep [_ k old-val new-val]
(if (= old-val (get @m k))
(swap! m assoc k new-val)
[_ k v]
(if-let [key (get @m k)]
(swap! m assoc k v)))
(get [_ k] (get @m k)))))
(def c-in-memory-event-store
(let [streams (jes)
empty-stream (->EventStream 0 [])]
(reify EventStore
(retrieve-event-stream [this aggregate-id]
(if (.putIfAbsent streams aggregate-id empty-stream)
(.get streams aggregate-id)
(append-events [this aggregate-id previous-event-stream events]
(let [next-event-stream (->EventStream (inc (:version previous-event-stream))
(conj (:transactions previous-event-stream)
replaced (.rep streams aggregate-id previous-event-stream next-event-stream)]
(when-not replaced (throw (ConcurrentModificationException.))))))))
(ns com.jayway.rps.main
(:require [com.jayway.rps.core :as c]
[com.jayway.rps.framework :as f]
[com.jayway.rps.domain :as d]
[criterium.core :as crit]))
(defn append-event [eventStore aggregateId]
(let [es (f/retrieve-event-stream eventStore aggregateId)]
(f/append-events eventStore aggregateId es ["event"])))
(defn pappend-test [totalCount partionSize eventStore]
(count (pmap
(fn [ids]
(count (map (fn [aggregateId] (append-event eventStore aggregateId)) ids)))
(partition partionSize (range 0 totalCount)))))
(crit/quick-bench (apply pappend-test [10000 10 f/j-in-memory-event-store]))
(crit/quick-bench (apply pappend-test [10000 10 f/c-in-memory-event-store]))
(defproject rock-paper-scissors "1.0.0-SNAPSHOT"
:description "The game rock-paper-scissors implemented using CQRS & Event Sourcing in Clojure"
:dependencies [[org.clojure/clojure "1.5.0"]
[criterium "0.3.1"]]
:main com.jayway.rps.main)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.