Move HTML rendering to btcprice, send raw JSON over both WebSockets
Price and Strike WebSockets now emit JSON instead of HTML fragments. Added Strike WebSocket endpoint and wired strike/ticker into API routes. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -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
|
||||
|
||||
@@ -37,27 +37,36 @@
|
||||
(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")))))))
|
||||
|
||||
@@ -66,9 +75,11 @@
|
||||
(log/info "Starting Strike ticker poller")
|
||||
(let [client (HttpClient/newHttpClient)
|
||||
running? (atom true)
|
||||
fut (start-poll-loop! client api-key query-fn running?)]
|
||||
latest-rates (atom nil)
|
||||
fut (start-poll-loop! client api-key query-fn running? latest-rates)]
|
||||
{:running? running?
|
||||
:future fut}))
|
||||
:future fut
|
||||
:latest-rates latest-rates}))
|
||||
|
||||
(defmethod ig/halt-key! :strike/ticker
|
||||
[_ {:keys [running? future]}]
|
||||
|
||||
@@ -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 "<div id=\"price-display\">"
|
||||
"<div class=\"bg-white rounded-2xl shadow-sm border border-gray-200 p-6\">"
|
||||
"<div class=\"text-center\">"
|
||||
"<p class=\"text-4xl font-bold " color "\">$" (da-fmt "%,.2f" p) "</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 "<div class=\"flex justify-center gap-4 mt-3\">"
|
||||
"<span class=\"text-2xl font-bold text-gray-600\">\u20AC" (da-fmt "%,.0f" p-eur) "</span>"
|
||||
"<span class=\"text-2xl font-bold text-gray-600\">" (da-fmt "%,.0f" p-dkk) " kr</span>"
|
||||
"</div>"
|
||||
"<div class=\"flex justify-center gap-4 mt-1\">"
|
||||
"<span class=\"text-xs text-gray-400\">EUR/USD " (da-fmt "%.4f" eur) "</span>"
|
||||
"<span class=\"text-xs text-gray-400\">DKK/USD " (da-fmt "%.4f" dkk) "</span>"
|
||||
"</div>")))
|
||||
"<p class=\"text-xs text-gray-400 mt-2\">Updated " ts "</p>"
|
||||
"</div></div></div>")))
|
||||
|
||||
(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
|
||||
|
||||
Reference in New Issue
Block a user