diff --git a/resources/system.edn b/resources/system.edn index 57195cf..856b673 100644 --- a/resources/system.edn +++ b/resources/system.edn @@ -47,7 +47,8 @@ {:base-path "/api" :query-fn #ig/ref :db.sql/query-fn :binance #ig/ref :ws/binance - :frankfurter #ig/ref :frankfurter/rates} + :frankfurter #ig/ref :frankfurter/rates + :strike #ig/ref :strike/ticker} :ws/binance {:query-fn #ig/ref :db.sql/query-fn diff --git a/src/clj/pmagnus/btcdata/strike/ticker.clj b/src/clj/pmagnus/btcdata/strike/ticker.clj index c870426..ad44c93 100644 --- a/src/clj/pmagnus/btcdata/strike/ticker.clj +++ b/src/clj/pmagnus/btcdata/strike/ticker.clj @@ -37,38 +37,49 @@ (str sourceCurrency "/" targetCurrency " " amount))) (str/join " "))) +(defn- rates->map + "Index the rates array by sourceCurrency/targetCurrency pair." + [rates] + (into {} + (map (fn [{:keys [sourceCurrency targetCurrency amount]}] + [(str sourceCurrency "/" targetCurrency) amount]) + rates))) + (defn- poll! - "Fetch rates and insert into the database." - [client api-key query-fn] + "Fetch rates and insert into the database. Updates latest-atom." + [client api-key query-fn latest-atom] (let [body (fetch-rates client api-key) rates (json/read-str body :key-fn keyword)] (query-fn :insert-strike-price! {:rates (->jsonb body)}) + (reset! latest-atom (rates->map rates)) (log/info "Strike:" (format-rates rates)))) (defn- start-poll-loop! "Fetch immediately, then poll every 10 seconds." - [client api-key query-fn running?] + [client api-key query-fn running? latest-atom] (future (try - (poll! client api-key query-fn) + (poll! client api-key query-fn latest-atom) (catch Exception e (log/error e "Strike initial fetch failed"))) (while @running? (Thread/sleep 10000) (when @running? (try - (poll! client api-key query-fn) + (poll! client api-key query-fn latest-atom) (catch Exception e (log/error e "Strike poll failed"))))))) (defmethod ig/init-key :strike/ticker [_ {:keys [query-fn api-key]}] (log/info "Starting Strike ticker poller") - (let [client (HttpClient/newHttpClient) - running? (atom true) - fut (start-poll-loop! client api-key query-fn running?)] - {:running? running? - :future fut})) + (let [client (HttpClient/newHttpClient) + running? (atom true) + latest-rates (atom nil) + fut (start-poll-loop! client api-key query-fn running? latest-rates)] + {:running? running? + :future fut + :latest-rates latest-rates})) (defmethod ig/halt-key! :strike/ticker [_ {:keys [running? future]}] diff --git a/src/clj/pmagnus/btcdata/web/routes/api.clj b/src/clj/pmagnus/btcdata/web/routes/api.clj index 6c602ec..632a558 100644 --- a/src/clj/pmagnus/btcdata/web/routes/api.clj +++ b/src/clj/pmagnus/btcdata/web/routes/api.clj @@ -12,9 +12,6 @@ [reitit.swagger :as swagger]) (:import [io.undertow.websockets.core WebSockets WebSocketChannel] - [java.time Instant ZoneId] - [java.time.format DateTimeFormatter] - [java.util Locale] [java.util.concurrent LinkedBlockingQueue TimeUnit])) (defn- price->json [{:keys [price prev-price recorded-at]} rates] @@ -25,41 +22,6 @@ rates (assoc :price_eur (str (.multiply (bigdec price) (:eur rates))) :price_dkk (str (.multiply (bigdec price) (:dkk rates))))))) -(def ^:private ts-fmt (DateTimeFormatter/ofPattern "dd-MM HH:mm:ss")) -(def ^:private da-locale (Locale. "da" "DK")) - -(defn- da-fmt [fmt val] - (String/format da-locale fmt (into-array Object [val]))) - -(defn- price->html [{:keys [price prev-price recorded-at]} rates] - (let [p (double (bigdec price)) - pp (when prev-price (double (bigdec prev-price))) - color (cond - (nil? pp) "text-gray-900" - (> p pp) "text-green-600" - (< p pp) "text-red-600" - :else "text-gray-900") - ts (.format (.atZone (Instant/from recorded-at) (ZoneId/systemDefault)) ts-fmt)] - (str "
" - "
" - "
" - "

$" (da-fmt "%,.2f" p) "

" - (when rates - (let [eur (double (:eur rates)) - dkk (double (:dkk rates)) - p-eur (double (.multiply (bigdec price) (:eur rates))) - p-dkk (double (.multiply (bigdec price) (:dkk rates)))] - (str "
" - "\u20AC" (da-fmt "%,.0f" p-eur) "" - "" (da-fmt "%,.0f" p-dkk) " kr" - "
" - "
" - "EUR/USD " (da-fmt "%.4f" eur) "" - "DKK/USD " (da-fmt "%.4f" dkk) "" - "
"))) - "

Updated " ts "

" - "
"))) - (defn- start-ws-send-loop! [^WebSocketChannel channel price-atom rates-atom] (let [queue (LinkedBlockingQueue.) wkey (keyword (gensym "ws-")) @@ -76,7 +38,7 @@ (when (.isOpen channel) (if (.poll queue 15 TimeUnit/SECONDS) (when-let [v @price-atom] - (WebSockets/sendTextBlocking (price->html v @rates-atom) channel)) + (WebSockets/sendTextBlocking (price->json v @rates-atom) channel)) (WebSockets/sendTextBlocking "" channel)) (recur))) (catch Exception _) @@ -93,6 +55,34 @@ {:on-open (fn [{:keys [^WebSocketChannel channel]}] (start-ws-send-loop! channel price-atom rates-atom))}})) +(defn- start-strike-ws-loop! [^WebSocketChannel channel strike-atom] + (let [queue (LinkedBlockingQueue.) + wkey (keyword (gensym "ws-s-"))] + (add-watch strike-atom wkey + (fn [_ _ _ _] (.offer queue :update))) + (when @strike-atom + (.offer queue :update)) + (future + (try + (loop [] + (when (.isOpen channel) + (if (.poll queue 15 TimeUnit/SECONDS) + (when-let [rates @strike-atom] + (WebSockets/sendTextBlocking (json/write-str rates) channel)) + (WebSockets/sendTextBlocking "" channel)) + (recur))) + (catch Exception _) + (finally + (remove-watch strike-atom wkey) + (when (.isOpen channel) + (try (.close channel) (catch Exception _)))))))) + +(defn- ws-strike-handler [{:keys [strike]} _req] + (let [strike-atom (:latest-rates strike)] + {:undertow/websocket + {:on-open (fn [{:keys [^WebSocketChannel channel]}] + (start-strike-ws-loop! channel strike-atom))}})) + (defn- latest-price-handler [{:keys [binance frankfurter]} _req] (let [price-atom (:latest-price binance) rates-atom (:rates frankfurter) @@ -117,7 +107,11 @@ :no-doc true :middleware []}] ["/price/latest" - {:get (fn [req] (latest-price-handler opts req))}]]) + {:get (fn [req] (latest-price-handler opts req))}] + ["/strike/ws" + {:get (fn [req] (ws-strike-handler opts req)) + :no-doc true + :middleware []}]]) (defn route-data [opts] (merge