pedestal async solution

hello.clj

(ns hello
  (:require [io.pedestal.http :as http]
            [io.pedestal.http.route :as route]
            [clojure.core.async :as async]
            [cheshire.core :as json]))

; deps.edn:
; {:paths ["src"]
; :deps  {io.pedestal/pedestal.service {:mvn/version "0.5.8"}
;         io.pedestal/pedestal.route   {:mvn/version "0.5.8"}
;         io.pedestal/pedestal.jetty   {:mvn/version "0.5.8"}
;         org.slf4j/slf4j-simple       {:mvn/version "1.7.28"}}}

(defn json-array-chan [c]
  (let [out (async/chan)]
    (async/go
      (async/>! out "[")
      (loop [first true
             v (async/<! c)]
        (when v
          (when-not first
            (async/>! out ","))
          (async/>! out (json/generate-string v))
          (recur false (async/<! c))))
      (async/>! out "]")
      (async/close! out))
    out))

(defn respond-async [request]
  (let [c (async/chan)]
    (async/go
      (doseq [e (range 10)] ; also works with for example 1e9
        (async/>! c {:foo e}))
      (async/close! c))
    {:status  200
     :headers {"Content-Type" "application/json;charset=UTF-8"}
     :body    (json-array-chan c)}))

(def routes
  (route/expand-routes
    #{["/async" :get respond-async :route-name :async]}))

(comment
  ; works as expected:
  (let [c (:body (respond-async nil))]
    (async/go-loop [v (async/<! c)]
      (if v
        (do (println "got value" v)
            (recur (async/<! c)))
        (println "done!")))))

(defn create-server []
  (http/create-server
    {::http/routes routes
     ::http/type   :jetty
     ::http/join?  false
     ::http/port   8890}))

(defonce server (atom nil))

(defn start []
  (swap! server
         (fn [old-server]
           (when old-server
             (http/stop old-server))
           (http/start (create-server)))))
;(start)
;[nRepl-session-a69fe87e-a2fb-47f4-b0dc-f21df1bf9530] INFO org.eclipse.jetty.server.AbstractConnector - Stopped ServerConnector@63697192{HTTP/1.1,[http/1.1, h2c]}{localhost:8890}
;[nRepl-session-a69fe87e-a2fb-47f4-b0dc-f21df1bf9530] INFO org.eclipse.jetty.server.handler.ContextHandler - Stopped o.e.j.s.ServletContextHandler@6edfd0f9{/,null,UNAVAILABLE}
;[nRepl-session-a69fe87e-a2fb-47f4-b0dc-f21df1bf9530] INFO org.eclipse.jetty.server.Server - jetty-9.4.18.v20190429; built: 2019-04-29T20:42:08.989Z; git: e1bc35120a6617ee3df052294e433f3a25ce7097; jvm 11.0.5+10
;[nRepl-session-a69fe87e-a2fb-47f4-b0dc-f21df1bf9530] INFO org.eclipse.jetty.server.handler.ContextHandler - Started o.e.j.s.ServletContextHandler@446e368e{/,null,AVAILABLE}
;[nRepl-session-a69fe87e-a2fb-47f4-b0dc-f21df1bf9530] INFO org.eclipse.jetty.server.AbstractConnector - Started ServerConnector@51525501{HTTP/1.1,[http/1.1, h2c]}{localhost:8890}
;[nRepl-session-a69fe87e-a2fb-47f4-b0dc-f21df1bf9530] INFO org.eclipse.jetty.server.Server - Started @155746ms
;=> #:io.pedestal.http{:routes ({:path "/async", :method :get, :path-re #"/\Qasync\E", :path-parts ["async"], :interceptors [#Interceptor{:name }], :route-name :async, :path-params []}), :server #object[org.eclipse.jetty.server.Server 0x5ea8069b "Server@5ea8069b{STARTED}[9.4.18.v20190429]"], :stop-fn #object[io.pedestal.http.jetty$server$fn__17479 0xde6825e "io.pedestal.http.jetty$server$fn__17479@de6825e"], :type :jetty, :port 8890, :servlet #object[io.pedestal.http.servlet.FnServlet 0x586fb3ae "io.pedestal.http.servlet.FnServlet@586fb3ae"], :host "localhost", :join? false, :service-fn #object[io.pedestal.http.impl.servlet_interceptor$interceptor_service_fn$fn__17030 0x6dfc2902 "io.pedestal.http.impl.servlet_interceptor$interceptor_service_fn$fn__17030@6dfc2902"], :interceptors [#Interceptor{:name :io.pedestal.http/log-request} #Interceptor{:name :io.pedestal.http/not-found} #Interceptor{:name :io.pedestal.http.ring-middlewares/content-type-interceptor} #Interceptor{:name :io.pedestal.http.route/query-params} #Interceptor{:name :io.pedestal.http.route/path-params-decoder} #Interceptor{:name :io.pedestal.http.route/method-param} #Interceptor{:name :io.pedestal.http.secure-headers/secure-headers} #Interceptor{:name :io.pedestal.http.route/router}], :start-fn #object[io.pedestal.http.jetty$server$fn__17477 0x5893b875 "io.pedestal.http.jetty$server$fn__17477@5893b875"]}

; $ curl http://localhost:8890/async
; => [{"foo":0},{"foo":1},{"foo":2},{"foo":3},{"foo":4},{"foo":5},{"foo":6},{"foo":7},{"foo":8},{"foo":9}]

Case 1: stream of json's on reponse

• use transducers
• use pedestal async.chan respons

(let [handler (fn [req]
                (let [c (async/chan 10 (map json/write-str))]
                  (async/go
                    (async/>! c {:foo 1})
                    (async/>! c {:bar 2})
                    (async/close! c))
                  {:status  200
                   :headers {"Content-Type" "application/json"}
                   :body    c}))
      routes #{["/" :get handler
                :route-name ::hello]}
      service-map {::http/routes routes}
      service-fn (-> service-map
                     http/default-interceptors
                     http/create-servlet
                     ::http/service-fn)]
  (response-for service-fn :get "/"))
