Skip to content

Commit 5388be8

Browse files
RkiouakRkiouak
committed
Add expected trailer behavior for grpc-web-text case
Signed-off-by: Matt Rkiouak <mrkiouak@gmail.com>
1 parent c3e40fb commit 5388be8

4 files changed

Lines changed: 149 additions & 32 deletions

File tree

‎src/protojure/pedestal/interceptors/grpc.clj‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -117,8 +117,9 @@
117117
assoc
118118
:headers {"Content-Type" "application/grpc+proto"}
119119
:status 200
120-
:body ""
121-
:trailers (generate-trailers {:grpc-status status :grpc-message msg})))
120+
:body (async/close! (async/chan 1))
121+
:trailers (generate-trailers {:grpc-status status :grpc-message msg})
122+
:grpc-error true))
122123

123124
(def error-interceptor
124125
(err/error-dispatch

‎src/protojure/pedestal/interceptors/grpc_web.clj‎

Lines changed: 97 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,11 @@
66
"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"
77
(:require [io.pedestal.interceptor :refer [->Interceptor]]
88
[clojure.core.async :as async]
9-
[clojure.data])
10-
(:refer-clojure :exclude [proxy]))
9+
[clojure.data]
10+
[promesa.core :as p])
11+
(:refer-clojure :exclude [proxy])
12+
(:import (org.apache.commons.codec.binary Base64InputStream Base64OutputStream)
13+
(java.io PipedOutputStream PipedInputStream)))
1114

1215
(set! *warn-on-reflection* true)
1316

@@ -29,29 +32,87 @@
2932
[{:keys [body-ch] :as request}]
3033
(let [dec-ch (async/chan 4056)
3134
decoder (java.util.Base64/getDecoder)]
32-
(async/go-loop [[final encoded] (async/<! (read-n body-ch 4))]
33-
(if (and (empty? encoded) final)
34-
(async/close! dec-ch)
35-
(do
36-
(doseq [b (.decode decoder (byte-array encoded))]
37-
(async/>! dec-ch b))
38-
(recur (async/<! (read-n body-ch 4))))))
39-
(assoc request :body-ch dec-ch)))
35+
(let [b64-decode-error-promise (p/promise (fn [resolve reject]
36+
(async/go-loop [[final encoded] (async/<! (read-n body-ch 4))]
37+
(if (and (empty? encoded) final)
38+
(do
39+
(resolve nil)
40+
(async/close! dec-ch))
41+
(do
42+
(try
43+
(doseq [b (.decode decoder (byte-array encoded))]
44+
(async/>! dec-ch b))
45+
(catch Exception e
46+
(async/close! dec-ch)
47+
(reject e)))
48+
(recur (async/<! (read-n body-ch 4))))))))]
49+
(-> (assoc request :body-ch dec-ch)
50+
;;FIXME the below promise is never checked for an error, but should be prior to responding -- this will present
51+
;; as a bug where b64 decode failures aren't reported to the client, rather it will be as if the request body
52+
;; was nil
53+
(assoc :b64-decode-error-promise b64-decode-error-promise)))))
4054

