⬆️ Update bundled vertx.

backend/src/vertx/core.clj

@@ -208,8 +208,6 @@
   (let [opts (DeploymentOptions.)]
     (when instances (.setInstances opts (int instances)))
     (when worker (.setWorker opts worker))
-    ;; (.setInstances opts 4)
-    ;; (.setWorkerPoolSize opts 4)
     opts))
 
 (defn- opts->vertx-options

backend/src/vertx/http.clj

@@ -11,11 +11,15 @@
             [promesa.core :as p]
             [vertx.util :as vu])
   (:import
+   java.util.Map$Entry
+   clojure.lang.MapEntry
    io.vertx.core.Vertx
    io.vertx.core.Verticle
    io.vertx.core.Handler
    io.vertx.core.Future
+   io.vertx.core.MultiMap
    io.vertx.core.Context
+   io.vertx.core.buffer.Buffer
    io.vertx.core.http.HttpServer
    io.vertx.core.http.HttpServerRequest
    io.vertx.core.http.HttpServerResponse
@@ -26,7 +30,38 @@
 
 ;; --- Public Api
 
-(s/def :vertx.http/handler fn?)
+(declare -handle-response)
+(declare -handle-body)
+
+(defn ->headers
+  [^HttpServerRequest request]
+  (let [headers (.headers request)
+        it (.iterator ^MultiMap headers)]
+    (loop [m (transient {})]
+      (if (.hasNext it)
+        (let [^Map$Entry me (.next it)
+              key (.toLowerCase (.getKey me))
+              val (.getValue me)]
+          (recur (assoc! m key val)))
+        (persistent! m)))))
+
+(defn- ->request
+  [^HttpServerRequest request]
+  {:method (-> request .rawMethod .toLowerCase keyword)
+   :path (.path request)
+   :headers (->headers request)
+   ::request request
+   ::response (.response request)})
+
+(defn handler
+  [vsm f]
+  (reify Handler
+    (handle [this request]
+      (let [ctx (->request request)]
+        (-handle-response (f ctx) ctx)))))
+
+(s/def :vertx.http/handler
+  (s/or :fn fn? :handler #(instance? Handler %)))
 (s/def :vertx.http/host string?)
 (s/def :vertx.http/port pos?)
 (s/def ::server-options
@@ -54,21 +89,62 @@
   (let [opts (HttpServerOptions.)]
     (.setReuseAddress opts true)
     (.setReusePort opts true)
-    ;; (.setTcpNoDelay opts true)
+    (.setTcpNoDelay opts true)
-    ;; (.setTcpFastOpen opts true)
+    (.setTcpFastOpen opts true)
     (when host (.setHost opts host))
     (when port (.setPort opts port))
     opts))
 
-(defn- fn->handler
-  [f]
-  (reify Handler
-    (handle [_ request]
-      (f request))))
-
 (defn- resolve-handler
   [handler]
   (cond
-    (fn? handler) (fn->handler handler)
+    (fn? handler) (vu/fn->handler handler)
     (instance? Handler handler) handler
     :else (throw (ex-info "invalid handler" {}))))
+
+(defn- assign-status-and-headers!
+  [^HttpServerResponse res response]
+  (let [headers (:headers response)
+        status (:status response 200)]
+    (when (map? headers)
+      (vu/doseq [[key val] headers]
+        (.putHeader res ^String (name key) ^String (str val))))
+    (.setStatusCode res status)))
+
+(defprotocol IAsyncResponse
+  (-handle-response [_ _]))
+
+(defprotocol IAsyncBody
+  (-handle-body [_ _]))
+
+(extend-protocol IAsyncResponse
+  java.util.concurrent.CompletionStage
+  (-handle-response [data ctx]
+    (p/then' data #(-handle-response % ctx)))
+
+  clojure.lang.IPersistentMap
+  (-handle-response [data ctx]
+    (let [body (:body data)
+          res (::response ctx)]
+      (assign-status-and-headers! res data)
+      (-handle-body body res))))
+
+(extend-protocol IAsyncBody
+  (Class/forName "[B")
+  (-handle-body [data res]
+    (.end ^HttpServerResponse res (Buffer/buffer data)))
+
+  Buffer
+  (-handle-body [data res]
+    (.end ^HttpServerResponse res ^Buffer data))
+
+  nil
+  (-handle-body [data res]
+    (.putHeader ^HttpServerResponse res "content-length" "0")
+    (.end ^HttpServerResponse res))
+
+  String
+  (-handle-body [data res]
+    (let [length (count data)]
+      (.putHeader ^HttpServerResponse res "content-length" (str length))
+      (.end ^HttpServerResponse res data))))

backend/src/vertx/timers.clj

@@ -33,23 +33,3 @@
       java.lang.AutoCloseable
       (close [_]
         (.cancelTimer system timer-id)))))
-
-(defn schedule-task!
-  [vsm ms f]
-  (let [^Vertx system (vu/resolve-system vsm)
-        tid* (atom nil)
-        task (fn wrapped-task []
-               (-> (p/do! (f))
-                   (p/then (fn [_]
-                             (let [tid (schedule-task! vsm ms wrapped-task)]
-                               (reset! tid* tid)
-                               nil)))))
-        tid  (schedule-task! vsm ms task)]
-    (reset! tid* tid)
-    (reify
-      java.lang.AutoCloseable
-      (close [this]
-        (locking this
-          (when-let [timer-id (deref tid*)]
-            (.cancelTimer system timer-id)
-            (reset! tid* nil)))))))

backend/src/vertx/util.clj

@@ -5,6 +5,7 @@
 ;; Copyright (c) 2019 Andrey Antukh <niwi@niwi.nz>
 
 (ns vertx.util
+  (:refer-clojure :exclude [doseq])
   (:require [promesa.core :as p])
   (:import io.vertx.core.Vertx
            io.vertx.core.Handler
@@ -38,3 +39,13 @@
         (p/reject! d (.cause ar))
         (p/resolve! d (.result ar))))))
 