;; =>
;; {:status 200,
;;  :body "{\"foo\":1}{\"bar\":2}",
;;  :headers {"Strict-Transport-Security" "max-age=31536000; includeSubdomains",
;;            "X-Frame-Options" "DENY",
;;            "X-Content-Type-Options" "nosniff",
;;            "X-XSS-Protection" "1; mode=block",
;;            "X-Download-Options" "noopen",
;;            "X-Permitted-Cross-Domain-Policies" "none",
;;            "Content-Security-Policy" "object-src 'none'; script-src 'unsafe-inline' 'unsafe-eval' 'strict-dynamic' https: http:;",
;;            "Content-Type" "application/json"}}

Case 2: a huge JSON list that do not fit in memory

• trust on lazyness (no async required)
• be sure that your JSON writer will write directly on "output", not on a "buffer"

(letfn [(chan->lazy-list! [c]
          (lazy-seq
            (let [v (async/<!! c)]
              (if (nil? v)
                []
                (cons v (chan->lazy-list! c))))))
        (handler [req]
          (let [c (async/chan)]
            (async/go
              (async/>! c {:foo 1})
              (async/>! c {:bar 2})
              (async/close! c))
            {:status  200
             :headers {"Content-Type" "application/json"}
             :body    (fn [w]
                        (with-open [w (io/writer w)]
                          (json/write (chan->lazy-list! c)
                                      w)))}))]
  (let [routes #{["/" :get handler
                  :route-name ::hello]}
        service-map {::http/routes routes}
        service-fn (-> service-map
                       http/default-interceptors
                       http/create-servlet
                       ::http/service-fn)]
    (response-for service-fn :get "/")))
;; =>
;; {:status 200,
;;  :body "[{\"foo\":1},{\"bar\":2}]",
;;  :headers {"Strict-Transport-Security" "max-age=31536000; includeSubdomains",
;;            "X-Frame-Options" "DENY",
;;            "X-Content-Type-Options" "nosniff",
;;            "X-XSS-Protection" "1; mode=block",
;;            "X-Download-Options" "noopen",
;;            "X-Permitted-Cross-Domain-Policies" "none",
;;            "Content-Security-Policy" "object-src 'none'; script-src 'unsafe-inline' 'unsafe-eval' 'strict-dynamic' https: http:;",
;;            "Content-Type" "application/json"}}