.gitignore

@@ -13,3 +13,4 @@ pom.xml.asc
 book.iml
 skirmish.iml
 full-stack
+fixtures

resources/config.edn

@@ -1,4 +1,5 @@
 {
+ :middleware :kafka
  :kafka
  {:bootstrap.servers #or [#env KAFKA_BOOTSTRAP_SERVERS "localhost:9092"]
   :topics-prefix "skirmish-01-"}

src/skirmish/core.clj

@@ -2,18 +2,29 @@
   (:require [skirmish.cli.core]
             [skirmish.rest.core]
             [skirmish.cli.benchmark]
-            [aero.core :refer (read-config)])
+            [skirmish.exchange.core]
+            [aero.core :refer (read-config)]
+            [clojure.core.async :as async])
   (:gen-class))
 
 (def apps {:api-server skirmish.rest.core/api-main
-           :exchange-core skirmish.rest.core/api-main ;; should be skirmish.exchange.core/api-main
+           :exchange-core skirmish.exchange.core/-main ;; should be skirmish.exchange.core/api-main
+           :bookkeeper skirmish.rest.core/bookeeper-main
            :benchmark skirmish.cli.benchmark/-main})
 
 
+(defn run-all-targets [target-fns config]
+  (let [threads (map #(async/thread (% config)) target-fns)]
+    (async/<!! (async/merge threads))))
+
+
 (defn -main [& argv]
   (let [config (-> "config.edn" clojure.java.io/resource read-config)
-        target (-> argv first keyword)
-        target-fn (get apps target)]
-    (if target-fn
-      (apply target-fn [config])
-      (println "target" target "not found"))))
+        xf (comp keyword? apps)
+        target-fns (sequence xf (rest argv))]
+    (if (seq target-fns)
+      (run-all-targets target-fns config)
+      (do
+        (println "target" target-fns "not found")
+        (println "all known targets are...")
+        (println (keys apps))))))

src/skirmish/exchange/core.clj

@@ -0,0 +1,17 @@
+(ns skirmish.exchange.core
+  (:require [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)]
+    (create-exchange-for-security initial-book x)))
+
+(defn -main [config]
+  (let [exchange-apis (create-exchange-apis-for config)]
+    (doseq [{sec-id :sec-id} (:securities exchange-apis)]
+      (create-exchange-for (get exchange-apis sec-id)))))
+  ;      ;api-routes (create-routes exchange-apis kafka-producer)]));;

src/skirmish/exchange/exchange.clj

@@ -79,7 +79,7 @@
 
 ;; bellow are the really interesting functions.
 
