diff --git a/CLAUDE.md b/CLAUDE.md index 2d243bf..d537d13 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,6 +1,6 @@ # 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 @@ -44,7 +44,7 @@ Docker runs alongside local dev on different ports and databases: ## API Endpoints - `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 ## Source Layout @@ -67,6 +67,6 @@ src/clj/pmagnus/btcdata/ │ ├── exception.clj # Exception handling │ └── formats.clj # Content negotiation └── routes/ - ├── api.clj # /api routes (JSON + SSE) + ├── api.clj # /api routes (JSON + WebSocket) └── utils.clj # Route utilities ``` diff --git a/Dockerfile b/Dockerfile index 37232ce..2b1a271 100644 --- a/Dockerfile +++ b/Dockerfile @@ -10,8 +10,8 @@ FROM eclipse-temurin:21-jre-alpine COPY --from=build /build/target/btcdata-standalone.jar /btcdata/btcdata-standalone.jar -EXPOSE 4100 -ENV BTCDATA_PORT=4100 +EXPOSE 4101 +ENV BTCDATA_PORT=4101 ENV CORS_ORIGIN=http://localhost:4000 ENV JDBC_URL=jdbc:postgresql://postgres:5432/btcprod diff --git a/docker-compose.yml b/docker-compose.yml index 440888b..8418d85 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -3,9 +3,9 @@ services: build: . restart: unless-stopped ports: - - "4101:4100" + - "4101:4101" environment: - BTCDATA_PORT: "4100" + BTCDATA_PORT: "4101" JDBC_URL: "jdbc:postgresql://postgres:5432/btcprod?user=postgres&password=ratata,123" CORS_ORIGIN: "http://localhost:4041" extra_hosts: diff --git a/free-port.sh b/free-port.sh new file mode 100755 index 0000000..858476c --- /dev/null +++ b/free-port.sh @@ -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 diff --git a/src/clj/pmagnus/btcdata/web/routes/api.clj b/src/clj/pmagnus/btcdata/web/routes/api.clj index d09d767..e336e95 100644 --- a/src/clj/pmagnus/btcdata/web/routes/api.clj +++ b/src/clj/pmagnus/btcdata/web/routes/api.clj @@ -1,7 +1,6 @@ (ns pmagnus.btcdata.web.routes.api (:require [clojure.data.json :as json] - [clojure.string :as str] [integrant.core :as ig] [pmagnus.btcdata.web.controllers.health :as health] [pmagnus.btcdata.web.middleware.exception :as exception] @@ -10,67 +9,45 @@ [reitit.ring.coercion :as coercion] [reitit.ring.middleware.muuntaja :as muuntaja] [reitit.ring.middleware.parameters :as parameters] - [reitit.swagger :as swagger]) + [reitit.swagger :as swagger] + [ring.adapter.undertow.websocket :as ws]) (:import [io.undertow.server HttpServerExchange] - [io.undertow.util HttpString] - [java.io BufferedWriter OutputStreamWriter] + [io.undertow.websockets.core WebSockets WebSocketChannel] [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]}] (json/write-str {:price (str price) :prev_price (when prev-price (str prev-price)) :recorded_at (str recorded-at)})) -(defn- run-sse-loop! [^HttpServerExchange exchange price-atom cors-origin] - (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] +(defn- ws-price-handler [{:keys [binance]} req] (let [^HttpServerExchange exchange (:server-exchange req) - price-atom (:latest-price binance) - cors-origin (or (System/getenv "CORS_ORIGIN") "*")] - ;; Dispatch to a worker thread so the Ring adapter never processes - ;; the return value — Undertow handles the exchange lifecycle directly. - (.dispatch exchange - (reify io.undertow.server.HttpHandler - (handleRequest [_ ex] - (run-sse-loop! ex price-atom cors-origin)))) - ;; Return nil; exchange is dispatched so adapter won't touch it. + price-atom (:latest-price binance)] + (ws/ws-request exchange nil + (ws/ws-callback + {:on-open + (fn [{:keys [^WebSocketChannel channel]}] + (let [queue (LinkedBlockingQueue.) + wkey (keyword (gensym "ws-"))] + (add-watch price-atom wkey + (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)) (defn- latest-price-handler [{:keys [binance]} _req] @@ -91,8 +68,8 @@ :handler (swagger/create-swagger-handler)}}] ["/health" {:get health/healthcheck!}] - ["/price/stream" - {:get (fn [req] (price-stream-handler opts req)) + ["/price/ws" + {:get (fn [req] (ws-price-handler opts req)) :no-doc true :middleware []}] ["/price/latest"