Update support for current application/grpc-web-text, grpc-web js

Rkiouak · Rkiouak · commit c3e40fb51b4d · 2020-11-23T16:37:17.000-05:00
behavior

Signed-off-by: Matt Rkiouak &lt;mrkiouak@gmail.com&gt;
diff --git a/src/protojure/pedestal/interceptors/grpc_web.clj b/src/protojure/pedestal/interceptors/grpc_web.clj
@@ -4,24 +4,65 @@
 
 (ns protojure.pedestal.interceptors.grpc-web
   "A [Pedestal](http://pedestal.io/) [interceptor](http://pedestal.io/reference/interceptors) for the [GRPC-WEB](https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-WEB.md) protocol"
-  (:require [io.pedestal.interceptor :refer [->Interceptor]])
-  (:import (org.apache.commons.codec.binary Base64InputStream))
+  (:require [io.pedestal.interceptor :refer [->Interceptor]]
+            [clojure.core.async :as async]
+            [clojure.data])
   (:refer-clojure :exclude [proxy]))
 
 (set! *warn-on-reflection* true)
 
+(defn read-n [from-chan n]
+  "Convenience method for consuming n [n] or less values from a channel [from-chan]"
+  (async/go-loop [res []]
+    (let [v (async/<! from-chan)]
+      (if (nil? v)
+        [true res]
+        (let [new-res (conj res v)]
+          (if (= n (count new-res))
+            [false new-res]
+            (recur new-res)))))))
+
 (defn- decode-body
-  [{:keys [body] :as request}]
-  (assoc request :body (Base64InputStream. body)))
+  "Consumes 4 base64 encoded bytes from the body, and writes to a new channel that replaces the prior :body-ch on the
+  request. Note that any consumption from the :body inputstream that is not followed by a write to the :body-ch key
+  results in that data being lost to the grpc interceptor :grpc-params/:body-ch key."
+  [{:keys [body-ch] :as request}]
+  (let [dec-ch (async/chan 4056)
+        decoder (java.util.Base64/getDecoder)]
+    (async/go-loop [[final encoded] (async/<! (read-n body-ch 4))]
+      (if (and (empty? encoded) final)
+        (async/close! dec-ch)
+        (do
+          (doseq [b (.decode decoder (byte-array encoded))]
+            (async/>! dec-ch b))
+          (recur (async/<! (read-n body-ch 4))))))
+    (assoc request :body-ch dec-ch)))
+
+(defn- encode-body
+  "Consumes bytes from the response body channel and base64 encodes the payload"
+  [{{:keys [body] :as response} :response :as ctx}]
+  (let [encoder (java.util.Base64/getEncoder)
+        out-ch (async/chan 4056)]
+    (async/go-loop [s (async/<! body)]
+      (if (not s)
+        (async/close! out-ch)
+        (do
+          (async/>! out-ch (.encode encoder ^bytes s))
+          (recur (async/<! body)))))
+    (-> (assoc-in ctx [:response :body] out-ch)
+        (update-in [:response :headers] #(merge % {"Content-Type" "application/grpc-web-text"})))))
 
 (def ^{:no-doc true :const true} content-types
-  #{"application/grpc-web-text"
-    "application/grpc-web-text+proto"})
+  #{"application/grpc-web-text"})
 
 (defn- web-text?
   [{{:strs [content-type]} :headers}]
   (contains? content-types content-type))
 
+(defn- accept-web-text?
+  [{{{:strs [accept]} :headers} :request}]
+  (contains? content-types accept))
+
 (defn- pred->
   "Threads 'item' through both the predicate and, when 'pred' evaluates true, 'xform' functions. Else, just returns 'item'"
   [item pred xform]
@@ -32,9 +73,9 @@
   (assoc ctx :request (pred-> request web-text? decode-body)))
 
 (defn- leave-handler
-  [ctx]
+  [{:keys [response] :as ctx}]
   ;; TODO "Clarify & implement grpc-web trailer behavior"
-  ctx)
+  (pred-> ctx accept-web-text? encode-body))
 
 (defn- exception-handler
   [ctx e]
diff --git a/test/protojure/grpc_web_test.clj b/test/protojure/grpc_web_test.clj
@@ -11,80 +11,125 @@
             [io.pedestal.http.body-params :as body-params]
             [example.types :as example]
             [protojure.protobuf :as pb]
-            [clojure.data.codec.base64 :as b64]))
+            [clojure.data.codec.base64 :as b64]
+            [protojure.pedestal.interceptors.grpc :as grpc]
+            [clojure.core.async :as async]
+            [example.hello.Greeter :as greeter]
+            [clojure.core.async :refer [<!! >!! <! >! go go-loop] :as async]
+            [protojure.test.utils :as test.utils]
+            [protojure.grpc.client.api :as grpc-api]
+            [protojure.grpc.client.providers.http2 :as grpc.http2]
+            [protojure.grpc.client.utils :as client.utils]
+            [promesa.core :as p]
+            [protojure.internal.grpc.client.providers.http2.jetty :as jetty-client]
+            [protojure.grpc.status :as grpc.status]
+            [protojure.pedestal.routes :as pedestal.routes]
+            [example.hello :refer [new-HelloRequest pb->HelloReply]]
+            [clj-http.client :as client]
+            [protojure.grpc.codec.lpm :as lpm]))
 
