more stuff

project.clj

@@ -5,14 +5,19 @@
             :url "http://www.eclipse.org/legal/epl-v10.html"}
   :dependencies [[org.clojure/clojure "1.9.0"]
                  [org.clojure/core.async "0.4.474"]
+                 [org.clojure/math.combinatorics "0.1.4"]
                  ;; web
                  [aero "1.1.3"]
                  [http-kit "2.3.0"]
                  [compojure "1.6.1"]
+                 [com.taoensso/nippy "2.14.0"]
                  [javax.servlet/javax.servlet-api "3.1.0"]
+                 [org.apache.kafka/kafka-streams "1.1.0"]
+                 [org.apache.kafka/kafka-clients "1.1.0"]
 
                  ;;utils
                  [slingshot "0.12.2"]
+                 [spootnik/kinsky "0.1.22"]
                  [cheshire "5.8.0"]]
   :java-source-paths ["src/java"]
   :main ^:skip-aot skirmish.core
@@ -21,5 +26,8 @@
   :profiles {
              :uberjar {:aot :all}
              :dev {:jvm-opts ["-XX:-OmitStackTraceInFastThrow"]
-                   :dependencies [[midje "1.9.1"]]}})
+                   :dependencies [[midje "1.9.1"]
+                                  [stylefruits/gniazdo "1.0.1"]
+                                  [com.stuartsierra/frequencies "0.1.0"]
+                                  [clj-http "3.9.0"]]}})
 

resources/config.edn

@@ -1,4 +1,10 @@
 {
+ :kafka
+ {:bootstrap.servers ["localhost:9092"]
+  :topics-prefix "skirmish-01-"}
+
+
+
  :webserver ;; possible config from http-kit's run-server
  {:port #long #or [#env PORT 8080]
   :ip "0.0.0.0"
@@ -9,4 +15,9 @@
  [{:sec-id :foo}
   {:sec-id :bar}
   {:sec-id :toto}
-    {:sec-id :tata}]}
+  {:sec-id :tata}]
+ :benchmark {
+             :number-of-requests-per-securities 30}}
+
+
+

src/skirmish/cli/benchmark.clj

