Implemented middleware in a somewhat stateless fashion

README.md

@@ -6,29 +6,36 @@ A stock exchange simulator.  The scope of this project is not yet defined but it
 allow end user(s?) to exchange securities via the common exchange rules.
 
 
-## Installation
+## Getting Started
-Okay, this should be easy but it is hard... 
+
+### Development
+
+In development, you need nothing else than [clojure](https://clojure.org/)
+and [leiningen](https://leiningen.org/).  The easiest way to install clojure is
+to make sure you have java (8 and above) and install leiningen.
+
+### Production
+
+You will need plenty of things, the majority of them are yet to define.
+But, a good kafka deployment will be required.  Probably a database,
+hopefully datomic.
 
 
 ### Configuration
 All configuration are set in the `resources/config.edn` file.
-Configuration uses [aero](https://github.com/juxt/aero).
+Configuration uses [aero](https://github.com/juxt/aero).  The configuration
+file should be self documented.
 
 ### Technologies
 
 #### Server
-* clojure
+* [clojure](https://clojure.org/) for programming language
-* kafka
+* [kafka](http://kafka.apache.org/) for message broker/database
 
 #### In the browser
-* https://aftertheflood.com/projects/sparks/
+* [clojure script](https://clojurescript.org/) for programming language
-* http://metricsgraphicsjs.org/
+* [sparks](https://aftertheflood.com/projects/sparks/) for trivial charts
-
+* [metricsgraphicsjs](http://metricsgraphicsjs.org/) for complex charts
-## Usage
-
-FIXME: explanation
-
-    $ java -jar book-0.1.0-standalone.jar [args]
 
 
 

resources/config.edn

@@ -1,11 +1,18 @@
 {
+ ;;
  :middleware :kafka
- :kafka
+ :kafka  ;; all the kafka configuration
  {:bootstrap.servers #or [#env KAFKA_BOOTSTRAP_SERVERS "localhost:9092"]
   :topics-prefix "skirmish-01-"}
 
- :webserver ;; possible config from http-kit's run-server
+ :api-server ;; possible config from http-kit's run-server that will affect the api-server
- {:port #long #or [#env PORT 8080]
+ {:port #long #or [#env API_SERVER_PORT 1602]
+  :ip "0.0.0.0"
+  :max-body 524288 ;; 512kb, small but enough
+  :max-ws 524288}
+
+ :api-bookkeeper ;; possible config from http-kit's run-server that will affect the api-bookkeeper
+ {:port #long #or [#env API_SERVER_PORT 1603]
   :ip "0.0.0.0"
   :max-body 524288 ;; 512kb, small but enough
   :max-ws 524288}

src/skirmish/cli/benchmark.clj

@@ -94,7 +94,7 @@
   (println "status map: " (:status results))
   (pp-durations (:durations results)))
 
-(defn -main [config]
+(defn -main [config exchange-apis]
   (let [benchmark-config (:benchmark config)
         requests (-> benchmark-config :number-of-requests-per-securities (make-requests (:securities config)))
         orders (create-orders-channel (:securities config))

src/skirmish/core.clj

@@ -3,28 +3,43 @@
             [skirmish.rest.core]
             [skirmish.cli.benchmark]
             [skirmish.exchange.core]
+            [skirmish.middleware.core :refer [create-exchange-apis-for]]
             [aero.core :refer (read-config)]
             [clojure.core.async :as async])
   (:gen-class))
 
-(def apps {:api-server skirmish.rest.core/api-main
+(def targets ;; a target is a function that blocks the thread and takes two
-           :exchange-core skirmish.exchange.core/-main ;; should be skirmish.exchange.core/api-main
+             ;; arguments:  the config map and the "exchange-api"
-           :bookkeeper skirmish.rest.core/bookeeper-main
+  {:api-server     skirmish.rest.core/api-main
-           :benchmark skirmish.cli.benchmark/-main})
+   :exchange-core  skirmish.exchange.core/-main ;; should be skirmish.exchange.core/api-main
+   :api-bookkeeper skirmish.rest.core/bookeeper-main})
+   ;:benchmark      skirmish.cli.benchmark/-main})
+
+(defn run-all-targets-in-separate-threads [target-fns config exchange-apis]
+  (println "running  multiple targets. This is okay in dev but horrible in production")
+  (println target-fns)
+  (let [threads (map #(async/thread (% config exchange-apis)) target-fns)
+        finished-threads (async/merge threads)];; need to lock here
+    (loop [finished-thread (async/<!! finished-threads)]
+      (when (some? finished-thread)
+          (println (str "thread " finished-thread " finished"))
+          (recur (async/<!! finished-threads))))))
 
 
 (defn run-all-targets [target-fns config]
-  (let [threads (map #(async/thread (% config)) target-fns)]
+  (let [exchange-apis (create-exchange-apis-for config)]
-    (async/<!! (async/merge threads))))
+    (if (= (count target-fns) 1)
-
+      ((first target-fns) config exchange-apis)
+      (run-all-targets-in-separate-threads target-fns config exchange-apis))))
 
 (defn -main [& argv]
   (let [config (-> "config.edn" clojure.java.io/resource read-config)
-        xf (comp keyword? apps)
+        xf (comp (map keyword) (map targets) (filter some?))
-        target-fns (sequence xf (rest argv))]
+        target-fns (sequence xf argv)]
     (if (seq target-fns)
       (run-all-targets target-fns config)
       (do
+        (println argv)
         (println "target" target-fns "not found")
         (println "all known targets are...")
-        (println (keys apps))))))
+        (println (keys targets))))))

src/skirmish/exchange/core.clj

@@ -1,17 +1,22 @@
 (ns skirmish.exchange.core
-  (:require [skirmish.middleware.core :refer [create-exchange-apis-for]]
+  (:require [skirmish.util :refer [block-thread]]
+            [skirmish.middleware.core :refer [create-exchange-apis-for]]
             [skirmish.exchange.exchange :refer [create-exchange-for-security empty-book]]
-            [clojure.core.async :as async]
             [kinsky.async]
             [kinsky.client]))
 
 (defn create-exchange-for [exchange-api]
   (let [initial-book empty-book
-        x  (:new-order-publisher exchange-api)]
+        new-order-publisher (:new-order-publisher exchange-api)
-    (create-exchange-for-security initial-book x)))
+        trades-submitter (:trades-submitter exchange-api)
+        orders-submitter (:orders-submitter exchange-api)]
 
-(defn -main [config]
+    (create-exchange-for-security initial-book
-  (let [exchange-apis (create-exchange-apis-for config)]
+                                  new-order-publisher
-    (doseq [{sec-id :sec-id} (:securities exchange-apis)]
+                                  trades-submitter
-      (create-exchange-for (get exchange-apis sec-id)))))
+                                  orders-submitter)))
-  ;      ;api-routes (create-routes exchange-apis kafka-producer)]));;
+
+(defn -main [config exchange-apis]
+  (doseq [{sec-id :sec-id} (:securities config)]
+    (create-exchange-for (get exchange-apis sec-id)))
+  (block-thread))

src/skirmish/exchange/exchange.clj

@@ -79,22 +79,16 @@
 
 ;; bellow are the really interesting functions.
 
-(defn create-exchange-for-security [initial-book incoming-order-channel trades-channel outcoming-orders-channel]
+(defn create-exchange-for-security [initial-book new-order-publisher trades-submitter orders-submitter]
-  ;; output - trades channel, pub channel
+  (async/go
-  ;;        - orders channel, pub channel
+    (loop [order (<! new-order-publisher) book initial-book]
-  (let [trades-channel (chan)
+      (let [[new-book new-order new-trades] (process-order book order)
-        outcoming-orders-channel (chan)]
+            orders (get-processed-orders order new-order new-trades book)
-    (async/go
+            orders-copy (async/onto-chan orders-submitter orders false)
-      (loop [order (<! incoming-order-channel) book initial-book]
+            trades-copy (async/onto-chan trades-submitter new-trades false)]
-        (println "received an order!  " order)
+        (<! orders-copy)
-        (let [[new-book new-order new-trades] (process-order book order)
+        (<! trades-copy)
-              orders (get-processed-orders order new-order new-trades book)
+        (recur (<! new-order-publisher) new-book)))))
-              orders-copy (async/onto-chan outcoming-orders-channel orders false)
-              trades-copy (async/onto-chan trades-channel new-trades false)]
-          (<! orders-copy)
-          (<! trades-copy)
-          (recur (<! incoming-order-channel) new-book))))
-    [(pub trades-channel :book.order/type) (pub outcoming-orders-channel :type)]))
 
 
 

src/skirmish/middleware/core.clj

@@ -4,87 +4,24 @@
             [kinsky.async]
             [kinsky.client]))
 