-(defn create-exchange-for-security [initial-book incoming-order-channel]
+(defn create-exchange-for-security [initial-book incoming-order-channel trades-channel outcoming-orders-channel]
   ;; output - trades channel, pub channel
   ;;        - orders channel, pub channel
   (let [trades-channel (chan)

src/skirmish/kafka_exchange.clj

@@ -1,41 +0,0 @@
-(ns skirmish.kafka-exchange
-  (:require [kafka-streams-clojure.api :as api])
-  (:import [org.apache.kafka.clients.producer KafkaProducer ProducerRecord]
-           [org.apache.kafka.streams StreamsConfig KafkaStreams]
-           [org.apache.kafka.streams.kstream KStreamBuilder]))
-
-
-(defn -main []
-  (def xform (comp (filter (fn [[k v]] (string? v)))
-                   (map (fn [[k v]] [v k]))
-                   (filter (fn [[k v]] (= "foo" v)))))
-
-  (def builder (KStreamBuilder.))
-
-  (def kstream (-> builder
-                   (.stream (into-array String ["tset"]))
-                   (api/transduce-kstream xform)
-                   (.to "test")))
-
-  (def kafka-streams
-    (KafkaStreams. builder
-                   (StreamsConfig. {StreamsConfig/APPLICATION_ID_CONFIG    "test-app-id"
-                                    StreamsConfig/BOOTSTRAP_SERVERS_CONFIG "localhost:9092"
-                                    StreamsConfig/KEY_SERDE_CLASS_CONFIG   org.apache.kafka.common.serialization.Serdes$StringSerde
-                                    StreamsConfig/VALUE_SERDE_CLASS_CONFIG org.apache.kafka.common.serialization.Serdes$StringSerde})))
-  (.start kafka-streams)
-
-  (def producer (KafkaProducer. {"bootstrap.servers" "localhost:9092"
-                                 "acks"              "all"
-                                 "retries"           "0"
-                                 "key.serializer"    "org.apache.kafka.common.serialization.StringSerializer"
-                                 "value.serializer"  "org.apache.kafka.common.serialization.StringSerializer"}))
-
-  @(.send producer (ProducerRecord. "tset" "foo" "bar"))
-  ;; Observe message come across topic "test" via kafka-console-consumer
-
-  @(.send producer (ProducerRecord. "tset" "baz" "quux"))
-  ;; Observe message does not come across topic "test" via kafka-console-consumer
-
-  (.close producer)
-  (.close kafka-streams))

src/skirmish/middleware/core.clj

@@ -0,0 +1,90 @@
+(ns skirmish.middleware.core
+  (:require [clojure.core.async :as async]
+            [skirmish.exchange.exchange :refer [empty-book create-exchange-for-security]] ;; remove this import.
+            [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
+     :trades-publisher-bis   trades
+     :trades-register        trades
+     :orders-publisher       orders
+     :orders-register        orders}))
+
+(def api-creators {:kafka create-kafka-based-exchange-api-for})
+
+(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)
+        f (fn [sec] [(:sec-id sec) (create-exchange-api-for (:sec-id sec))])
+        ret (into {} (map f security-definitions))]
+    (set-middleware ret config)))
+;;

src/skirmish/rest/core.clj

@@ -1,17 +1,42 @@
 (ns skirmish.rest.core
   (:require [org.httpkit.server :refer [run-server]]
             [compojure.handler :refer [api]]
+            [clojure.core.async :as async]
             [kinsky.client]
-            [skirmish.rest.exchange :refer [create-exchanges-api-for
-                                            create-routes]]))
+            [kinsky.async]
+            [skirmish.middleware.core :refer [create-exchange-apis-for]]
+            [skirmish.rest.exchange :refer [create-routes]]))
+
+
+(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 api-main [config]
   (let [webserver-config (:webserver config)
-        exchange-apis (create-exchanges-api-for (:securities config))
-        kafka-producer (kinsky.client/producer {:bootstrap.servers (get-in config [:kafka :bootstrap.servers])}
-                                               (kinsky.client/keyword-serializer)
-                                               (kinsky.client/edn-serializer))
-        api-routes (create-routes exchange-apis kafka-producer)]
+        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)))
+    (run-server (api api-routes) webserver-config)))
+
+
+(defn bookeeper-main [config]
+  (println "not yet implemented"))

src/skirmish/rest/exchange.clj

@@ -30,20 +30,6 @@
 ;;
 
 
-(defn create-exchange-api-for [sec-id]
-  (let [initial-book empty-book
-        incoming-order-channel (chan)
-        [trades-publisher new-book-order-publisher] (create-exchange-for-security initial-book incoming-order-channel)]
-    {:sec-id                 sec-id
-     :incoming-order-channel incoming-order-channel
-     :trades-publisher         trades-publisher
-     :new-book-order-publisher new-book-order-publisher}))
-
-(defn create-exchanges-api-for [security-definitions]
-  (let [f (fn [sec] [(:sec-id sec) (create-exchange-api-for (:sec-id sec))])]
-    (into {} (map f security-definitions))))
-
-
 (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)
@@ -97,7 +83,7 @@
     (throw+ {:error-type :4XX :status 404 :body {:error "Security not available."}})))
 
 
-(defn create-order [exchange-apis kafka-producer]
+(defn create-order [exchange-apis]
   (fn [{{sec-id-str :sec-id} :route-params body :body :as request}]
     (try+
       (let [asked-order (get-body body)
@@ -117,25 +103,28 @@
       (send! ws-chan (build-response-body request e))
       (recur))))
 
+
 (defn listen-to-orders [exchange-apis]
   (fn [{{sec-id-str :sec-id} :route-params :as request}]
     (with-channel request ws-chan
       (let [order-channel (get-new-order-channel exchange-apis (keyword sec-id-str))]
         (chan-to-websocket request order-channel ws-chan)))))
 
+
 (defn listen-to-trades [exchange-apis]
   (fn [{{sec-id-str :sec-id} :route-params :as request}]
     (with-channel request ws-chan
                   (let [order-channel (get-new-trades-channel exchange-apis (keyword sec-id-str))]
                     (chan-to-websocket request order-channel ws-chan)))))
 
-(defn create-routes [exchange-apis kafka-producer]
+
+(defn create-routes [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 kafka-producer))
+      (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/middleware/test_core.clj

@@ -0,0 +1,4 @@
+(ns skirmish.middleware.test-core
+  (:require [midje.sweet :refer :all]
+            [skirmish.middleware.core :refer :all]))
+