Clojure 网络服务深入：HTTP 客户端、WebSocket、SSE 与 gRPC

(require '[clj-http.client :as http])

;; 带超时、连接池、重定向控制的请求
(def resp
  (http/get "https://api.example.com/orders"
            {:query-params  {:page 1 :size 20}
             :headers       {"Authorization" "Bearer xxx"
                             "Accept"        "application/json"}
             :timeout       3000
             :conn-timeout  1000
             :socket-timeout 3000
             :as            :json
             :accept        :json}))

(:status resp)  ; => 200
(:body resp)    ; => 解析后的 JSON 数据

(defn request-with-retry [f & {:keys [retries delay] :or {retries 3 delay 100}}]
  (loop [n retries]
    (let [res (try (f) (catch Exception e (assoc {} :error e)))]
      (cond
        (and (contains? res :error) (> n 0))
        (do (Thread/sleep delay) (recur (dec n)))

        (and (contains? res :status)
             (>= (:status res) 500)          ; 5xx 可重试
             (> n 0))
        (do (Thread/sleep delay) (recur (dec n)))

        :else res))))

(request-with-retry #(http/get "https://api.example.com/health"))

(require '[org.httpkit.client :as http])

;; 异步发起请求，返回 promise/future，不阻塞线程
(def future-resp
  (http/get "https://api.example.com/orders"
            {:timeout 3000
             :query-params {:page 1}}))

;; 后续任意时刻取结果（可配合 channel）
(let [resp @future-resp]   ; deref 阻塞直到完成
  (:status resp))

(require '[clojure.core.async :as async])

;; BFF 场景：并行调用多个下游，聚合返回
(defn aggregate []
  (let [ch (async/chan 3)
        _  (async/go (async/>! ch (:body @(http/get url-a))))
        _  (async/go (async/>! ch (:body @(http/get url-b))))
        _  (async/go (async/>! ch (:body @(http/get url-c))))]
    {:a (async/<!! ch) :b (async/<!! ch) :c (async/<!! ch)}))

(require '[org.httpkit.server :as hk])

(defn handler [req] {:status 200 :body "hello"})
(hk/run-server handler {:port 8080 :threads 4 :queue-size 10000})

(require '[org.httpkit.server :as hk])

;; 全局连接注册表
(def connections (atom {}))

(defn ws-handler [req]
  (hk/with-channel req ch
    (hk/on-receive ch (fn [msg]
                        (let [parsed (json/parse-string msg true)]
                          (case (:op parsed)
                            :join (swap! connections assoc (:uid parsed) ch)
                            :say  (doseq [[_ c] @connections]
                                    (hk/send! c (:text parsed)))))))
    (hk/on-close ch (fn [_] (doseq [[uid c] @connections] (when (= c ch) (swap! connections dissoc uid)))))))

;; 路由：WebSocket 路径
(def routes {"/ws" ws-handler})

(require '[clj-websocket.client :as ws])

(def conn @(ws/connect "wss://example.com/ws"))
(ws/send-msg conn (json/write-str {:op :say :text "hello"}))

;; Ring handler 返回 text/event-stream，持续推送
(defn sse-handler [req]
  (hk/with-channel req ch
    (async/go-loop []
      (let [event (async/<! (event-chan))]
        (hk/send! ch (str "data: " (json/write-str event) "\n\n")))
      (recur))
    {:status 200
     :headers {"Content-Type" "text/event-stream"
               "Cache-Control" "no-cache"}}))

;; 定义 proto → 生成 OrderServiceGrpc
(require '[OrderServiceGrpc :as grpc-stub])
(require '[io.grpc.ManagedChannelBuilder])

(def channel
  (.build (ManagedChannelBuilder/forAddress "order-svc" 9090)
          (.usePlaintext)))

;; 一元调用
(def stub (grpc-stub/newBlockingStub channel))
(def resp (.placeOrder stub
            (Order$PlaceOrderRequest/newBuilder
              (.setUserId ...) (.setAmount ...) (.build))))

;; 流式调用：请求/响应的流式接口同样可用

;; 用 core.async 通道缓冲 + 背压
(def chan (async/chan 100))   ; 缓冲 100，满则 send! 阻塞/丢弃

;; 配合信号量限流，防止打垮下游
(def sem (java.util.concurrent.Semaphore. 50))
(defn throttled-call [f]
  (.acquire sem)
  (try (f) (finally (.release sem))))