Replace SSE price stream with WebSocket, use separate Docker port

- Replace SSE endpoint (/api/price/stream) with WebSocket (/api/price/ws)
  using ring-undertow-adapter's built-in WebSocket support
- Change Docker internal port from 4100 to 4101 to avoid conflicts with
  local dev (make run still uses 4100)
- Add free-port.sh utility script

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-23 21:32:19 +01:00
co-authored by Claude Opus 4.6
parent 6f20cd63c6
commit 9283826ac4
5 changed files with 62 additions and 60 deletions
+3 -3
View File
@@ -1,6 +1,6 @@
# btcdata # btcdata
Bitcoin data backend — Clojure Kit application. Fetches BTC prices from Binance WebSocket and Kraken HTTP, stores in PostgreSQL, exposes a JSON API with SSE price stream. Bitcoin data backend — Clojure Kit application. Fetches BTC prices from Binance WebSocket and Kraken HTTP, stores in PostgreSQL, exposes a JSON API with WebSocket price stream.
## Build & Development Commands ## Build & Development Commands
@@ -44,7 +44,7 @@ Docker runs alongside local dev on different ports and databases:
## API Endpoints ## API Endpoints
- `GET /api/health` — Health check - `GET /api/health` — Health check
- `GET /api/price/stream` — SSE stream with JSON price updates - `GET /api/price/ws` — WebSocket endpoint for live JSON price updates
- `GET /api/price/latest` — Latest price as JSON - `GET /api/price/latest` — Latest price as JSON
## Source Layout ## Source Layout
@@ -67,6 +67,6 @@ src/clj/pmagnus/btcdata/
│ ├── exception.clj # Exception handling │ ├── exception.clj # Exception handling
│ └── formats.clj # Content negotiation │ └── formats.clj # Content negotiation
└── routes/ └── routes/
├── api.clj # /api routes (JSON + SSE) ├── api.clj # /api routes (JSON + WebSocket)
└── utils.clj # Route utilities └── utils.clj # Route utilities
``` ```
+2 -2
View File
@@ -10,8 +10,8 @@ FROM eclipse-temurin:21-jre-alpine
COPY --from=build /build/target/btcdata-standalone.jar /btcdata/btcdata-standalone.jar COPY --from=build /build/target/btcdata-standalone.jar /btcdata/btcdata-standalone.jar
EXPOSE 4100 EXPOSE 4101
ENV BTCDATA_PORT=4100 ENV BTCDATA_PORT=4101
ENV CORS_ORIGIN=http://localhost:4000 ENV CORS_ORIGIN=http://localhost:4000
ENV JDBC_URL=jdbc:postgresql://postgres:5432/btcprod ENV JDBC_URL=jdbc:postgresql://postgres:5432/btcprod
+2 -2
View File
@@ -3,9 +3,9 @@ services:
build: . build: .
restart: unless-stopped restart: unless-stopped
ports: ports:
- "4101:4100" - "4101:4101"
environment: environment:
BTCDATA_PORT: "4100" BTCDATA_PORT: "4101"
JDBC_URL: "jdbc:postgresql://postgres:5432/btcprod?user=postgres&password=ratata,123" JDBC_URL: "jdbc:postgresql://postgres:5432/btcprod?user=postgres&password=ratata,123"
CORS_ORIGIN: "http://localhost:4041" CORS_ORIGIN: "http://localhost:4041"
extra_hosts: extra_hosts:
Executable
+25
View File
@@ -0,0 +1,25 @@
#!/usr/bin/env bash
# Free port 4100 by stopping Docker containers and killing local processes
PORT=4100
# Stop Docker containers bound to the port
for cid in $(docker ps -q --filter "publish=$PORT" 2>/dev/null); do
name=$(docker inspect --format '{{.Name}}' "$cid" | sed 's|^/||')
echo "Stopping Docker container: $name ($cid)"
docker stop "$cid"
done
# Kill any local processes on the port
if command -v fuser &>/dev/null; then
fuser -k "$PORT/tcp" 2>/dev/null && echo "Killed local process on port $PORT"
fi
# Verify
sleep 1
if ss -tlnp sport = :"$PORT" 2>/dev/null | grep -q "$PORT"; then
echo "WARNING: port $PORT still in use"
ss -tlnp sport = :"$PORT"
exit 1
else
echo "Port $PORT is free"
fi
+30 -53
View File
@@ -1,7 +1,6 @@
(ns pmagnus.btcdata.web.routes.api (ns pmagnus.btcdata.web.routes.api
(:require (:require
[clojure.data.json :as json] [clojure.data.json :as json]
[clojure.string :as str]
[integrant.core :as ig] [integrant.core :as ig]
[pmagnus.btcdata.web.controllers.health :as health] [pmagnus.btcdata.web.controllers.health :as health]
[pmagnus.btcdata.web.middleware.exception :as exception] [pmagnus.btcdata.web.middleware.exception :as exception]
@@ -10,67 +9,45 @@
[reitit.ring.coercion :as coercion] [reitit.ring.coercion :as coercion]
[reitit.ring.middleware.muuntaja :as muuntaja] [reitit.ring.middleware.muuntaja :as muuntaja]
[reitit.ring.middleware.parameters :as parameters] [reitit.ring.middleware.parameters :as parameters]
[reitit.swagger :as swagger]) [reitit.swagger :as swagger]
[ring.adapter.undertow.websocket :as ws])
(:import (:import
[io.undertow.server HttpServerExchange] [io.undertow.server HttpServerExchange]
[io.undertow.util HttpString] [io.undertow.websockets.core WebSockets WebSocketChannel]
[java.io BufferedWriter OutputStreamWriter]
[java.util.concurrent LinkedBlockingQueue TimeUnit])) [java.util.concurrent LinkedBlockingQueue TimeUnit]))
(defn- format-sse [event data]
(str "event: " event "\n"
(str/join "\n" (map #(str "data: " %) (str/split-lines data)))
"\n\n"))
(defn- price->json [{:keys [price prev-price recorded-at]}] (defn- price->json [{:keys [price prev-price recorded-at]}]
(json/write-str (json/write-str
{:price (str price) {:price (str price)
:prev_price (when prev-price (str prev-price)) :prev_price (when prev-price (str prev-price))
:recorded_at (str recorded-at)})) :recorded_at (str recorded-at)}))
(defn- run-sse-loop! [^HttpServerExchange exchange price-atom cors-origin] (defn- ws-price-handler [{:keys [binance]} req]
(let [headers (.getResponseHeaders exchange)]
(.setStatusCode exchange 200)
(.put headers (HttpString. "Content-Type") "text/event-stream")
(.put headers (HttpString. "Cache-Control") "no-cache")
(.put headers (HttpString. "X-Accel-Buffering") "no")
(.put headers (HttpString. "Access-Control-Allow-Origin") cors-origin)
(when-not (.isBlocking exchange)
(.startBlocking exchange))
(let [out (.getOutputStream exchange)
writer (BufferedWriter. (OutputStreamWriter. out "UTF-8"))
queue (LinkedBlockingQueue.)
wkey (keyword (gensym "sse-"))]
(add-watch price-atom wkey
(fn [_ _ _ v] (.offer queue v)))
(when-let [v @price-atom]
(.offer queue v))
(try
(loop []
(if-let [v (.poll queue 15 TimeUnit/SECONDS)]
(do (.write writer (format-sse "price-update" (price->json v)))
(.flush writer))
(do (.write writer ": heartbeat\n\n")
(.flush writer)))
(recur))
(catch Exception _)
(finally
(remove-watch price-atom wkey)
(try (.close writer) (catch Exception _))
(when-not (.isComplete exchange)
(.endExchange exchange)))))))
(defn- price-stream-handler [{:keys [binance]} req]
(let [^HttpServerExchange exchange (:server-exchange req) (let [^HttpServerExchange exchange (:server-exchange req)
price-atom (:latest-price binance) price-atom (:latest-price binance)]
cors-origin (or (System/getenv "CORS_ORIGIN") "*")] (ws/ws-request exchange nil
;; Dispatch to a worker thread so the Ring adapter never processes (ws/ws-callback
;; the return value — Undertow handles the exchange lifecycle directly. {:on-open
(.dispatch exchange (fn [{:keys [^WebSocketChannel channel]}]
(reify io.undertow.server.HttpHandler (let [queue (LinkedBlockingQueue.)
(handleRequest [_ ex] wkey (keyword (gensym "ws-"))]
(run-sse-loop! ex price-atom cors-origin)))) (add-watch price-atom wkey
;; Return nil; exchange is dispatched so adapter won't touch it. (fn [_ _ _ v] (.offer queue v)))
(when-let [v @price-atom]
(.offer queue v))
(future
(try
(loop []
(when (.isOpen channel)
(if-let [v (.poll queue 15 TimeUnit/SECONDS)]
(WebSockets/sendTextBlocking (price->json v) channel)
(WebSockets/sendTextBlocking "{\"ping\":true}" channel))
(recur)))
(catch Exception _)
(finally
(remove-watch price-atom wkey)
(when (.isOpen channel)
(try (.close channel) (catch Exception _))))))))}))
nil)) nil))
(defn- latest-price-handler [{:keys [binance]} _req] (defn- latest-price-handler [{:keys [binance]} _req]
@@ -91,8 +68,8 @@
:handler (swagger/create-swagger-handler)}}] :handler (swagger/create-swagger-handler)}}]
["/health" ["/health"
{:get health/healthcheck!}] {:get health/healthcheck!}]
["/price/stream" ["/price/ws"
{:get (fn [req] (price-stream-handler opts req)) {:get (fn [req] (ws-price-handler opts req))
:no-doc true :no-doc true
:middleware []}] :middleware []}]
["/price/latest" ["/price/latest"