Use HTMX-compatible HTML fragments on WebSocket, fix multi-connection
- WebSocket sends HTML fragments with id="price-display" instead of JSON for direct HTMX OOB swap on the client - Switch from ring-undertow-adapter ws wrapper to raw Undertow WebSocket API (proxy) to fix single-connection limitation - Keep JSON endpoint at /api/price/latest unchanged - Update CLAUDE.md and README.md Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -44,7 +44,7 @@ Docker runs alongside local dev on different ports and databases:
|
||||
## API Endpoints
|
||||
|
||||
- `GET /api/health` — Health check
|
||||
- `GET /api/price/ws` — WebSocket endpoint for live JSON price updates
|
||||
- `GET /api/price/ws` — WebSocket endpoint for live HTML price fragments (HTMX-compatible)
|
||||
- `GET /api/price/latest` — Latest price as JSON
|
||||
|
||||
## Source Layout
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
# btcdata
|
||||
|
||||
Bitcoin data backend service. Fetches BTC prices from Binance (WebSocket) and Kraken (HTTP polling), stores them in PostgreSQL, and exposes a JSON API with Server-Sent Events for live price streaming.
|
||||
Bitcoin data backend service. Fetches BTC prices from Binance (WebSocket) and Kraken (HTTP polling), stores them in PostgreSQL, and exposes a JSON API with WebSocket for live price streaming.
|
||||
|
||||
## API
|
||||
|
||||
| Endpoint | Description |
|
||||
|---|---|
|
||||
| `GET /api/price/stream` | SSE stream — emits `price-update` events with JSON `{price, prev_price, recorded_at}` |
|
||||
| `GET /api/price/ws` | WebSocket — sends HTML fragments with live price updates (HTMX-compatible) |
|
||||
| `GET /api/price/latest` | Latest price as JSON |
|
||||
| `GET /api/health` | Health check |
|
||||
|
||||
|
||||
@@ -9,11 +9,13 @@
|
||||
[reitit.ring.coercion :as coercion]
|
||||
[reitit.ring.middleware.muuntaja :as muuntaja]
|
||||
[reitit.ring.middleware.parameters :as parameters]
|
||||
[reitit.swagger :as swagger]
|
||||
[ring.adapter.undertow.websocket :as ws])
|
||||
[reitit.swagger :as swagger])
|
||||
(:import
|
||||
[io.undertow.server HttpServerExchange]
|
||||
[io.undertow.websockets.core WebSockets WebSocketChannel]
|
||||
[io.undertow.websockets WebSocketConnectionCallback WebSocketProtocolHandshakeHandler]
|
||||
[io.undertow.websockets.core AbstractReceiveListener WebSockets WebSocketChannel]
|
||||
[java.time Instant ZoneId]
|
||||
[java.time.format DateTimeFormatter]
|
||||
[java.util.concurrent LinkedBlockingQueue TimeUnit]))
|
||||
|
||||
(defn- price->json [{:keys [price prev-price recorded-at]}]
|
||||
@@ -22,15 +24,30 @@
|
||||
:prev_price (when prev-price (str prev-price))
|
||||
:recorded_at (str recorded-at)}))
|
||||
|
||||
(defn- ws-price-handler [{:keys [binance]} req]
|
||||
(let [^HttpServerExchange exchange (:server-exchange req)
|
||||
price-atom (:latest-price binance)]
|
||||
(ws/ws-request exchange nil
|
||||
(ws/ws-callback
|
||||
{:on-open
|
||||
(fn [{:keys [^WebSocketChannel channel]}]
|
||||
(def ^:private ts-fmt (DateTimeFormatter/ofPattern "dd-MM HH:mm:ss"))
|
||||
|
||||
(defn- price->html [{:keys [price prev-price recorded-at]}]
|
||||
(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 "\">$" (format "%,.2f" p) "</p>"
|
||||
"<p class=\"text-xs text-gray-400 mt-2\">Updated " ts "</p>"
|
||||
"</div></div></div>")))
|
||||
|
||||
(defn- start-ws-send-loop! [^WebSocketChannel channel price-atom]
|
||||
(let [queue (LinkedBlockingQueue.)
|
||||
wkey (keyword (gensym "ws-"))]
|
||||
(.set (.getReceiveSetter channel)
|
||||
(proxy [AbstractReceiveListener] []))
|
||||
(.resumeReceives channel)
|
||||
(add-watch price-atom wkey
|
||||
(fn [_ _ _ v] (.offer queue v)))
|
||||
(when-let [v @price-atom]
|
||||
@@ -40,14 +57,23 @@
|
||||
(loop []
|
||||
(when (.isOpen channel)
|
||||
(if-let [v (.poll queue 15 TimeUnit/SECONDS)]
|
||||
(WebSockets/sendTextBlocking (price->json v) channel)
|
||||
(WebSockets/sendTextBlocking "{\"ping\":true}" channel))
|
||||
(WebSockets/sendTextBlocking (price->html v) channel)
|
||||
(WebSockets/sendTextBlocking "" channel))
|
||||
(recur)))
|
||||
(catch Exception _)
|
||||
(finally
|
||||
(remove-watch price-atom wkey)
|
||||
(when (.isOpen channel)
|
||||
(try (.close channel) (catch Exception _))))))))}))
|
||||
(try (.close channel) (catch Exception _))))))))
|
||||
|
||||
(defn- ws-price-handler [{:keys [binance]} req]
|
||||
(let [^HttpServerExchange exchange (:server-exchange req)
|
||||
price-atom (:latest-price binance)
|
||||
callback (proxy [WebSocketConnectionCallback] []
|
||||
(onConnect [_ws-exchange channel]
|
||||
(start-ws-send-loop! channel price-atom)))
|
||||
handler (WebSocketProtocolHandshakeHandler. callback)]
|
||||
(.handleRequest handler exchange)
|
||||
nil))
|
||||
|
||||
(defn- latest-price-handler [{:keys [binance]} _req]
|
||||
|
||||
Reference in New Issue
Block a user