feat(grpc): Add gRPC routing and service support

.github/workflows/test.yml

@@ -20,6 +20,9 @@ jobs:
         with:
           cli: 1.11.1.1182
 
+      - name: Compile test Java sources
+        run: clojure -X:compile-test-java
+
       - name: Run tests
         run: clojure -X:test s-exp.hirundo-test-runner/run
 

README.md

@@ -76,6 +76,7 @@ middlewares out there.
 
 * `:tls` - A `io.helidon.nima.common.tls.Tls` instance
 
+* `:grpc-services` - vector of gRPC service descriptors (see [gRPC](#grpc) section)
 
 You can hook into the server builder via `s-exp.hirundo.options/set-server-option!`
 multimethod at runtime and add/modify whatever you want if you need anything
@@ -84,6 +85,133 @@ extra we don't provide (yet).
 http2 (h2 & h2c) is supported out of the box, iif a client connects with http2
 it will do the protocol switch automatically.
 
+## gRPC
+
+Hirundo supports hosting gRPC services on the same port as HTTP/WebSocket via
+Helidon's native gRPC module. You bring your own generated protobuf Java classes;
+hirundo passes raw protobuf objects to your handlers — serialization is your
+responsibility (e.g. wrap with [pronto](https://github.com/AppsFlyer/pronto) for
+idiomatic Clojure maps).
+
+```clojure
+(require '[s-exp.hirundo :as hirundo])
+(require '[s-exp.hirundo.grpc :as grpc])
+
+(hirundo/start!
+  {:grpc-services
+   [{:proto   MyProtoOuterClass/getDescriptor   ; Descriptors$FileDescriptor
+     :name    "Greeter"                          ; optional — defaults to proto service name
+     :methods {"SayHello"
+               {:type    :unary
+                :handler (fn [^HelloRequest req observer]
+                           (grpc/complete! observer (build-reply req)))}}}]})
+```
+
+### StreamObserver helpers
+
+`s-exp.hirundo.grpc` provides thin wrappers around `StreamObserver` so you don't
+have to call Java methods directly:
+
+| Function | Description |
+|---|---|
+| `(grpc/send! observer msg)` | Send a message to the client |
+| `(grpc/complete! observer)` | Signal successful stream completion |
+| `(grpc/complete! observer msg)` | Send one message then complete (unary shorthand) |
+| `(grpc/error! observer throwable)` | Signal an error |
+| `(grpc/stream-observer {:on-next f :on-error f :on-completed f})` | Build a `StreamObserver` from callback fns (all optional) |
+
+### Service descriptor
+
+Each entry in `:grpc-services` can be one of:
+
+* **A map** `{:proto <Descriptors$FileDescriptor> :name "ServiceName" :methods {...}}`
+* **A `GrpcService` instance** — passed through unchanged
+* **A zero-arg fn** — called once at startup, must return a descriptor map (useful for
+  stateful or dynamically-configured services)
+
+### Method types and handler signatures
+
+| `:type`         | Handler signature                                     |
+|-----------------|-------------------------------------------------------|
+| `:unary`        | `(fn [request ^StreamObserver observer])`             |
+| `:server-stream`| `(fn [request ^StreamObserver observer])`             |
+| `:client-stream`| `(fn [^StreamObserver response-observer]) => StreamObserver` |
+| `:bidi`         | `(fn [^StreamObserver response-observer]) => StreamObserver` |
+
+For `:client-stream` and `:bidi`, the handler receives the response observer and
+must return a `StreamObserver` that handles incoming client messages.
+
+### Example — server streaming
+
+```clojure
+{:proto   svc-file-descriptor
+ :name    "DataService"
+ :methods {"ListItems"
+           {:type    :server-stream
+            :handler (fn [^ListRequest req observer]
+                       (doseq [item (fetch-items req)]
+                         (grpc/send! observer item))
+                       (grpc/complete! observer))}}}
+```
+
+### core.async handler wrappers
+
+For handlers that naturally produce or consume streams of messages,
+`s-exp.hirundo.grpc` provides wrappers that expose `core.async` channels
+instead of raw `StreamObserver` methods:
+
+| Wrapper | Method type | User fn signature |
+|---|---|---|
+| `unary-async-handler` | `:unary` | `(fn [request out-ch])` |
+| `server-stream-async-handler` | `:server-stream` | `(fn [request out-ch])` |
+| `client-stream-async-handler` | `:client-stream` | `(fn [in-ch out-ch])` |
+| `bidi-async-handler` | `:bidi` | `(fn [in-ch out-ch])` |
+
+For unary/server-stream handlers: put messages onto `out-ch` then close it to
+end the stream. For client-stream/bidi handlers: incoming client messages
+arrive on `in-ch` (closed on completion or error); put responses onto `out-ch`
+and close it when done.
+
+All four wrappers accept optional kwargs to override the channel factories
+(zero-arg fns called once per request):
+* `:out-ch-fn` — factory for the outgoing channel (all wrappers)
+* `:in-ch-fn` — factory for the incoming channel (`client-stream-async-handler`, `bidi-async-handler`)
+
+```clojure
+;; unary — put one response, close
+{"SayHello"
+ {:type :unary
+  :handler (grpc/unary-async-handler
+            (fn [req out-ch]
+              (async/>!! out-ch (build-reply req))
+              (async/close! out-ch)))}}
+
+;; bidi — echo loop via go-loop
+{"Chat"
+ {:type :bidi
+  :handler (grpc/bidi-async-handler
+            (fn [in-ch out-ch]
+              (async/go-loop []
+                (if-some [msg (async/<! in-ch)]
+                  (do (async/>! out-ch (echo msg))
+                      (recur))
+                  (async/close! out-ch)))))}}
+```
+
+### Example — bidi streaming
+
+```clojure
+{:proto   svc-file-descriptor
+ :name    "ChatService"
+ :methods {"Chat"
+           {:type :bidi
+            :handler (fn [response-observer]
+                       (grpc/stream-observer
+                        {:on-next #(grpc/send! response-observer (echo %))
+                         :on-error #(grpc/error! response-observer %)
+                         :on-completed #(grpc/complete! response-observer)}))}}}
+```
+
 ## SSE (Server-Sent Events)
 
 Hirundo provides built-in SSE support via `s-exp.hirundo.sse/stream!`.

build-src/s_exp/hirundo/build_test.clj

@@ -0,0 +1,7 @@
+(ns s-exp.hirundo.build-test
+  (:require [clojure.tools.build.api :as b]))
+
+(defn compile-java [_]
+  (b/javac {:src-dirs  ["test/java"]
+            :class-dir "test/classes"
+            :basis     (b/create-basis {:aliases [:test]})}))

deps.edn

@@ -4,19 +4,25 @@
         io.helidon.http/helidon-http {:mvn/version "4.3.1"}
         io.helidon.webserver/helidon-webserver {:mvn/version "4.3.1"}
         io.helidon.webserver/helidon-webserver-websocket {:mvn/version "4.3.1"}
+        io.helidon.webserver/helidon-webserver-grpc {:mvn/version "4.3.1"}
         io.helidon.webserver/helidon-webserver-http2 {:mvn/version "4.3.1"}
         io.helidon.logging/helidon-logging-slf4j {:mvn/version "4.3.1"}
         org.ring-clojure/ring-core-protocols {:mvn/version "1.14.2"}
         com.aayushatharva.brotli4j/brotli4j {:mvn/version "1.18.0"}}
 
  :aliases
- {:test {:extra-paths ["test"]
+ {:test {:extra-paths ["test" "test/classes"]
          :extra-deps {org.clojure/test.check {:mvn/version "1.1.1"}
                       less-awful-ssl/less-awful-ssl {:mvn/version "1.0.6"}
                       eftest/eftest {:mvn/version "0.6.0"}
                       stylefruits/gniazdo {:mvn/version "1.2.2"}
-                      clj-http/clj-http {:mvn/version "3.12.0"}}
+                      clj-http/clj-http {:mvn/version "3.12.0"}
+                      io.grpc/grpc-netty-shaded {:mvn/version "1.73.0"}}
          :exec-fn s-exp.hirundo-test-runner/run}
+  :compile-test-java
+  {:extra-paths ["build-src"]
+   :deps {io.github.clojure/tools.build {:git/tag "v0.10.9" :git/sha "e405aac"}}
+   :exec-fn s-exp.hirundo.build-test/compile-java}
   :build
   {:deps {io.github.clojure/tools.build {:git/tag "v0.10.9" :git/sha "e405aac"}
           io.github.slipset/deps-deploy {:git/sha "07022b92d768590ab25b9ceb619ef17d2922da9a"}}

src/s_exp/hirundo.clj

@@ -1,5 +1,6 @@
 (ns s-exp.hirundo
-  (:require [s-exp.hirundo.http.routing]
+  (:require [s-exp.hirundo.grpc.routing]
+            [s-exp.hirundo.http.routing]
             [s-exp.hirundo.options :as options]
             [s-exp.hirundo.websocket]
             [s-exp.hirundo.websocket.routing])
@@ -25,6 +26,13 @@
   * `:http-handler` - ring http handler function
 
   * `:websocket-endpoints` - websocket endpoints (map-of string-endpoint handler-fns-map), where handler if can be of `:message`, `:ping`, `:pong`, `:close`, `:error`, `:open`, `:http-upgrade`. `handler-fns-map` can also contain 2 extra keys, `:extensions`, `:subprotocols`, which are sets of exts/subprotos acceptable by the server.
+
+  * `:grpc-services` - vector of gRPC service descriptors. Each entry can be:
+    - a map `{:proto <Descriptors$FileDescriptor> :name \"ServiceName\" :methods {\"Method\" {:type :unary|:server-stream|:client-stream|:bidi :handler <fn>}}}`
+    - a `GrpcService` instance
+    - a zero-arg fn returning a descriptor map
+    Handler fns for `:unary`/`:server-stream` take `[request response-observer]`;
+    for `:client-stream`/`:bidi` take `[response-observer]` and return a `StreamObserver`.
 
   * `:host` - host of the default socket
 

src/s_exp/hirundo/grpc.clj

@@ -0,0 +1,137 @@
+(ns s-exp.hirundo.grpc
+  (:require [clojure.core.async :as async])
+  (:import (io.grpc.stub StreamObserver)))
+
+(set! *warn-on-reflection* true)
+
+(defn send!
+  "Sends `msg` to the client via `observer`."
+  [^StreamObserver observer msg]
+  (.onNext observer msg))
+
+(defn complete!
+  "Signals successful stream completion to the client. With 2 args, sends
+  `msg` first then completes — convenience for unary and single-response
+  server-streaming handlers."
+  ([^StreamObserver observer]
+   (.onCompleted observer))
+  ([^StreamObserver observer msg]
+   (.onNext observer msg)
+   (.onCompleted observer)))
+
+(defn error!
+  "Signals an error to the client via `observer`. `throwable` must be a
+  `java.lang.Throwable`."
+  [^StreamObserver observer ^Throwable throwable]
+  (.onError observer throwable))
+
+(defn stream-observer
+  "Returns a `StreamObserver` backed by callback fns supplied as a map:
+
+   * `:on-next` — `(fn [msg])` — called for each incoming message
+   * `:on-error` — `(fn [throwable])` — called on stream error
+   * `:on-completed` — `(fn [])` — called when the client signals completion
+
+  All keys are optional; unset callbacks are no-ops."
+  ^StreamObserver [{:keys [on-next on-error on-completed]}]
+  (reify StreamObserver
+    (onNext [_ msg]
+      (when on-next (on-next msg)))
+    (onError [_ t]
+      (when on-error (on-error t)))
+    (onCompleted [_]
+      (when on-completed (on-completed)))))
+
+;;; core.async handler wrappers ------------------------------------------------
+
+(defn- drain-to-observer!
+  "Reads msgs from `out-ch` and forwards to `observer` until the channel
+  closes, then completes the stream. Runs on an io-thread."
+  [^StreamObserver observer out-ch]
+  (async/io-thread
+   (try
+     (loop []
+       (when-some [msg (async/<!! out-ch)]
+         (send! observer msg)
+         (recur)))
+     (complete! observer)
+     (catch Throwable t
+       (error! observer t)))))
+
+(defn unary-async-handler
+  "Wraps `(fn [request out-ch])` as a `:unary` handler.
+
+  The user fn receives the request and a `core.async` channel. Put exactly
+  one response message onto `out-ch` then close it. Closing `out-ch`
+  completes the stream.
+
+  Options:
+  * `:out-ch-fn` — zero-arg fn returning the outgoing channel (default: `#(async/chan 1)`)"
+  [f & {:keys [out-ch-fn] :or {out-ch-fn #(async/chan 1)}}]
+  (fn [req ^StreamObserver observer]
+    (let [out-ch (out-ch-fn)]
+      (drain-to-observer! observer out-ch)
+      (f req out-ch))))
+
+(defn server-stream-async-handler
+  "Wraps `(fn [request out-ch])` as a `:server-stream` handler.
+
+  The user fn receives the request and a `core.async` channel. Put any
+  number of response messages onto `out-ch` then close it to end the stream.
+
+  Options:
+  * `:out-ch-fn` — zero-arg fn returning the outgoing channel (default: `#(async/chan 16)`)"
+  [f & {:keys [out-ch-fn] :or {out-ch-fn #(async/chan 16)}}]
+  (fn [req ^StreamObserver observer]
+    (let [out-ch (out-ch-fn)]
+      (drain-to-observer! observer out-ch)
+      (f req out-ch))))
+
+(defn- input-observer
+  "Returns a StreamObserver that feeds incoming messages into `in-ch` and
+  closes it on completion or error."
+  [in-ch]
+  (stream-observer
+   {:on-next #(async/>!! in-ch %)
+    :on-error (fn [_] (async/close! in-ch))
+    :on-completed (fn [] (async/close! in-ch))}))
+
+(defn client-stream-async-handler
+  "Wraps `(fn [in-ch out-ch])` as a `:client-stream` handler.
+
+  Incoming client messages arrive on `in-ch`; it is closed when the client
+  signals completion or an error occurs. Put response messages onto `out-ch`
+  and close it to end the stream.
+
+  Options:
+  * `:in-ch-fn`  — zero-arg fn returning the incoming channel (default: `#(async/chan 16)`)
+  * `:out-ch-fn` — zero-arg fn returning the outgoing channel (default: `#(async/chan 1)`)"
+  [f & {:keys [in-ch-fn out-ch-fn]
+        :or {in-ch-fn #(async/chan 16)
+             out-ch-fn #(async/chan 1)}}]
+  (fn [^StreamObserver observer]
+    (let [in-ch (in-ch-fn)
+          out-ch (out-ch-fn)]
+      (drain-to-observer! observer out-ch)
+      (f in-ch out-ch)
+      (input-observer in-ch))))
+
+(defn bidi-async-handler
+  "Wraps `(fn [in-ch out-ch])` as a `:bidi` handler.
+
+  Incoming client messages arrive on `in-ch`; it is closed when the client
+  signals completion or an error occurs. Put response messages onto `out-ch`
+  and close it to end the stream.
+
+  Options:
+  * `:in-ch-fn`  — zero-arg fn returning the incoming channel (default: `#(async/chan 16)`)
+  * `:out-ch-fn` — zero-arg fn returning the outgoing channel (default: `#(async/chan 16)`)"
+  [f & {:keys [in-ch-fn out-ch-fn]
+        :or {in-ch-fn #(async/chan 16)
+             out-ch-fn #(async/chan 16)}}]
+  (fn [^StreamObserver observer]
+    (let [in-ch (in-ch-fn)
+          out-ch (out-ch-fn)]
+      (drain-to-observer! observer out-ch)
+      (f in-ch out-ch)
+      (input-observer in-ch))))

src/s_exp/hirundo/grpc/routing.clj

@@ -0,0 +1,39 @@
+(ns s-exp.hirundo.grpc.routing
+  (:require [s-exp.hirundo.grpc.service :as svc]
+            [s-exp.hirundo.options :as options])
+  (:import (io.helidon.webserver WebServerConfig$Builder)
+           (io.helidon.webserver.grpc GrpcRouting GrpcRouting$Builder GrpcService)))
+
+(set! *warn-on-reflection* true)
+
+(defn- add-service!
+  [^GrpcRouting$Builder grpc-builder service-descriptor]
+  (let [^GrpcService grpc-svc
+        (cond
+          (map? service-descriptor)
+          (svc/service service-descriptor)
+
+          (instance? GrpcService service-descriptor)
+          service-descriptor
+
+          (ifn? service-descriptor)
+          (svc/service (service-descriptor))
+
+          :else
+          (throw
+           (ex-info (format "Invalid gRPC service descriptor type: %s"
+                            (type service-descriptor))
+                    {:type :s-exp.hirundo.grpc.routing/invalid-service})))]
+    (.service grpc-builder grpc-svc)))
+
+(defn set-grpc-services!
+  ^WebServerConfig$Builder
+  [^WebServerConfig$Builder builder services _options]
+  (doto builder
+    (.addRouting
+     ^GrpcRouting$Builder
+     (reduce add-service! (GrpcRouting/builder) services))))
+
+(defmethod options/set-server-option! :grpc-services
+  [^WebServerConfig$Builder builder _ services options]
+  (set-grpc-services! builder services options))