-
-(defn set-kafka-as-middleware-in-exchange-api [sec-id api kafka-output-channel]
-  (let [str-sec-id (str sec-id)
-        xf (map #(hash-map :topic str-sec-id :key (:book.order/id %) :value %))
-        from (async/chan 1 xf)]
-    (async/pipe from kafka-output-channel)
-    (assoc api :incoming-order-channel from)))
-
-
-(defn create-async-kafka-producer [config]
-  (kinsky.async/producer (:kafka config)
-                         (kinsky.client/edn-serializer)
-                         (kinsky.client/edn-serializer)))
-
-
-(defn set-middleware [exchange-apis config]
-  (if (= (:middleware config) :kafka)
-    (let [kafka-output-channel (create-async-kafka-producer config)]
-      (into {} (map (fn [[k v]] [k (set-kafka-as-middleware-in-exchange-api k v kafka-output-channel)]) exchange-apis)))
-    exchange-apis))
-
-;(defn create-exchange-api-for [sec-id]
-;  (let [initial-book empty-book
-;        incoming-order-channel (async/chan)
-;        [trades-publisher new-book-order-publisher] (create-exchange-for-security initial-book incoming-order-channel)]
-;    {:sec-id                 sec-id
-;     :new-order-publisher    nil  ;; anything published here will be processed as
-;     :new-order-register     nil
-;     :incoming-order-channel incoming-order-channel
-;     :trades-publisher-bis   nil
-;     :trades-register       nil
-;     :orders-publisher       nil
-;     :orders-register        nil
-;
-;     :trades-publisher         trades-publisher
-;     :new-book-order-publisher new-book-order-publisher}))
-
-(def chan-based-exchange-apis (atom {}))
-
-(defn create-new-chan-based-exchange-api-for [sec-id]
-  ;[(pub trades-channel :book.order/type) (pub outcoming-orders-channel :type)])
-  (let [new-order (async/chan)
-        trades (async/chan)
-        orders (async/chan)]
-    {:sec-id                 sec-id
-     :new-order-publisher    new-order  ;; anything published here will be processed as
-     :new-order-register     new-order
-     :trades-publisher       trades
-     :trades-register        (async/pub trades :book.order/type)
-     :orders-publisher       orders
-     :orders-register        (async/pub orders :type)}))
-
-(defn chan-based-exchange-api-for-swapper [sec-id exchange-api]
-  (if (contains? exchange-api sec-id)
-    exchange-api
-    (assoc exchange-api sec-id (create-new-chan-based-exchange-api-for sec-id))))
-
 (defn create-chan-based-exchange-api-for [sec-id]
-  (let [swapper (partial chan-based-exchange-api-for-swapper sec-id)
-        exchange-apis (swap! chan-based-exchange-apis swapper)]
-    (get exchange-apis sec-id)))
-
-(defn create-kafka-based-exchange-api-for [sec-id]
   (let [new-order (async/chan)
         trades (async/chan)
         orders (async/chan)]
     {:sec-id                 sec-id
      :new-order-publisher    new-order  ;; anything published here will be processed as
-     :new-order-register     new-order
+     :new-order-submitter    new-order
-     :trades-publisher-bis   trades
+     :trades-publisher       (async/pub trades :book.order/type)
-     :trades-register        trades
+     :trades-submitter       trades
-     :orders-publisher       orders
+     :orders-publisher       (async/pub orders :type)
-     :orders-register        orders}))
+     :orders-submitter       orders}))
 
