[new] [#477] Add gzip-wrapping packer

Author: ptaoussanis

src/taoensso/sente/packers/gzip.cljc

@@ -0,0 +1,67 @@
+(ns taoensso.sente.packers.gzip
+  {:author "Peter Taoussanis (@ptaoussanis)"}
+  (:require
+   [taoensso.truss :as truss]
+   [taoensso.sente.packers.impl.gzip :as impl]
+   [taoensso.sente.interfaces :as i]))
+
+#?(:clj (defn- ba->str ^String [^bytes ba] (String. ba  java.nio.charset.StandardCharsets/UTF_8)))
+#?(:clj (defn- str->ba ^bytes  [^String s] (.getBytes s java.nio.charset.StandardCharsets/UTF_8)))
+
+#?(:cljs (let [encoder (js/TextEncoder.)]         (defn- str->u8s [^string s] (.encode encoder s))))
+#?(:cljs (let [decoder (js/TextDecoder. "utf-8")] (defn- u8s->str [^js u8s]   (.decode decoder u8s))))
+
+(defn wrap-packer
+  "Experimental, please test carefully and report any issues!
+
+  Returns Sente packer that wraps another with gzip compression.
+  Needs `js/CompressionStream` browser support.
+
+  If `packer` takes+returns platform bytes:   `binary?` should be true.
+  If `packer` takes+returns platform strings: `binary?` should be false."
+
+  [packer {:keys [binary?]}]
+  #?(:clj
+     (reify i/IPacker2
+       (unpack [_ ws? ba cb]
+         (if binary?
+           (i/unpack packer ws?          (impl/gunzip ba)  cb)
+           (i/unpack packer ws? (ba->str (impl/gunzip ba)) cb)))
+
+       (pack [_ ws? x cb]
+         (i/pack packer ws? x
+           (fn [{:keys [value error]}]
+             (if           error
+               (cb {:error error})
+               (cb {:value (impl/gzip (if binary? value (str->ba value)))}))))))
+
+     :cljs
+     (do
+       (when-not (exists? js/CompressionStream)   (truss/ex-info!   "CompressionStream not supported"))
+       (when-not (exists? js/DecompressionStream) (truss/ex-info! "DecompressionStream not supported"))
+       (reify i/IPacker2
+         (unpack [_ ws? packed cb]
+           (->
+             (impl/gunzip (impl/as-u8s packed)) ; Decompress -> u8s promise
+             (.then
+               (fn [u8s]
+                 (if binary?
+                   (i/unpack packer ws?           u8s  cb)
+                   (i/unpack packer ws? (u8s->str u8s) cb))))
+             (.catch (fn [err] (cb {:error err})))))
+
+         (pack [_ ws? cljs-val cb]
+           (i/pack packer ws? cljs-val
+             (fn [{:keys [error value]}] ; Serialize -> text/bin value
+               (if           error
+                 (cb {:error error})
+                 (-> (impl/gzip (if binary? (impl/as-u8s value) (str->u8s value))) ; Compress -> u8s promise
+                   (.then  (fn [u8s] (cb {:value u8s})))
+                   (.catch (fn [err] (cb {:error err}))))))))))))
+
+(comment
+  (let [p (wrap-packer (taoensso.sente.packers.edn/get-packer) {:binary? false})]
+    (i/pack p :ws [:chsk/ws-ping "foo"]
+      (fn [{packed :value}]
+        (i/unpack p :ws packed
+          (fn [{clj :value}] clj))))))

src/taoensso/sente/packers/impl/gzip.clj