-(defn grpc-echo [{:keys [body] :as request}]
-  {:status   200
-   :body     (example/pb->Money body)
-   :trailers {"grpc-status" 0 "grpc-message" "Got it!"}})
+(defonce test-env (atom {}))
 
-(def interceptors [(body-params/body-params)
-                   grpc-web/proxy])
+;;-----------------------------------------------------------------------------
+;; "Greeter" service endpoint
+;;-----------------------------------------------------------------------------
+(deftype Greeter []
+  greeter/Service
+  (SayHello
+    [this {{:keys [name]} :grpc-params :as request}]
+    {:status 200
+     :body {:message (str "Hello, " name)}})
+  (SayRepeatHello
+    [this {{:keys [name]} :grpc-params :as request}]
+    (let [resp-chan (:grpc-out request)]
+      (go
+        (dotimes [_ 3]
+          (>! resp-chan {:message (str "Hello, " name)}))
+        (async/close! resp-chan))
+      {:status 200
+       :body resp-chan}))
+  (SayHelloOnDemand
+    [this {:keys [grpc-params] :as request}]
+    (let [out-chan (:grpc-out request)]
+      (go-loop [name (:name (<! grpc-params))]
+        (if name
+          (do
+            (>! out-chan {:message (str "Hello, " name)})
+            (recur (:name (<! grpc-params))))
+          (async/close! out-chan)))
+      {:status 200
+       :body out-chan}))
+  (SayHelloError
+    [this req]
+    {:status 200
+     :body "This isn't a protobuf message"})
+  (SayNil
+    [this req]
+    (grpc.status/error :unauthenticated)))
 