-(def api-creators {:kafka create-kafka-based-exchange-api-for})
+(def api-creators {})
 
 (defn create-exchange-apis-for [config]
   ;; todo: spec this function
   (let [security-definitions (map :sec-id (:securities config))
-        create-exchange-api-for (:get api-creators (:middleware config) create-chan-based-exchange-api-for)
+        create-exchange-api-for (get api-creators (:middleware config) create-chan-based-exchange-api-for)
-        f (fn [sec] [(:sec-id sec) (create-exchange-api-for (:sec-id sec))])
+        f (fn [sec] [sec (create-exchange-api-for sec)])
         ret (into {} (map f security-definitions))]
-    (set-middleware ret config)))
+    ret))
-;;

src/skirmish/rest/core.clj

@@ -1,42 +1,23 @@
 (ns skirmish.rest.core
   (:require [org.httpkit.server :refer [run-server]]
             [compojure.handler :refer [api]]
-            [clojure.core.async :as async]
+            [skirmish.util :refer [block-thread]]
-            [kinsky.client]
-            [kinsky.async]
             [skirmish.middleware.core :refer [create-exchange-apis-for]]
-            [skirmish.rest.exchange :refer [create-routes]]))
+            [skirmish.rest.exchange :refer [create-routes-for-exchange
+                                            create-routes-for-bookkeeper]]))
 
 
-(defn set-kafka-as-middleware-in-exchange-api [sec-id api kafka-output-channel]
+(defn api-main [config exchange-apis]
-  (let [str-sec-id (str sec-id)
+  (let [webserver-config (:api-server config)
-        xf (map #(hash-map :topic str-sec-id :key (:book.order/id %) :value %))
+        api-routes (create-routes-for-exchange exchange-apis)]
-        from (async/chan 1 xf)]
-    (async/pipe from kafka-output-channel)
-    (assoc api :incoming-order-channel from)))
-
-
-(defn create-async-kafka-producer [config]
-  (kinsky.async/producer (:kafka config)
-                         (kinsky.client/edn-serializer)
-                         (kinsky.client/edn-serializer)))
-
-
-(defn set-middleware [exchange-apis config]
-  (if (= (:middleware config) :kafka)
-    (let [kafka-output-channel (create-async-kafka-producer config)]
-      (into {} (map (fn [[k v]] [k (set-kafka-as-middleware-in-exchange-api k v kafka-output-channel)]) exchange-apis)))
-    exchange-apis))
-
-
-(defn api-main [config]
-  (let [webserver-config (:webserver config)
-        exchange-apis (create-exchange-apis-for config)
-        api-routes (create-routes (set-middleware exchange-apis config))]
     (println "Running server with bellow config")
     (println webserver-config)
     (run-server (api api-routes) webserver-config)))
 
 
-(defn bookeeper-main [config]
+(defn bookeeper-main [config exchange-apis]
-  (println "not yet implemented"))
+  (let [webserver-config (:api-bookkeeper config)
+        api-routes (create-routes-for-bookkeeper exchange-apis)]
+    (println "Running server with bellow config")
+    (println webserver-config)
+    (run-server (api api-routes) webserver-config)))

src/skirmish/rest/exchange.clj

@@ -32,7 +32,7 @@
 
 (defn get-new-order-channel [exchange-apis sec-id]
   (let [exchange (get exchange-apis sec-id)
-        new-book-order-channel (:new-book-order-publisher exchange)
+        new-book-order-channel (:orders-publisher exchange)
         ret (chan)]
     (sub new-book-order-channel :sell ret)
     (sub new-book-order-channel :buy ret)
@@ -88,7 +88,7 @@
     (try+
       (let [asked-order (get-body body)
             exchange-api (get-exchange-api (keyword sec-id-str) exchange-apis)
-            incoming-order-channel (:incoming-order-channel exchange-api)
+            incoming-order-channel (:new-order-submitter exchange-api)
             order (validate-order asked-order (:sec-id exchange-api))]
         (>!! incoming-order-channel order)
         (build-response request order))
@@ -118,13 +118,19 @@
                     (chan-to-websocket request order-channel ws-chan)))))
 
 
-(defn create-routes [exchange-apis]
+(defn create-routes-for-exchange [exchange-apis]
+  (routes
+    (context "/api/1" []
+      (POST "/securities/:sec-id" [] (create-order exchange-apis)))
+    (not-found "not found")))
+
+
+(defn create-routes-for-bookkeeper [exchange-apis]
   (routes
     (context "/api/1" []
       (GET "/" [] api-description)
       (GET "/securities" [] (securities exchange-apis))
       (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")))

src/skirmish/util.clj

@@ -1,4 +1,5 @@
 (ns skirmish.util
+  (:require [clojure.core.async :as async])
   (:import [skirmish.PersistentSortedList]))
 
 (defn sorted-list [& keys]
@@ -7,3 +8,6 @@
 (defn sorted-list-by [comparator & keys]
   (skirmish.PersistentSortedList/create comparator (seq keys)))
 
+(defn block-thread []
+  "Block the current thread until chuck norris gets weaker."
+  (async/<!! (async/chan)))

test/skirmish/middleware/test_core.clj

@@ -2,3 +2,12 @@
   (:require [midje.sweet :refer :all]
             [skirmish.middleware.core :refer :all]))
 
+(defn all? [xf]
+  #(every? true? (sequence xf %)))
+
+(facts "we should have a async/chan base middleware for developing."
+  (fact "It should exchanges for each security in the config"
+    (let [config {:securities [{:sec-id :molly} {:sec-id :pepper}]}
+          exchanges (create-exchange-apis-for config)]
+      (keys exchanges) => [:molly :pepper]
+      (vals exchanges) => #(every? map? %))))