From 0ccbb6bc1a289e81b9d5c39b04ad7852005bdf24 Mon Sep 17 00:00:00 2001 From: didier amyot Date: Wed, 11 Jul 2018 19:41:01 -0400 Subject: [PATCH] initial commit --- .gitignore | 1 + resources/config.edn | 1 + src/skirmish/core.clj | 25 +++++-- src/skirmish/exchange/core.clj | 17 +++++ src/skirmish/exchange/exchange.clj | 2 +- src/skirmish/kafka_exchange.clj | 41 ------------ src/skirmish/middleware/core.clj | 90 ++++++++++++++++++++++++++ src/skirmish/rest/core.clj | 41 +++++++++--- src/skirmish/rest/exchange.clj | 23 ++----- test/skirmish/middleware/test_core.clj | 4 ++ 10 files changed, 171 insertions(+), 74 deletions(-) create mode 100644 src/skirmish/exchange/core.clj delete mode 100644 src/skirmish/kafka_exchange.clj create mode 100644 src/skirmish/middleware/core.clj create mode 100644 test/skirmish/middleware/test_core.clj diff --git a/.gitignore b/.gitignore index 1511d86..dae6af5 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,4 @@ pom.xml.asc book.iml skirmish.iml full-stack +fixtures diff --git a/resources/config.edn b/resources/config.edn index 9ecd30b..296a801 100644 --- a/resources/config.edn +++ b/resources/config.edn @@ -1,4 +1,5 @@ { + :middleware :kafka :kafka {:bootstrap.servers #or [#env KAFKA_BOOTSTRAP_SERVERS "localhost:9092"] :topics-prefix "skirmish-01-"} diff --git a/src/skirmish/core.clj b/src/skirmish/core.clj index 9b4a57c..b479c9e 100644 --- a/src/skirmish/core.clj +++ b/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/ "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")))) \ No newline at end of file + 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)))))) \ No newline at end of file diff --git a/src/skirmish/exchange/core.clj b/src/skirmish/exchange/core.clj new file mode 100644 index 0000000..20244cf --- /dev/null +++ b/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)]));; \ No newline at end of file diff --git a/src/skirmish/exchange/exchange.clj b/src/skirmish/exchange/exchange.clj index 72c1450..f99934a 100644 --- a/src/skirmish/exchange/exchange.clj +++ b/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) diff --git a/src/skirmish/kafka_exchange.clj b/src/skirmish/kafka_exchange.clj deleted file mode 100644 index f3bac69..0000000 --- a/src/skirmish/kafka_exchange.clj +++ /dev/null @@ -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)) \ No newline at end of file diff --git a/src/skirmish/middleware/core.clj b/src/skirmish/middleware/core.clj new file mode 100644 index 0000000..7b03f9a --- /dev/null +++ b/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))) +;; \ No newline at end of file diff --git a/src/skirmish/rest/core.clj b/src/skirmish/rest/core.clj index 5ed50b2..e489dab 100644 --- a/src/skirmish/rest/core.clj +++ b/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))) \ No newline at end of file + (run-server (api api-routes) webserver-config))) + + +(defn bookeeper-main [config] + (println "not yet implemented")) \ No newline at end of file diff --git a/src/skirmish/rest/exchange.clj b/src/skirmish/rest/exchange.clj index e35d67b..28931bb 100644 --- a/src/skirmish/rest/exchange.clj +++ b/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"))) diff --git a/test/skirmish/middleware/test_core.clj b/test/skirmish/middleware/test_core.clj new file mode 100644 index 0000000..d70557b --- /dev/null +++ b/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])) +