@@ -0,0 +1,51 @@
+(ns taoensso.sente.packers.impl.gzip)
+
+(def ^:private ^:const prefix-raw  (byte 0x00))
+(def ^:private ^:const prefix-gzip (byte 0x01))
+(def ^:private ^:const min-bytes   1024)
+
+(defn- prefix
+  "Returns given byte[] with added flag prefix."
+  ^bytes [gzip? ^bytes ba]
+  (let [len (alength ba)
+        out (byte-array (inc len))]
+    (aset-byte             out 0 (if gzip? prefix-gzip prefix-raw))
+    (System/arraycopy ba 0 out 1 len)
+    (do                    out)))
+
+(defn gzip
+  "Uncompressed byte[] -> compressed byte[]."
+  ^bytes [^bytes ba]
+  (let [len (alength ba)]
+    (if (< len min-bytes)
+      (prefix false ba)
+      (with-open [baos (java.io.ByteArrayOutputStream.)]
+        (.write baos (int prefix-gzip))
+        (with-open [gos (java.util.zip.GZIPOutputStream. baos)]
+          (.write gos ba 0 len))
+        (.toByteArray baos)))))
+
+(defn gunzip
+  "Compressed byte[] -> uncompressed byte[]."
+  ^bytes [^bytes ba]
+  (let [len   (alength ba)
+        flag  (bit-and 0xFF (aget ba 0))]
+    (if-not (== flag prefix-gzip)
+      (java.util.Arrays/copyOfRange ba 1 len)
+      (with-open [bais (java.io.ByteArrayInputStream.  ba 1 (dec len))
+                  gis  (java.util.zip.GZIPInputStream. bais)
+                  baos (java.io.ByteArrayOutputStream.)]
+        (.transferTo gis baos)
+        (.toByteArray    baos)))))
+
+(comment
+  (defn bn [len] (byte-array (take len (cycle (range 128)))))
+
+  (let [ba-sm (bn 128)
+        ba-lg (bn 2048)]
+    (taoensso.encore/qb 1e4 ; [3.6 280.32]
+      (gunzip (gzip ba-sm))
+      (gunzip (gzip ba-lg))))
+
+  (take 16 (gunzip (gzip (bn 128))))
+  (take 16 (gunzip (gzip (bn 2048)))))

src/taoensso/sente/packers/impl/gzip.cljs