-(def routes [["/" :get (conj interceptors `grpc-echo)]])
+(defn- greeter-mock-routes [interceptors]
+  (pedestal.routes/->tablesyntax {:rpc-metadata     greeter/rpc-metadata
+                                  :interceptors     interceptors
+                                  :callback-context (Greeter.)}))
 
-(def service (let [service-params {:env                      :prod
-                                   ::pedestal/routes         (into #{} routes)
-                                   ::pedestal/type           protojure.pedestal/config
-                                   ::pedestal/chain-provider protojure.pedestal/provider}]
-               (:io.pedestal.http/service-fn (io.pedestal.http/create-servlet service-params))))
+(defn- grpc-connect
+  ([] (grpc-connect (:port @test-env)))
+  ([port]
+   @(grpc.http2/connect {:uri (str "http://localhost:" port) :content-coding "gzip"})))
 
-(deftest grpc-web-text-check
-  (testing "Check that a round-trip GRPC request works"
-    (let [input-msg (pb/->pb (example/new-Money {:currency_code (apply str (repeat 20 "foobar")) :units 42 :nanos 750000000}))]
-      (is
-       (=
-        (with-out-str (pr (example/pb->Money input-msg)))
-        (:body (response-for
-                service
-                :get "/"
-                :headers {"Content-Type" "application/grpc-web-text"}
-                :body (clojure.java.io/input-stream (b64/encode input-msg)))))))))
+;;-----------------------------------------------------------------------------
+;; Fixtures
+;;-----------------------------------------------------------------------------
+(defn create-service []
+  (let [port (test.utils/get-free-port)
+        interceptors [(body-params/body-params)
+                      pedestal/html-body]
+        server-params {:env                      :prod
+                       ::pedestal/routes         (into #{} (greeter-mock-routes interceptors))
+                       ::pedestal/port           port
 
-(deftest grpc-web-check
-  (testing "Check that a round-trip GRPC request works"
-    (let [input-msg (pb/->pb (example/new-Money {:currency_code (apply str (repeat 20 "foobar")) :units 42 :nanos 750000000}))]
-      (is
-       (=
-        (with-out-str (pr (example/pb->Money input-msg)))
-        (:body (response-for
-                service
-                :get "/"
-                :headers {"Content-Type" "application/grpc-web"}
-                :body (clojure.java.io/input-stream input-msg))))))))
+                       ::pedestal/type           protojure.pedestal/config
+                       ::pedestal/chain-provider protojure.pedestal/provider}
+        client-params {:port port}]
 
-(deftest grpc-web-proto-check
-  (testing "Check that a round-trip GRPC request works"
-    (let [input-msg (pb/->pb (example/new-Money {:currency_code (apply str (repeat 20 "foobar")) :units 42 :nanos 750000000}))]
-      (is
-       (=
-        (with-out-str (pr (example/pb->Money input-msg)))
-        (:body (response-for
-                service
-                :get "/"
-                :headers {"Content-Type" "application/grpc-web+proto"}
-                :body (clojure.java.io/input-stream input-msg))))))))
+    (let [server (test.utils/start-pedestal-server server-params)
+          client @(jetty-client/connect client-params)
+          grpc-client (grpc-connect port)]
+      (swap! test-env assoc :port port :server server :client client :grpc-client grpc-client))))
 
-(deftest grpc-web-text-proto-check
-  (testing "Check that a round-trip GRPC request works"
-    (let [input-msg (pb/->pb (example/new-Money {:currency_code (apply str (repeat 20 "foobar")) :units 42 :nanos 750000000}))]
-      (is
-       (=
-        (with-out-str (pr (example/pb->Money input-msg)))
-        (:body (response-for
-                service
-                :get "/"
-                :headers {"Content-Type" "application/grpc-web-text+proto"}
-                :body (clojure.java.io/input-stream (b64/encode input-msg)))))))))
+(defn destroy-service []
+  (swap! test-env update :grpc-client grpc-api/disconnect)
+  (swap! test-env update :client jetty-client/disconnect)
+  (swap! test-env update :server pedestal/stop))
 
-(deftest grpc-web-no-header-match-check
-  (testing "Check that a round-trip GRPC request works"
-    (let [input-msg (pb/->pb (example/new-Money {:currency_code (apply str (repeat 20 "foobar")) :units 42 :nanos 750000000}))]
-      (is
-       (=
-        (with-out-str (pr (example/pb->Money input-msg)))
-        (:body (response-for
-                service
-                :get "/"
-                :headers {"Content-Type" "application/grpc"}
-                :body (clojure.java.io/input-stream input-msg))))))))
+(defn wrap-service [test-fn]
+  (create-service)
+  (test-fn)
+  (destroy-service))
+
+(use-fixtures :once wrap-service)
+
+(deftest grpc-web-test-check
+  (let [in (async/chan 10)
+        out (async/chan 10)
+        resp-in (async/chan 10)
+        resp-out (async/chan 10)]
+    (lpm/encode new-HelloRequest in out {:encoding identity})
+    (lpm/decode pb->HelloReply resp-in resp-out {:encoding identity})
+    (async/>!! in {:name "World"})
+    (async/close! in)
+    (testing "Check that a round-trip unary grpc-web-text request works"
+      (let [lpm (async/<!! (async/into [] out))
+            b64-encoded (-> (java.util.Base64/getEncoder)
+                            (.encode (byte-array lpm)))
+            body (-> (java.util.Base64/getDecoder)
+                     (.decode (-> (client/post
+                                   (str "http://localhost:" (:port @test-env) "/example.hello.Greeter/SayHello")
+                                   {:body    b64-encoded
+                                    :content-type "application/grpc-web-text"
+                                    :accept       "application/grpc-web-text"})
+                                  :body)))]
+        (doseq [b body]
+          (async/>!! resp-in b))
+        (async/close! resp-in)
+        (is (= "Hello, World" (:message (async/<!! resp-out))))))))
diff --git a/test/protojure/pedestal/interceptors/grpc_web_test.clj b/test/protojure/pedestal/interceptors/grpc_web_test.clj
@@ -0,0 +1,14 @@
+(ns protojure.pedestal.interceptors.grpc-web-test
+  (:require [clojure.test :refer :all])
+  (:require [protojure.pedestal.interceptors.grpc-web :refer [read-n]]
+            [clojure.core.async :as async]))
+
+(deftest read-n-test
+  (testing "test that read-n reads the appropriate values and returns final on final"
+    (let [test-ch (async/chan 10)]
+      (doseq [n [1 2 3 4]]
+        (async/>!! test-ch n))
+      (async/close! test-ch)
+      (let [result (async/<!! (read-n test-ch 3))]
+        (is (= result [false [1 2 3]]))
+        (is (= [true [4]] (async/<!! (read-n test-ch 3))))))))