+(defmacro doseq
+  "A faster version of doseq."
+  [[bsym csym] & body]
+  `(let [it# (.iterator ~csym)]
+     (loop []
+       (when (.hasNext it#)
+         (let [~bsym (.next it#)]
+           ~@body
+           (recur))))))
+

backend/src/vertx/util/transit.clj

@@ -1,87 +0,0 @@
-;; This Source Code Form is subject to the terms of the Mozilla Public
-;; License, v. 2.0. If a copy of the MPL was not distributed with this
-;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
-;;
-;; Copyright (c) 2019 Andrey Antukh <niwi@niwi.nz>
-
-(ns vertx.util.transit
-  (:require [cognitect.transit :as t]
-            [clojure.java.io :as io])
-  (:import java.io.ByteArrayInputStream
-           java.io.ByteArrayOutputStream
-           java.time.Instant))
-
-(def ^:private write-handler
-  (t/write-handler
-   (constantly "m")
-   (fn [v] (str (.toEpochMilli v)))))
-
-(def ^:private read-handler
-  (t/read-handler
-   (fn [v] (-> (Long/parseLong v)
-               (Instant/ofEpochMilli)))))
-
-(def +read-handlers+
-  {"m" read-handler})
-
-(def +write-handlers+
-  {Instant write-handler})
-
-(defmethod print-method Instant
-  [mv ^java.io.Writer writer]
-  (.write writer (str "#instant \"" (.toString mv) "\"")))
-
-(defmethod print-dup Instant [o w]
-  (print-method o w))
-
-;; --- Low-Level Api
-
-(defn reader
-  ([istream]
-   (reader istream nil))
-  ([istream {:keys [type] :or {type :msgpack}}]
-   (t/reader istream type {:handlers +read-handlers+})))
-
-(defn read!
-  "Read value from streamed transit reader."
-  [reader]
-  (t/read reader))
-
-(defn writer
-  ([ostream]
-   (writer ostream nil))
-  ([ostream {:keys [type] :or {type :msgpack}}]
-   (t/writer ostream type {:handlers +write-handlers+})))
-
-(defn write!
-  [writer data]
-  (t/write writer data))
-
-;; --- High-Level Api
-
-;; TODO: check performance of different options
-
-(defn decode
-  ([data]
-   (decode data nil))
-  ([data opts]
-   (cond
-     (string? data)
-     (decode (.getBytes data "UTF-8") opts)
-
-     (bytes? data)
-     (with-open [input (ByteArrayInputStream. data)]
-       (read! (reader input opts)))
-
-     :else
-     (with-open [input (io/input-stream data)]
-       (read! (reader input opts))))))
-
-(defn encode
-  (^bytes [data]
-   (encode data nil))
-  (^bytes [data opts]
-   (with-open [out (ByteArrayOutputStream.)]
-     (let [w (writer out opts)]
-       (write! w data)
-       (.toByteArray out)))))

backend/src/vertx/web.clj

@@ -6,36 +6,33 @@
 
 (ns vertx.web
   "High level api for http servers."
-  (:require [clojure.spec.alpha :as s]
+  (:require
-            [promesa.core :as p]
+   [clojure.tools.logging :as log]
-            [sieppari.core :as sp]
+   [clojure.spec.alpha :as s]
-            [reitit.core :as rt]
+   [promesa.core :as p]
-            [vertx.http :as vxh]
+   [sieppari.core :as sp]
-            [vertx.util :as vu])
+   [reitit.core :as rt]
+   [vertx.http :as vh]
+   [vertx.util :as vu])
   (:import
-   clojure.lang.Keyword
    clojure.lang.IPersistentMap
-   io.vertx.core.Vertx
+   clojure.lang.Keyword
-   io.vertx.core.Handler
    io.vertx.core.Future
+   io.vertx.core.Handler
+   io.vertx.core.Vertx
    io.vertx.core.buffer.Buffer
    io.vertx.core.http.Cookie
    io.vertx.core.http.HttpServer
+   io.vertx.core.http.HttpServerOptions
    io.vertx.core.http.HttpServerRequest
    io.vertx.core.http.HttpServerResponse
-   io.vertx.core.http.HttpServerOptions
    io.vertx.ext.web.Route
    io.vertx.ext.web.Router
    io.vertx.ext.web.RoutingContext
    io.vertx.ext.web.handler.BodyHandler
-   io.vertx.ext.web.handler.StaticHandler
+   io.vertx.ext.web.handler.LoggerHandler
    io.vertx.ext.web.handler.ResponseTimeHandler
-   io.vertx.ext.web.handler.LoggerHandler))
+   io.vertx.ext.web.handler.StaticHandler))
-
-;; --- Constants & Declarations
-
-(declare -handle-response)
-(declare -handle-body)
 
 ;; --- Public Api
 
@@ -43,16 +40,17 @@
   (s/or :fn fn?
         :vec (s/every fn? :kind vector?)))
 
-(defn- make-ctx
+(defn- ->request
   [^RoutingContext routing-context]
   (let [^HttpServerRequest request (.request ^RoutingContext routing-context)
         ^HttpServerResponse response (.response ^RoutingContext routing-context)
         ^Vertx system (.vertx routing-context)]
     {:body (.getBody routing-context)
      :path (.path request)
+     :headers (vh/->headers request)
      :method (-> request .rawMethod .toLowerCase keyword)
-     ::request request
+     ::vh/request request
-     ::response response
+     ::vh/response response
      ::execution-context (.getContext system)
      ::routing-context routing-context}))
 
@@ -84,6 +82,12 @@
     {:status 405}
     {:status 404}))
 
+(defn- default-on-error
+  [err req]
+  (log/error err)
+  {:status 500
+   :body "Internal server error!\n"})
+
 (defn- run-chain
   [ctx chain handler]
   (let [d (p/deferred)]
@@ -106,64 +110,47 @@
   ([routes] (router routes {}))
   ([routes {:keys [delete-uploads?
                    upload-dir
+                   on-error
                    log-requests?
                    time-response?]
             :or {delete-uploads? true
                  upload-dir "/tmp/vertx.uploads"
+                 on-error default-on-error
                  log-requests? false
                  time-response? true}
             :as options}]
    (let [rtr (rt/router routes options)
-         hdr #(router-handler rtr %)]
+         f #(router-handler rtr %)]
      (fn [^Router router]
        (let [^Route route (.route router)]
          (when time-response? (.handler route (ResponseTimeHandler/create)))
          (when log-requests? (.handler route (LoggerHandler/create)))
 
-         (.handler route (doto (BodyHandler/create true)
+         (doto route
-                           (.setDeleteUploadedFilesOnEnd delete-uploads?)
+           (.failureHandler
-                           (.setUploadsDirectory upload-dir)))
+            (reify Handler
-         (.handler route (reify Handler
+              (handle [_ rc]
-                           (handle [_ context]
+                (let [err (.failure ^RoutingContext rc)
-                             (let [ctx (make-ctx context)]
+                      req (.get ^RoutingContext rc "vertx$clj$req")]
-                               (-> (p/do! (hdr ctx))
+                  (-> (p/do! (on-error err req))
-                                   (p/then' #(-handle-response % ctx))
+                      (vh/-handle-response req))))))
-                                   (p/catch #(do (prn %) (.fail (:context ctx) %)))))))))
-       router))))
 
-;; --- Impl
+           (.handler
+            (doto (BodyHandler/create true)
+              (.setDeleteUploadedFilesOnEnd delete-uploads?)
+              (.setUploadsDirectory upload-dir)))
 
-(defprotocol IAsyncResponse
-  (-handle-response [_ _]))
 
-(extend-protocol IAsyncResponse
+           (.handler
-  clojure.lang.IPersistentMap
+            (reify Handler
-  (-handle-response [data ctx]
+              (handle [_ rc]
-    (let [status (or (:status data) 200)
+                (let [req (->request rc)
-          body (:body data)
+                      efn (fn [err]
-          res (::response ctx)]
+                            (.put ^RoutingContext rc "vertx$clj$req" req)
-      (.setStatusCode ^HttpServerResponse res status)
+                            (.fail ^RoutingContext rc err))]
-      (-handle-body body res))))
+                  (try
-
+                    (-> (vh/-handle-response (f req) req)
-(defprotocol IAsyncBody
+                        (p/catch' efn))
-  (-handle-body [_ _]))
+                    (catch Exception err
-
+                      (efn err)))))))))
-(extend-protocol IAsyncBody
+         router))))
-  (Class/forName "[B")
-  (-handle-body [data res]
-    (.end ^HttpServerResponse res (Buffer/buffer data)))
-
-  Buffer
-  (-handle-body [data res]
-    (.end ^HttpServerResponse res ^Buffer data))
-
-  nil
-  (-handle-body [data res]
-    (.putHeader ^HttpServerResponse res "content-length" "0")
-    (.end ^HttpServerResponse res))
-
-  String
-  (-handle-body [data res]
-    (let [length (count data)]
-      (.putHeader ^HttpServerResponse res "content-length" (str length))
-      (.end ^HttpServerResponse res data))))

backend/src/vertx/web/interceptors.clj

@@ -11,22 +11,25 @@
    [clojure.string :as str]
    [promesa.core :as p]
    [reitit.core :as r]
+   [vertx.http :as vh]
    [vertx.web :as vw]
+   [vertx.util :as vu]
    [sieppari.context :as spx]
    [sieppari.core :as sp])
   (:import
    clojure.lang.Keyword
    clojure.lang.MapEntry
-   java.util.Map
-   java.util.Map$Entry
-   io.vertx.core.Vertx
-   io.vertx.core.Handler
    io.vertx.core.Future
+   io.vertx.core.Handler
+   io.vertx.core.MultiMap
+   io.vertx.core.Vertx
    io.vertx.core.http.Cookie
    io.vertx.core.http.HttpServerRequest
    io.vertx.core.http.HttpServerResponse
    io.vertx.ext.web.FileUpload
-   io.vertx.ext.web.RoutingContext))
+   io.vertx.ext.web.RoutingContext
+   java.util.Map
+   java.util.Map$Entry))
 
 ;; --- Cookies
 
@@ -40,71 +43,50 @@
 
 (defn cookies
   []
-  {:enter (fn [data]
+  {:enter
-            (let [^HttpServerRequest req (get-in data [:request ::vw/request])
+   (fn [data]
-                  parse-cookie (fn [^Cookie item] [(.getName item) (.getValue item)])
+     (let [^HttpServerRequest req (get-in data [:request ::vh/request])
-                  cookies (into {} (map parse-cookie) (vals (.cookieMap req)))]
+           parse-cookie (fn [^Cookie item] [(.getName item) (.getValue item)])
-              (update data :request assoc :cookies cookies)))
+           cookies (into {} (map parse-cookie) (vals (.cookieMap req)))]
-   :leave (fn [data]
+       (update data :request assoc :cookies cookies)))
-            (let [cookies (get-in data [:response :cookies])
+   :leave
-                  ^HttpServerResponse res (get-in data [:request ::vw/response])]
+   (fn [data]
-              (when (map? cookies)
+     (let [cookies (get-in data [:response :cookies])
-                (reduce-kv #(.addCookie res (build-cookie %1 %2)) nil cookies))
+           ^HttpServerResponse res (get-in data [:request ::vh/response])]
-              data))})
+       (when (map? cookies)
-;; --- Headers
+         (vu/doseq [[key val] cookies]
-
+           (.addCookie res (build-cookie key val))))
-(def ^:private lowercase-keys-t
+       data))})
-  (map (fn [^Map$Entry entry]
-         (MapEntry. (.toLowerCase (.getKey entry)) (.getValue entry)))))
-
-(defn- parse-headers
-  [req]
-  (let [^HttpServerRequest request (::vw/request req)]
-    (into {} lowercase-keys-t (.headers request))))
-
-(defn headers
-  []
-  {:enter (fn [data]
-            (update data :request assoc :headers (parse-headers (:request data))))
-   :leave (fn [data]
-            (let [^HttpServerResponse res (get-in data [:request ::vw/response])
-                  headers (get-in data [:response :headers])]
-              (run! (fn [[key value]]
-                      (.putHeader ^HttpServerResponse res
-                                  ^String (name key)
-                                  ^String (str value)))
-                    headers)
-              data))})
 
 ;; --- Params
 
-(defn- parse-param-entry
-  [acc ^Map$Entry item]
-  (let [key (keyword (.toLowerCase (.getKey item)))
-        prv (get acc key ::default)]
-    (cond
-      (= prv ::default)
-      (assoc! acc key (.getValue item))
-
-      (vector? prv)
-      (assoc! acc key (conj prv (.getValue item)))
-
-      :else
-      (assoc! acc key [prv (.getValue item)]))))
-
 (defn- parse-params
-  [req]
+  [^HttpServerRequest request]
-  (let [request (::vw/request req)]
+  (let [params (.params request)
-    (persistent!
+        it (.iterator ^MultiMap params)]
-     (reduce parse-param-entry
+    (loop [m (transient {})]
-             (transient {})
+      (if (.hasNext it)
-             (.params ^HttpServerResponse request)))))
+        (let [^Map$Entry o (.next it)
+              key (keyword (.toLowerCase (.getKey o)))
+              prv (get m key ::default)
+              val (.getValue o)]
+          (cond
+            (= prv ::default)
+            (recur (assoc! m key val))
+
+            (vector? prv)
+            (recur (assoc! m key (conj prv val)))
+
+            :else
+            (recur (assoc! m key [prv val]))))
+        (persistent! m)))))
 
 (defn params
   ([] (params nil))
   ([{:keys [attr] :or {attr :params}}]
    {:enter (fn [data]
-             (let [params (parse-params (:request data))]
+             (let [request (get-in data [:request ::vh/request])
+                   params (parse-params request)]
                (update data :request assoc attr params)))}))
 
 ;; --- Uploads