@@ -0,0 +1,62 @@
+(ns taoensso.sente.packers.impl.gzip)
+
+(def ^:private ^:const prefix-raw  0x00)
+(def ^:private ^:const prefix-gzip 0x01)
+(def ^:private ^:const min-bytes   1024)
+
+(defn- type-name [x]
+  (let [ctor (type x)]
+    (or
+      (.-name ctor) (.-displayName ctor) (goog/typeOf x)
+      (try (pr-str ctor) (catch :default _ nil))
+      "<unknown>")))
+
+(defn- as-u8s
+  "#{Uint8Array ArrayBuffer DataView} -> Uint8Array"
+  ^js [input]
+  (cond
+    (instance? js/Uint8Array  input)                 input
+    (instance? js/ArrayBuffer input) (js/Uint8Array. input)
+    (instance? js/DataView    input) (js/Uint8Array. (.-buffer     input)
+                                                     (.-byteOffset input)
+                                                     (.-byteLength input))
+    :else
+    (throw
+      (ex-info "Unexpected input type"
+        {:type (type-name input),
+         :expected '#{Uint8Array ArrayBuffer DataView}}))))
+
+(defn- prefix
+  "Returns given Uint8Array with added flag prefix."
+  [gzip? ^js u8s]
+  (let [out (js/Uint8Array. (inc (.-length u8s)))]
+    (aset out 0 (if gzip? prefix-gzip prefix-raw))
+    (.set out u8s 1)
+    (do   out)))
+
+(defn- stream->u8s [readable] (-> (js/Response. readable) (.arrayBuffer) (.then #(js/Uint8Array. %))))
+
+(defn gzip
+  "Uncompressed Uint8Array -> compressed Uint8Array Promise."
+  [^js u8s]
+  (if (< (.-length u8s) min-bytes)
+    (js/Promise.resolve (prefix false u8s))
+    (let [cs     (js/CompressionStream. "gzip")
+          writer (.getWriter (.-writable cs))]
+      (->
+        (.write writer u8s)
+        (.then (fn [] (.close writer) (stream->u8s (.-readable cs))))
+        (.then (fn [u8s] (prefix true u8s)))))))
+
+(defn gunzip
+  "Compressed Uint8Array -> uncompressed Uint8Array Promise."
+  [^js u8s]
+  (let [flag (aget u8s 0)
+        body (.subarray u8s 1)]
+    (if-not (== flag prefix-gzip)
+      (js/Promise.resolve body)
+      (let [ds     (js/DecompressionStream. "gzip")
+            writer (.getWriter (.-writable ds))]
+        (->
+          (.write writer body)
+          (.then (fn [] (.close writer) (stream->u8s (.-readable ds)))))))))

src/taoensso/sente/packers/msgpack.cljc

@@ -6,17 +6,17 @@
    [taoensso.sente.interfaces :as i]))
 
 (defn get-packer
-  "Returns Sente packer that uses the binary MessagePack
+  "Experimental, please test carefully and report any issues!
+
+  Returns Sente packer that uses the binary MessagePack
   format, Ref. <https://msgpack.org/index.html>.
 
   Clj/s MessagePack implementation adapted from
-  <https://github.com/rosejn/msgpack-cljc>.
-
-  Experimental, please test carefully and report any issues!"
+  <https://github.com/rosejn/msgpack-cljc>."
   []
   (reify i/IPacker2
-    (pack   [_ _ws? x  cb-fn] (cb-fn {:value (msgpack/pack    x)}))
-    (unpack [_ _ws? ba cb-fn] (cb-fn {:value (msgpack/unpack ba)}))))
+    (pack   [_ _ws? x  cb] (cb {:value (msgpack/pack    x)}))
+    (unpack [_ _ws? in cb] (cb {:value (msgpack/unpack in)}))))
 
 (comment
   (let [p (get-packer)]

test/taoensso/sente_tests.cljc

@@ -18,24 +18,26 @@
    [taoensso.encore :as enc]
 
    ;; :cljs cannot compile taoensso.sente under :nodejs target
-   #?(:clj  [taoensso.sente :as sente])
+   #?(:clj [taoensso.sente :as sente])
 
    [taoensso.sente.interfaces      :as i]
    [taoensso.sente.packers.edn     :as ep]
    [taoensso.sente.packers.transit :as tp]
-   [taoensso.sente.packers.msgpack :as mp])
-
-  #?(:cljs
-     (:require-macros
-      [taoensso.sente-tests :refer [get-bench-data]])))
+   [taoensso.sente.packers.msgpack :as mp]
+   [taoensso.sente.packers.gzip    :as gz]))
 
 (comment
   (remove-ns      'taoensso.sente-tests)
   (test/run-tests 'taoensso.sente-tests))
 
 ;;;;
 
-(defn get-bench-data []
+(def ^:const timeout-msecs 4000)
+(def ^:const bench-laps 1e2)
+
+;;;; Test data
+
+(defn get-test-data []
   #_{:a :A :b :B :c "foo", :v (vec (range 128)), :s (set (range 128))}
   {:ids (vec (repeatedly 1024 (fn [] (rand-int 32767))))
    :data
@@ -63,29 +65,97 @@
           :submit-date  (str (enc/now-inst))
           :display-name (enc/uuid-str)
           :user nil
+          :uuid (random-uuid)
           :a-vector-of-integers (vec (repeatedly 64 (fn [] (rand-int 128))))
           :children-count (rand-int 1024)
           :a-usually-absent-string nil})))})
 
-(defn packed-len [x]
-  #?(:cljs (if (instance? js/Uint8Array x) (.-length x) (count x))
-     :clj (count x)))
-
-(defn bench1 [packer laps data]
-  {:time (enc/qb laps (i/pack packer nil data (fn [x] (i/unpack packer nil (get x :value) (fn [y] (get y :value))))))
-   :size (packed-len  (i/pack packer nil data (fn [x]                      (get x :value))))})
-
-(deftest bench-packers
-  (let [laps 1e2
-        bd (get-bench-data)
-        ep (ep/get-packer)
-        tp (tp/get-packer)
-        mp (mp/get-packer)]
-
-    (println "Benching comparison:"
-      {:ep (bench1 ep laps bd)
-       :tp (bench1 tp laps bd)
-       :mp (bench1 mp laps bd)})))
+(def td1 (get-test-data))
+(def td2 (get-test-data))
+
+;;;; Packers
+
+(defn roundtrip
+  "Returns Clj/s promise."
+  ([data   packer cb-fn]
+   (i/pack packer nil data
+     (fn [packed]
+       (i/unpack packer nil    (get packed   :value)
+         (fn [unpacked] (cb-fn (get unpacked :value)))))))
+
+  ([data packer]
+   #?(:clj                  (let [p (promise)] (roundtrip data packer (fn [v] (deliver p v))) p)
+      :cljs (js/Promise. (fn [resolve _reject] (roundtrip data packer (fn [v] (resolve   v))))))))
+
+(comment @(roundtrip :data (ep/get-packer)))
+
+(defn test-promise [p f] ; f is test-fn
+  #?(:clj (f (deref p timeout-msecs :timeout))
+     :cljs
+     (test/async done
+       (let [done  (let [run?_ (atom false)] (fn [] (when (compare-and-set! run?_ false true) (done))))
+             timer (js/setTimeout #(do (f :timeout) (done)) timeout-msecs)]
+         (-> p
+           (.then    (fn [v] (js/clearTimeout timer) (f v)))
+           (.catch   (fn [e] (js/clearTimeout timer) (f e)))
+           (.finally (fn [ ] (js/clearTimeout timer) (done))))))))
+
+(do
+  (deftest test-packer-edn        (test-promise (roundtrip td1                 (ep/get-packer))                   #(is (= td1 %))))
+  (deftest test-packer-transit    (test-promise (roundtrip td1                 (tp/get-packer))                   #(is (= td1 %))))
+  (deftest test-packer-msgpack    (test-promise (roundtrip td1                 (mp/get-packer))                   #(is (= td1 %))))
+  (deftest test-packer-gz+edn     (test-promise (roundtrip td1 (gz/wrap-packer (ep/get-packer) {:binary? false})) #(is (= td1 %))))
+  (deftest test-packer-gz+transit (test-promise (roundtrip td1 (gz/wrap-packer (tp/get-packer) {:binary? false})) #(is (= td1 %))))
+  (deftest test-packer-gz+msgpack (test-promise (roundtrip td1 (gz/wrap-packer (mp/get-packer) {:binary?  true})) #(is (= td1 %)))))
+
+;;;; Benching
+
+(defn packed-len [x] #?(:clj (count x), :cljs (if (instance? js/Uint8Array x) (.-length x) (count x))))
+
+(defn bench1
+  "Returns Clj/s promise."
+  [data packer]
+  #?(:clj
+     (let [size (let [p (promise)] (i/pack packer nil data #(deliver p (packed-len (get % :value)))) (deref p timeout-msecs -1))
+           t0   (enc/now-nano)]
+
+       (dotimes [_ bench-laps] (deref (roundtrip data packer) timeout-msecs :timeout))
+       (let [msecs (enc/round (/ (- (enc/now-nano) t0) 1e6))]
+         (deliver (promise) {:size size, :msecs msecs})))
+
+     :cljs
+     (->
+       (js/Promise. (fn [resolve _] (i/pack packer nil data #(resolve (get % :value)))))
+       (.then
+         (fn [packed]
+           (let [size (packed-len packed)
+                 t0   (enc/now-nano)
+                 step
+                 (fn step [i]
+                   (if (< i bench-laps)
+                     (-> (roundtrip data packer) (.then (fn [_] (step (inc i)))))
+                     (let [msecs (enc/round (/ (- (enc/now-nano) t0) 1e6))]
+                       (js/Promise.resolve {:size size, :msecs msecs}))))]
+             (step 0)))))))
+
+(def   benched_ (atom {}))
+(defn >benched [id] (fn [val] (swap! benched_ assoc id val)))
+
+(test/use-fixtures :once
+  (taoensso.encore/test-fixtures
+    {:after
+     (fn []
+       (println "Benchmarks:")
+       (doseq [[id results] (sort @benched_)]
+         (println (str id ": " results))))}))
+
+(do
+  (deftest bench-packer-edn        (test-promise (bench1 td1                 (ep/get-packer))                   (>benched :edn)))
+  (deftest bench-packer-transit    (test-promise (bench1 td1                 (tp/get-packer))                   (>benched :transit)))
+  (deftest bench-packer-msgpack    (test-promise (bench1 td1                 (mp/get-packer))                   (>benched :msg-pack)))
+  (deftest bench-packer-gz+edn     (test-promise (bench1 td1 (gz/wrap-packer (ep/get-packer) {:binary? false})) (>benched :edn+gz)))
+  (deftest bench-packer-gz+transit (test-promise (bench1 td1 (gz/wrap-packer (tp/get-packer) {:binary? false})) (>benched :transit+gz)))
+  (deftest bench-packer-gz+msgpack (test-promise (bench1 td1 (gz/wrap-packer (mp/get-packer) {:binary?  true})) (>benched :msg-pack+gz))))
 
 ;;;;