@@ -0,0 +1,116 @@
+(ns skirmish.cli.benchmark
+  (:require [clojure.math.combinatorics :refer [cartesian-product]]
+            [clojure.core.async :refer [go chan <! >! close! to-chan close!] :as async]
+            [gniazdo.core :as ws]
+            [clj-http.client :as client]
+            [com.stuartsierra.frequencies :as freq]
+            [clojure.edn :as edn]))
+
+
+(def ids (atom 0))
+
+
+(defn web-socket-to-channel [url output-ch]
+  (let [on-receive #(go (>! output-ch %))
+        headers {"accept" "application/edn"}]
+    (ws/connect url :headers headers :on-receive on-receive)))
+
+
+(defn create-trades-channel [securities]
+  (let [output (chan 1 (map edn/read-string))
+        url #(format "ws://localhost:%s/api/1/securities/%s/trades" 8080 %)
+        xf (comp (map :sec-id)
+                 (map name)
+                 (map url)
+                 (map #(web-socket-to-channel % output)))]
+    (doall (sequence xf securities))
+    (async/pub output :book.order/order-id)))
+
+
+(defn create-orders-channel [securities]
+  (let [output (chan 1 (map edn/read-string))
+        url #(format "ws://localhost:%s/api/1/securities/%s/orders" 8080 %)
+        xf (comp (map :sec-id)
+                 (map name)
+                 (map url)
+                 (map #(web-socket-to-channel % output)))]
+    (doall (sequence xf securities))
+    (async/pub output :book.order/id)))
+
+
+
+(defn create-requests-payload-for-security [i security id]
+  (let [rd (new java.util.Random i)
+        rand-bool (= (mod (. rd nextInt) 2) 0)  ;; java random number generator REALLY sucks
+        type (if rand-bool :sell :buy)
+        price (+ (* 3 (. rd nextGaussian)) 30)
+        int-price (int (* price 100))
+        quantity (reduce + 1 (map #(. rd nextInt (inc %)) (range 100)))]
+
+    {:type type :quantity quantity :price int-price :security security :benchmark/id id}))
+
+
+(defn create-requests-for-security [[i security]]
+  (let [req-id (swap! ids inc)
+        payload (create-requests-payload-for-security i (str (:sec-id security)) req-id)
+        sec-id-str (-> security :sec-id name)
+        params {:throw-exceptions false :content-type :edn :headers {"accept" "application/edn"}}
+        url (format "http://localhost:%s/api/1/securities/%s" 8080 sec-id-str)]
+    (fn []
+      (let [now (System/nanoTime)]
+        (client/post url (assoc params :form-params (assoc payload :benchmark/request-time now)))))))
+
+
+(defn make-requests [n securities]
+  (map create-requests-for-security (cartesian-product (range n) securities)))
+
+
+(defn no-more-then-n-simultaneously [f input-chan n]
+  (let [buffer (chan)]
+    (go (async/pipeline n buffer (map f) input-chan))
+    buffer))
+
+(defn decode-response [response]
+  (let [now (System/nanoTime)
+        ret {}
+        ret (assoc ret :status (:status response))
+        ret (assoc ret :order (edn/read-string (:body response)))
+        ret (assoc ret :duration (/ (- now  (get-in ret [:order :benchmark/request-time])) (Math/pow 10 9)))]
+    ret))
+
+
+(defn benchmark [requests order trades]
+  (let [decoded-responses (chan 1 (map decode-response))
+        responses (no-more-then-n-simultaneously #(%) (to-chan requests) 10)]
+    (go (async/pipe responses decoded-responses))
+    decoded-responses))
+
+
+(defn compute-result [results new-result]
+  (let [status-code (:status new-result)
+        duration (:duration new-result)
+        old-status-code-count  (get-in results [:status status-code] 0)
+        ret results
+        ret (assoc-in ret [:status status-code] (inc old-status-code-count))
+        ret (assoc ret :durations (conj (:durations ret) duration))]
+    ret))
+
+(defn pp-durations [durations]
+  (let [freq-map (frequencies durations)]
+    (print (freq/stats freq-map))))
+
+(defn display-results [results]
+  (println results)
+  (println "number of request: " (:n results))
+  (println "status map: " (:status results))
+  (pp-durations (:durations results)))
+
+(defn -main [config]
+  (let [benchmark-config (:benchmark config)
+        requests (-> benchmark-config :number-of-requests-per-securities (make-requests (:securities config)))
+        trades (create-trades-channel (:securities config))
+        orders (create-orders-channel (:securities config))
+        results (benchmark requests orders trades)]
+    (display-results (async/<!! (async/reduce compute-result {} results)))))
+
+

src/skirmish/core.clj

@@ -1,10 +1,12 @@
 (ns skirmish.core
   (:require [skirmish.cli.core]
             [skirmish.rest.core]
+            [skirmish.cli.benchmark]
             [aero.core :refer (read-config)])
   (:gen-class))
 
-(def apps {:webserver skirmish.rest.core/-main})
+(def apps {:webserver skirmish.rest.core/-main
+           :benchmark skirmish.cli.benchmark/-main})
 
 (defn -main [& argv]
   (let [config (-> "config.edn" clojure.java.io/resource read-config)

src/skirmish/rest/marketdata.clj

@@ -1 +1,13 @@
-(ns skirmish.rest.marketdata)
+(ns skirmish.rest.marketdata
+  (:require [compojure.core :refer :all]))
+
+
+(defn create-routes [exchange-apis]
+  (routes
+    (context "/api/1" []
+             (GET "/securities" [] )
+             (GET "/securities/:sec-id" [] (get-security exchange-apis))
+             (POST "/securities/:sec-id" [] (create-order exchange-apis))
+             (GET "/securities/:sec-id/trades" [] (listen-to-trades exchange-apis))
+             (GET "/securities/:sec-id/orders" [] (listen-to-orders exchange-apis)))
+    (not-found "not found")))

test/skirmish/cli/benchmark_test.clj

@@ -0,0 +1,30 @@
+(ns skirmish.cli.benchmark-test
+  (:require [skirmish.cli.benchmark :refer :all]
+            [midje.sweet :refer :all]
+            [clojure.core.async :as async]))
+
+(facts "creating simple orders"
+  (fact "buy order"
+    (let [buy-order (create-requests-payload-for-security 666 (str :equity/ibm) 1)
+          expt {:benchmark/id 1 :price 2922 :quantity 2476 :security ":equity/ibm" :type :buy}]
+      buy-order => expt));;
+  (fact "sell order"
+   (let [buy-order (create-requests-payload-for-security 667 (str :equity/ibm) 2)
+         expt {:benchmark/id 2 :price 2740 :quantity 2516 :security ":equity/ibm" :type :sell}]
+     buy-order => expt)))
+
+(facts "the utility functions"
+  (fact "no-more-then-n-simultaneously"
+     (let [simultatious (atom [])
+           input-chan (async/to-chan (range 10))
+           sleep-time #(+ (mod (+ (* 5142 %) 8765) 200) 50)
+           f (fn [i]
+               (swap! simultatious #(conj % {:id i :status :start :value 1}))
+               (Thread/sleep (sleep-time i))
+               (swap! simultatious #(conj % {:id i :status :end :value -1}))
+               i)
+           res (async/<!! (async/into [] (no-more-then-n-simultaneously f input-chan 3)))
+           simultatious-over-time (reductions #(+ %1 (:value %2)) 0 @simultatious)]
+       (apply max simultatious-over-time) => 3
+       (first simultatious-over-time) => 0
+       (last simultatious-over-time) => 0)))