41-
(defn- encode-body
42-
"Consumes bytes from the response body channel and base64 encodes the payload"
43-
[{{:keys [body] :as response} :response :as ctx}]
44-
(let [encoder (java.util.Base64/getEncoder)
45-
out-ch (async/chan 4056)]
55+
(defn- num->bytes
56+
"Serializes an integer to a byte-array."
57+
[num]
58+
(byte-array (for [i (range 4)]
59+
(-> (unsigned-bit-shift-right num
60+
(* 8 (- 4 i 1)))
61+
(bit-and 0x0FF)))))
62+
63+
(defn- make-grpc-web-trailers-string [trailers]
64+
(reduce (fn [s [k v]]
65+
(str s k ":" v "\r\n")) "" trailers))
66+
67+
(defn- make-grpc-web-trailers-frame [trailers]
68+
(let [trailer-bytes (.getBytes ^String (make-grpc-web-trailers-string trailers))]
69+
(byte-array
70+
(concat
71+
[0x80]
72+
(into [] (num->bytes (count trailer-bytes)))
73+
(into [] trailer-bytes)))))
74+
75+
(defmulti encode-body "Consumes bytes from the response body and base64 encodes the payload"
76+
(fn [x] (type (-> x :response :body))))
77+
78+
(defmethod encode-body clojure.core.async.impl.channels.ManyToManyChannel
79+
[{{:keys [body trailers] :as response} :response :as ctx}]
80+
(let [pos (PipedOutputStream.)
81+
pis (PipedInputStream. pos)
82+
;; N.B. passing a string instead of nil in the last position (the line end) caused no data to send
83+
b64-is (Base64InputStream. pis true -1 nil)]
4684
(async/go-loop [s (async/<! body)]
4785
(if (not s)
48-
(async/close! out-ch)
86+
(let [frame (make-grpc-web-trailers-frame (async/<! trailers))]
87+
;;Write trailer frame
88+
(.write pos ^bytes frame)
89+
(.flush pos)
90+
(.close pos))
4991
(do
50-
(async/>! out-ch (.encode encoder ^bytes s))
92+
(.write pos ^bytes s)
5193
(recur (async/<! body)))))
52-
(-> (assoc-in ctx [:response :body] out-ch)
94+
(-> (assoc-in ctx [:response :body] b64-is)
5395
(update-in [:response :headers] #(merge % {"Content-Type" "application/grpc-web-text"})))))
5496

97+
(defmethod encode-body nil
98+
[{{:keys [trailers] :as response} :response :as ctx}]
99+
(let [frame (make-grpc-web-trailers-frame trailers)
100+
pos (PipedOutputStream.)
101+
pis (PipedInputStream. pos)
102+
b64-is (Base64InputStream. pis true -1 nil)]
103+
;;Write trailer frame
104+
(.write pos ^bytes frame)
105+
(.flush pos)
106+
(.close pos)
107+
(-> (assoc-in ctx [:response :body] b64-is)
108+
(update-in [:response :headers] #(merge % {"Content-Type" "application/grpc-web-text"})))))
109+
110+
(defmethod encode-body :default
111+
[{{:keys [body] :as response} :response :as ctx}]
112+
(throw (ex-info "grpc-web interceptor encountered an unexpected body type on-leave"
113+
{:causes #{:incompatible-body-value-type}
114+
:body-value-type (type body)})))
115+
55116
(def ^{:no-doc true :const true} content-types
56117
#{"application/grpc-web-text"})
57118

@@ -84,3 +145,21 @@
84145
(def proxy
85146
"Interceptor that provides a transparent proxy for the [GRPC-WEB](https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-WEB.md) protocol to standard protojure grpc protocol"
86147
(->Interceptor ::proxy enter-handler leave-handler exception-handler))
148+
149+
(defn error-leave-handler [{{:keys [grpc-error]} :response :as ctx}]
150+
(if grpc-error
151+
(pred-> ctx accept-web-text? encode-body)
152+
ctx))
153+
;;FIXME when HTTP/3 has a grpc specification
154+
;; since we rely on protojure.pedestal.interceptors.grpc/error-interceptor to form the grpc compliant trailers,
155+
;; we expose this error interceptor (and insert in protojure.pedestal.routes/->tablesyntax prior to
156+
;; interceptors.grpc/error-interceptor) so that this interceptor can check for the grpc-web-text accept content type
157+
;; and encode appropriately when an exception is thrown.
158+
;; Once we have a third grpc specification based on transport, better to fix these abstractions such that we have
159+
;; HTTP1.1/HTTP2/HTTP3 based encoding
160+
(def error-interceptor
161+
"Interceptor that writes grpc exception information in a grpc-web compatible encoding"
162+
(->Interceptor ::grpc-web-error
163+
identity
164+
error-leave-handler
165+
exception-handler))

‎src/protojure/pedestal/routes.clj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@
3636
name (keyword fqs (str method "-handler"))
3737
handler (handler name (partial method-fn callback-context))]
3838
[(str "/" fqs "/" method)
39-
:post (-> (consv grpc/error-interceptor interceptors)
39+
:post (-> (vec (concat [grpc.web/error-interceptor grpc/error-interceptor] interceptors))
4040
(conj grpc.web/proxy
4141
(grpc/route-interceptor rpc)
4242
handler))

‎test/protojure/grpc_web_test.clj‎

Lines changed: 48 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,18 @@
2626
[protojure.pedestal.routes :as pedestal.routes]
2727
[example.hello :refer [new-HelloRequest pb->HelloReply]]
2828
[clj-http.client :as client]
29-
[protojure.grpc.codec.lpm :as lpm]))
29+
[protojure.grpc.codec.lpm :as lpm]
30+
[protojure.test.grpc.TestService.server :as test.server]
31+
[protojure.pedestal.interceptors.grpc-web :as grpc.web]
32+
[taoensso.timbre.appenders.core :as appenders]
33+
[taoensso.timbre :as log]
34+
[taoensso.timbre.tools.logging :refer [use-timbre]]))
35+
36+
(use-timbre)
37+
38+
(log/set-config! {:level :trace
39+
:ns-whitelist ["protojure.*"]
40+
:appenders {:println (appenders/println-appender {:stream :auto})}})
3041

3142
(defonce test-env (atom {}))
3243

@@ -72,6 +83,21 @@
7283
:interceptors interceptors
7384
:callback-context (Greeter.)}))
7485

86+
;;-----------------------------------------------------------------------------
87+
;; TestService service endpoint
88+
;;-----------------------------------------------------------------------------
89+
90+
(deftype TestService []
91+
test.server/Service
92+
(Metadata
93+
[_ request]
94+
(throw (Exception. "foobar"))))
95+
96+
(defn- test-service-mock-routes [interceptors]
97+
(pedestal.routes/->tablesyntax {:rpc-metadata test.server/rpc-metadata
98+
:interceptors interceptors
99+
:callback-context (TestService.)}))
100+
75101
(defn- grpc-connect
76102
([] (grpc-connect (:port @test-env)))
77103
([port]
@@ -85,7 +111,9 @@
85111
interceptors [(body-params/body-params)
86112
pedestal/html-body]
87113
server-params {:env :prod
88-
::pedestal/routes (into #{} (greeter-mock-routes interceptors))
114+
::pedestal/routes (into #{} (concat
115+
(greeter-mock-routes interceptors)
116+
(test-service-mock-routes interceptors)))
89117
::pedestal/port port
90118

91119
::pedestal/type protojure.pedestal/config
@@ -122,14 +150,23 @@
122150
(let [lpm (async/<!! (async/into [] out))
123151
b64-encoded (-> (java.util.Base64/getEncoder)
124152
(.encode (byte-array lpm)))
125-
body (-> (java.util.Base64/getDecoder)
126-
(.decode (-> (client/post
127-
(str "http://localhost:" (:port @test-env) "/example.hello.Greeter/SayHello")
128-
{:body b64-encoded
129-
:content-type "application/grpc-web-text"
130-
:accept "application/grpc-web-text"})
131-
:body)))]
132-
(doseq [b body]
153+
body (-> (client/post
154+
(str "http://localhost:" (:port @test-env) "/example.hello.Greeter/SayHello")
155+
{:body b64-encoded
156+
:content-type "application/grpc-web-text"
157+
:accept "application/grpc-web-text"})
158+
:body)
159+
decoded-body (.decode (java.util.Base64/getDecoder) body)]
160+
(doseq [b (into [] decoded-body)]
133161
(async/>!! resp-in b))
134162
(async/close! resp-in)
135-
(is (= "Hello, World" (:message (async/<!! resp-out))))))))
163+
(is (= "Hello, World" (:message (async/<!! resp-out))))))))
164+
165+
(deftest grpc-web-exception-check
166+
(testing "Check that wff grpc-web trailers are received when a handler throws an exception"
167+
(let [resp (client/post
168+
(str "http://localhost:" (:port @test-env) "/protojure.test.grpc.TestService/Metadata")
169+
{:content-type "application/grpc-web-text"
170+
:accept "application/grpc-web-text"})]
171+
(is (= "gAAAAHxncnBjLXN0YXR1czoxMw0KZ3JwYy1tZXNzYWdlOmphdmEubGFuZy5FeGNlcHRpb24gaW4gSW50ZXJjZXB0b3IgOnByb3RvanVyZS50ZXN0LmdycGMuVGVzdFNlcnZpY2UvTWV0YWRhdGEtaGFuZGxlciAtIGZvb2Jhcg0K"
172+
(:body resp))))))

0 commit comments

Comments
 (0)