started to play with kafka. hit a solid wall

src/skirmish/core.clj

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

src/skirmish/kafka/book.clj

@@ -0,0 +1 @@
+(ns skirmish.kafka.book)

src/skirmish/kafka/exchange_simulator.clj

@@ -0,0 +1,22 @@
+(ns skirmish.kafka.exchange-simulator
+  (:require [kinsky.client :as client]))
+
+(def mock-kafka-messages
+  [[:sell-1 {:action :received            :order {:type :sell :quantity 10 :price 10 :book.order/id :sell-1}}]
+   [:sell-1 {:action :inserted-in-book    :order {:type :sell :quantity 10 :price 10 :book.order/id :sell-1}}]
+   [:sell-2 {:action :received            :order {:type :sell :quantity 5  :price 9 :book.order/id :sell-2}}]
+   [:sell-2 {:action :inserted-in-book    :order {:type :sell :quantity 5  :price 9 :book.order/id :sell-2}}]
+   [:buy-1  {:action :received            :order {:type :buy  :quantity 7  :price 9 :book.order/id :buy-1}}]
+   [:buy-1  {:action :executed-before     :order {:type :buy  :quantity 7  :price 9 :book.order/id :buy-1}}]])
+
+
+
+
+(defn -main [config]
+  (let [p (client/producer {:bootstrap.servers (-> config :kafka :bootstrap.servers)}
+                           (client/keyword-serializer)
+                           (client/edn-serializer))]
+    (doseq [[key val] mock-kafka-messages]
+      @(client/send! p "account" key val)
+      (println "<<>>" key "--" val)))
+  (print "finished"))

src/skirmish/kafka_exchange.clj

@@ -0,0 +1,41 @@
+(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/rest/core.clj

@@ -1,11 +1,17 @@
 (ns skirmish.rest.core
   (:require [org.httpkit.server :refer [run-server]]
             [compojure.handler :refer [api]]
+            [kinsky.client]
             [skirmish.rest.exchange :refer [create-exchanges-api-for
                                             create-routes]]))
 
 (defn -main [config]
   (let [webserver-config (:webserver config)
         exchange-apis (create-exchanges-api-for (:securities config))
-        api-routes (create-routes exchange-apis)]
+        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)]
+    (println "Running server with bellow config")
+    (println webserver-config)
     (run-server (api api-routes) webserver-config)))

src/skirmish/rest/exchange.clj

@@ -8,6 +8,7 @@
             [skirmish.exchange :refer [create-exchange-for-security]]
             [skirmish.rest.util :refer [build-response build-response-body]]
             [clojure.spec.alpha :as s]
+            [kinsky.client]
             [slingshot.slingshot :refer [try+ throw+]]
             [clojure.core.async :refer [chan go go-loop close! <! >!! sub]]
             [clojure.edn :as edn]))
@@ -97,14 +98,17 @@
     (throw+ {:error-type :4XX :status 404 :body {:error "Security not available."}})))
 
 
-(defn create-order [exchange-apis]
+(defn create-order [exchange-apis kafka-producer]
   (fn [{{sec-id-str :sec-id} :route-params body :body :as request}]
     (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)
             order (validate-order asked-order (:sec-id exchange-api))]
-        (>!! incoming-order-channel order)
+        @(kinsky.client/send! kafka-producer
+                              (str "new-orders-" sec-id-str)
+                              (:book.order/id order)
+                              order)
         (build-response request order))
       (catch [:error-type :4XX] e (build-response request e))
       (catch [:error-type :5XX] e (build-response request e)))))
@@ -129,13 +133,13 @@
                   (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]
+(defn create-routes [exchange-apis kafka-producer]
   (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))
+      (POST "/securities/:sec-id" [] (create-order exchange-apis kafka-producer))
       (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/rest/marketdata.clj

@@ -1,13 +1 @@
-(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")))
+(ns skirmish.rest.marketdata)