diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 45f8568..3c1f7a9 100644 --- a/.github/workflows/test.yml +++ b/.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 diff --git a/README.md b/README.md index 21db949..f7bfbcc 100644 --- a/README.md +++ b/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 :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/! 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!`. diff --git a/build-src/s_exp/hirundo/build_test.clj b/build-src/s_exp/hirundo/build_test.clj new file mode 100644 index 0000000..fedd23e --- /dev/null +++ b/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]})})) diff --git a/deps.edn b/deps.edn index 41eefed..d29a4a5 100644 --- a/deps.edn +++ b/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"}} diff --git a/src/s_exp/hirundo.clj b/src/s_exp/hirundo.clj index 7ce01dd..488416b 100644 --- a/src/s_exp/hirundo.clj +++ b/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 :name \"ServiceName\" :methods {\"Method\" {:type :unary|:server-stream|:client-stream|:bidi :handler }}}` + - 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 diff --git a/src/s_exp/hirundo/grpc.clj b/src/s_exp/hirundo/grpc.clj new file mode 100644 index 0000000..5f9f4d2 --- /dev/null +++ b/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/!! 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)))) diff --git a/src/s_exp/hirundo/grpc/routing.clj b/src/s_exp/hirundo/grpc/routing.clj new file mode 100644 index 0000000..ca69f97 --- /dev/null +++ b/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)) diff --git a/src/s_exp/hirundo/grpc/service.clj b/src/s_exp/hirundo/grpc/service.clj new file mode 100644 index 0000000..01dc60c --- /dev/null +++ b/src/s_exp/hirundo/grpc/service.clj @@ -0,0 +1,60 @@ +(ns s-exp.hirundo.grpc.service + (:import (com.google.protobuf Descriptors$FileDescriptor) + (io.grpc.stub ServerCalls$UnaryMethod + ServerCalls$ServerStreamingMethod + ServerCalls$ClientStreamingMethod + ServerCalls$BidiStreamingMethod) + (io.helidon.webserver.grpc GrpcService GrpcService$Routing))) + +(set! *warn-on-reflection* true) + +(defn- register-method! + [^GrpcService$Routing routing method-name {:keys [type handler]}] + (case type + :unary + (.unary routing ^String method-name + (reify ServerCalls$UnaryMethod + (invoke [_ req response-observer] + (handler req response-observer)))) + :server-stream + (.serverStream routing ^String method-name + (reify ServerCalls$ServerStreamingMethod + (invoke [_ req response-observer] + (handler req response-observer)))) + :client-stream + (.clientStream routing ^String method-name + (reify ServerCalls$ClientStreamingMethod + (invoke [_ response-observer] + (handler response-observer)))) + :bidi + (.bidi routing ^String method-name + (reify ServerCalls$BidiStreamingMethod + (invoke [_ response-observer] + (handler response-observer)))) + (throw (ex-info (str "Unknown gRPC method type: " type) + {:type :s-exp.hirundo.grpc/unknown-method-type + :method-name method-name})))) + +(defn service + "Creates a `GrpcService` from a descriptor map: + + {:proto ^Descriptors$FileDescriptor ; proto file descriptor + :name \"MyService\" ; optional service name + :methods {\"MethodName\" {:type :unary | :server-stream | :client-stream | :bidi + :handler }}} + + Handler signatures: + - `:unary`, `:server-stream` — `(fn [request ^StreamObserver response-observer])` + - `:client-stream`, `:bidi` — `(fn [^StreamObserver response-observer]) => StreamObserver` + + Handlers receive raw protobuf objects. Use pronto or direct Java interop to + encode/decode messages within the handler." + ^GrpcService + [{:keys [^Descriptors$FileDescriptor proto name methods]}] + (reify GrpcService + (proto [_] proto) + (serviceName [_] (or name "")) + (update [_ routing] + (run! (fn [[method-name method-descriptor]] + (register-method! routing method-name method-descriptor)) + methods)))) diff --git a/test/java/hirundo/test/Person.java b/test/java/hirundo/test/Person.java new file mode 100644 index 0000000..933cbc2 --- /dev/null +++ b/test/java/hirundo/test/Person.java @@ -0,0 +1,746 @@ +// Generated by the protocol buffer compiler. DO NOT EDIT! +// source: person.proto + +package hirundo.test; + +/** + * Protobuf type {@code hirundo.test.Person} + */ +public final class Person extends + com.google.protobuf.GeneratedMessageV3 implements + // @@protoc_insertion_point(message_implements:hirundo.test.Person) + PersonOrBuilder { +private static final long serialVersionUID = 0L; + // Use Person.newBuilder() to construct. + private Person(com.google.protobuf.GeneratedMessageV3.Builder builder) { + super(builder); + } + private Person() { + name_ = ""; + email_ = ""; + } + + @java.lang.Override + @SuppressWarnings({"unused"}) + protected java.lang.Object newInstance( + UnusedPrivateParameter unused) { + return new Person(); + } + + @java.lang.Override + public final com.google.protobuf.UnknownFieldSet + getUnknownFields() { + return this.unknownFields; + } + public static final com.google.protobuf.Descriptors.Descriptor + getDescriptor() { + return hirundo.test.PersonOuterClass.internal_static_hirundo_test_Person_descriptor; + } + + @java.lang.Override + protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable + internalGetFieldAccessorTable() { + return hirundo.test.PersonOuterClass.internal_static_hirundo_test_Person_fieldAccessorTable + .ensureFieldAccessorsInitialized( + hirundo.test.Person.class, hirundo.test.Person.Builder.class); + } + + public static final int ID_FIELD_NUMBER = 1; + private int id_ = 0; + /** + * int32 id = 1; + * @return The id. + */ + @java.lang.Override + public int getId() { + return id_; + } + + public static final int NAME_FIELD_NUMBER = 2; + @SuppressWarnings("serial") + private volatile java.lang.Object name_ = ""; + /** + * string name = 2; + * @return The name. + */ + @java.lang.Override + public java.lang.String getName() { + java.lang.Object ref = name_; + if (ref instanceof java.lang.String) { + return (java.lang.String) ref; + } else { + com.google.protobuf.ByteString bs = + (com.google.protobuf.ByteString) ref; + java.lang.String s = bs.toStringUtf8(); + name_ = s; + return s; + } + } + /** + * string name = 2; + * @return The bytes for name. + */ + @java.lang.Override + public com.google.protobuf.ByteString + getNameBytes() { + java.lang.Object ref = name_; + if (ref instanceof java.lang.String) { + com.google.protobuf.ByteString b = + com.google.protobuf.ByteString.copyFromUtf8( + (java.lang.String) ref); + name_ = b; + return b; + } else { + return (com.google.protobuf.ByteString) ref; + } + } + + public static final int EMAIL_FIELD_NUMBER = 3; + @SuppressWarnings("serial") + private volatile java.lang.Object email_ = ""; + /** + * string email = 3; + * @return The email. + */ + @java.lang.Override + public java.lang.String getEmail() { + java.lang.Object ref = email_; + if (ref instanceof java.lang.String) { + return (java.lang.String) ref; + } else { + com.google.protobuf.ByteString bs = + (com.google.protobuf.ByteString) ref; + java.lang.String s = bs.toStringUtf8(); + email_ = s; + return s; + } + } + /** + * string email = 3; + * @return The bytes for email. + */ + @java.lang.Override + public com.google.protobuf.ByteString + getEmailBytes() { + java.lang.Object ref = email_; + if (ref instanceof java.lang.String) { + com.google.protobuf.ByteString b = + com.google.protobuf.ByteString.copyFromUtf8( + (java.lang.String) ref); + email_ = b; + return b; + } else { + return (com.google.protobuf.ByteString) ref; + } + } + + private byte memoizedIsInitialized = -1; + @java.lang.Override + public final boolean isInitialized() { + byte isInitialized = memoizedIsInitialized; + if (isInitialized == 1) return true; + if (isInitialized == 0) return false; + + memoizedIsInitialized = 1; + return true; + } + + @java.lang.Override + public void writeTo(com.google.protobuf.CodedOutputStream output) + throws java.io.IOException { + if (id_ != 0) { + output.writeInt32(1, id_); + } + if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(name_)) { + com.google.protobuf.GeneratedMessageV3.writeString(output, 2, name_); + } + if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(email_)) { + com.google.protobuf.GeneratedMessageV3.writeString(output, 3, email_); + } + getUnknownFields().writeTo(output); + } + + @java.lang.Override + public int getSerializedSize() { + int size = memoizedSize; + if (size != -1) return size; + + size = 0; + if (id_ != 0) { + size += com.google.protobuf.CodedOutputStream + .computeInt32Size(1, id_); + } + if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(name_)) { + size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, name_); + } + if (!com.google.protobuf.GeneratedMessageV3.isStringEmpty(email_)) { + size += com.google.protobuf.GeneratedMessageV3.computeStringSize(3, email_); + } + size += getUnknownFields().getSerializedSize(); + memoizedSize = size; + return size; + } + + @java.lang.Override + public boolean equals(final java.lang.Object obj) { + if (obj == this) { + return true; + } + if (!(obj instanceof hirundo.test.Person)) { + return super.equals(obj); + } + hirundo.test.Person other = (hirundo.test.Person) obj; + + if (getId() + != other.getId()) return false; + if (!getName() + .equals(other.getName())) return false; + if (!getEmail() + .equals(other.getEmail())) return false; + if (!getUnknownFields().equals(other.getUnknownFields())) return false; + return true; + } + + @java.lang.Override + public int hashCode() { + if (memoizedHashCode != 0) { + return memoizedHashCode; + } + int hash = 41; + hash = (19 * hash) + getDescriptor().hashCode(); + hash = (37 * hash) + ID_FIELD_NUMBER; + hash = (53 * hash) + getId(); + hash = (37 * hash) + NAME_FIELD_NUMBER; + hash = (53 * hash) + getName().hashCode(); + hash = (37 * hash) + EMAIL_FIELD_NUMBER; + hash = (53 * hash) + getEmail().hashCode(); + hash = (29 * hash) + getUnknownFields().hashCode(); + memoizedHashCode = hash; + return hash; + } + + public static hirundo.test.Person parseFrom( + java.nio.ByteBuffer data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + public static hirundo.test.Person parseFrom( + java.nio.ByteBuffer data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + public static hirundo.test.Person parseFrom( + com.google.protobuf.ByteString data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + public static hirundo.test.Person parseFrom( + com.google.protobuf.ByteString data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + public static hirundo.test.Person parseFrom(byte[] data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + public static hirundo.test.Person parseFrom( + byte[] data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + public static hirundo.test.Person parseFrom(java.io.InputStream input) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3 + .parseWithIOException(PARSER, input); + } + public static hirundo.test.Person parseFrom( + java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3 + .parseWithIOException(PARSER, input, extensionRegistry); + } + public static hirundo.test.Person parseDelimitedFrom(java.io.InputStream input) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3 + .parseDelimitedWithIOException(PARSER, input); + } + public static hirundo.test.Person parseDelimitedFrom( + java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3 + .parseDelimitedWithIOException(PARSER, input, extensionRegistry); + } + public static hirundo.test.Person parseFrom( + com.google.protobuf.CodedInputStream input) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3 + .parseWithIOException(PARSER, input); + } + public static hirundo.test.Person parseFrom( + com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3 + .parseWithIOException(PARSER, input, extensionRegistry); + } + + @java.lang.Override + public Builder newBuilderForType() { return newBuilder(); } + public static Builder newBuilder() { + return DEFAULT_INSTANCE.toBuilder(); + } + public static Builder newBuilder(hirundo.test.Person prototype) { + return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype); + } + @java.lang.Override + public Builder toBuilder() { + return this == DEFAULT_INSTANCE + ? new Builder() : new Builder().mergeFrom(this); + } + + @java.lang.Override + protected Builder newBuilderForType( + com.google.protobuf.GeneratedMessageV3.BuilderParent parent) { + Builder builder = new Builder(parent); + return builder; + } + /** + * Protobuf type {@code hirundo.test.Person} + */ + public static final class Builder extends + com.google.protobuf.GeneratedMessageV3.Builder implements + // @@protoc_insertion_point(builder_implements:hirundo.test.Person) + hirundo.test.PersonOrBuilder { + public static final com.google.protobuf.Descriptors.Descriptor + getDescriptor() { + return hirundo.test.PersonOuterClass.internal_static_hirundo_test_Person_descriptor; + } + + @java.lang.Override + protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable + internalGetFieldAccessorTable() { + return hirundo.test.PersonOuterClass.internal_static_hirundo_test_Person_fieldAccessorTable + .ensureFieldAccessorsInitialized( + hirundo.test.Person.class, hirundo.test.Person.Builder.class); + } + + // Construct using hirundo.test.Person.newBuilder() + private Builder() { + + } + + private Builder( + com.google.protobuf.GeneratedMessageV3.BuilderParent parent) { + super(parent); + + } + @java.lang.Override + public Builder clear() { + super.clear(); + bitField0_ = 0; + id_ = 0; + name_ = ""; + email_ = ""; + return this; + } + + @java.lang.Override + public com.google.protobuf.Descriptors.Descriptor + getDescriptorForType() { + return hirundo.test.PersonOuterClass.internal_static_hirundo_test_Person_descriptor; + } + + @java.lang.Override + public hirundo.test.Person getDefaultInstanceForType() { + return hirundo.test.Person.getDefaultInstance(); + } + + @java.lang.Override + public hirundo.test.Person build() { + hirundo.test.Person result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException(result); + } + return result; + } + + @java.lang.Override + public hirundo.test.Person buildPartial() { + hirundo.test.Person result = new hirundo.test.Person(this); + if (bitField0_ != 0) { buildPartial0(result); } + onBuilt(); + return result; + } + + private void buildPartial0(hirundo.test.Person result) { + int from_bitField0_ = bitField0_; + if (((from_bitField0_ & 0x00000001) != 0)) { + result.id_ = id_; + } + if (((from_bitField0_ & 0x00000002) != 0)) { + result.name_ = name_; + } + if (((from_bitField0_ & 0x00000004) != 0)) { + result.email_ = email_; + } + } + + @java.lang.Override + public Builder clone() { + return super.clone(); + } + @java.lang.Override + public Builder setField( + com.google.protobuf.Descriptors.FieldDescriptor field, + java.lang.Object value) { + return super.setField(field, value); + } + @java.lang.Override + public Builder clearField( + com.google.protobuf.Descriptors.FieldDescriptor field) { + return super.clearField(field); + } + @java.lang.Override + public Builder clearOneof( + com.google.protobuf.Descriptors.OneofDescriptor oneof) { + return super.clearOneof(oneof); + } + @java.lang.Override + public Builder setRepeatedField( + com.google.protobuf.Descriptors.FieldDescriptor field, + int index, java.lang.Object value) { + return super.setRepeatedField(field, index, value); + } + @java.lang.Override + public Builder addRepeatedField( + com.google.protobuf.Descriptors.FieldDescriptor field, + java.lang.Object value) { + return super.addRepeatedField(field, value); + } + @java.lang.Override + public Builder mergeFrom(com.google.protobuf.Message other) { + if (other instanceof hirundo.test.Person) { + return mergeFrom((hirundo.test.Person)other); + } else { + super.mergeFrom(other); + return this; + } + } + + public Builder mergeFrom(hirundo.test.Person other) { + if (other == hirundo.test.Person.getDefaultInstance()) return this; + if (other.getId() != 0) { + setId(other.getId()); + } + if (!other.getName().isEmpty()) { + name_ = other.name_; + bitField0_ |= 0x00000002; + onChanged(); + } + if (!other.getEmail().isEmpty()) { + email_ = other.email_; + bitField0_ |= 0x00000004; + onChanged(); + } + this.mergeUnknownFields(other.getUnknownFields()); + onChanged(); + return this; + } + + @java.lang.Override + public final boolean isInitialized() { + return true; + } + + @java.lang.Override + public Builder mergeFrom( + com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + if (extensionRegistry == null) { + throw new java.lang.NullPointerException(); + } + try { + boolean done = false; + while (!done) { + int tag = input.readTag(); + switch (tag) { + case 0: + done = true; + break; + case 8: { + id_ = input.readInt32(); + bitField0_ |= 0x00000001; + break; + } // case 8 + case 18: { + name_ = input.readStringRequireUtf8(); + bitField0_ |= 0x00000002; + break; + } // case 18 + case 26: { + email_ = input.readStringRequireUtf8(); + bitField0_ |= 0x00000004; + break; + } // case 26 + default: { + if (!super.parseUnknownField(input, extensionRegistry, tag)) { + done = true; // was an endgroup tag + } + break; + } // default: + } // switch (tag) + } // while (!done) + } catch (com.google.protobuf.InvalidProtocolBufferException e) { + throw e.unwrapIOException(); + } finally { + onChanged(); + } // finally + return this; + } + private int bitField0_; + + private int id_ ; + /** + * int32 id = 1; + * @return The id. + */ + @java.lang.Override + public int getId() { + return id_; + } + /** + * int32 id = 1; + * @param value The id to set. + * @return This builder for chaining. + */ + public Builder setId(int value) { + + id_ = value; + bitField0_ |= 0x00000001; + onChanged(); + return this; + } + /** + * int32 id = 1; + * @return This builder for chaining. + */ + public Builder clearId() { + bitField0_ = (bitField0_ & ~0x00000001); + id_ = 0; + onChanged(); + return this; + } + + private java.lang.Object name_ = ""; + /** + * string name = 2; + * @return The name. + */ + public java.lang.String getName() { + java.lang.Object ref = name_; + if (!(ref instanceof java.lang.String)) { + com.google.protobuf.ByteString bs = + (com.google.protobuf.ByteString) ref; + java.lang.String s = bs.toStringUtf8(); + name_ = s; + return s; + } else { + return (java.lang.String) ref; + } + } + /** + * string name = 2; + * @return The bytes for name. + */ + public com.google.protobuf.ByteString + getNameBytes() { + java.lang.Object ref = name_; + if (ref instanceof String) { + com.google.protobuf.ByteString b = + com.google.protobuf.ByteString.copyFromUtf8( + (java.lang.String) ref); + name_ = b; + return b; + } else { + return (com.google.protobuf.ByteString) ref; + } + } + /** + * string name = 2; + * @param value The name to set. + * @return This builder for chaining. + */ + public Builder setName( + java.lang.String value) { + if (value == null) { throw new NullPointerException(); } + name_ = value; + bitField0_ |= 0x00000002; + onChanged(); + return this; + } + /** + * string name = 2; + * @return This builder for chaining. + */ + public Builder clearName() { + name_ = getDefaultInstance().getName(); + bitField0_ = (bitField0_ & ~0x00000002); + onChanged(); + return this; + } + /** + * string name = 2; + * @param value The bytes for name to set. + * @return This builder for chaining. + */ + public Builder setNameBytes( + com.google.protobuf.ByteString value) { + if (value == null) { throw new NullPointerException(); } + checkByteStringIsUtf8(value); + name_ = value; + bitField0_ |= 0x00000002; + onChanged(); + return this; + } + + private java.lang.Object email_ = ""; + /** + * string email = 3; + * @return The email. + */ + public java.lang.String getEmail() { + java.lang.Object ref = email_; + if (!(ref instanceof java.lang.String)) { + com.google.protobuf.ByteString bs = + (com.google.protobuf.ByteString) ref; + java.lang.String s = bs.toStringUtf8(); + email_ = s; + return s; + } else { + return (java.lang.String) ref; + } + } + /** + * string email = 3; + * @return The bytes for email. + */ + public com.google.protobuf.ByteString + getEmailBytes() { + java.lang.Object ref = email_; + if (ref instanceof String) { + com.google.protobuf.ByteString b = + com.google.protobuf.ByteString.copyFromUtf8( + (java.lang.String) ref); + email_ = b; + return b; + } else { + return (com.google.protobuf.ByteString) ref; + } + } + /** + * string email = 3; + * @param value The email to set. + * @return This builder for chaining. + */ + public Builder setEmail( + java.lang.String value) { + if (value == null) { throw new NullPointerException(); } + email_ = value; + bitField0_ |= 0x00000004; + onChanged(); + return this; + } + /** + * string email = 3; + * @return This builder for chaining. + */ + public Builder clearEmail() { + email_ = getDefaultInstance().getEmail(); + bitField0_ = (bitField0_ & ~0x00000004); + onChanged(); + return this; + } + /** + * string email = 3; + * @param value The bytes for email to set. + * @return This builder for chaining. + */ + public Builder setEmailBytes( + com.google.protobuf.ByteString value) { + if (value == null) { throw new NullPointerException(); } + checkByteStringIsUtf8(value); + email_ = value; + bitField0_ |= 0x00000004; + onChanged(); + return this; + } + @java.lang.Override + public final Builder setUnknownFields( + final com.google.protobuf.UnknownFieldSet unknownFields) { + return super.setUnknownFields(unknownFields); + } + + @java.lang.Override + public final Builder mergeUnknownFields( + final com.google.protobuf.UnknownFieldSet unknownFields) { + return super.mergeUnknownFields(unknownFields); + } + + + // @@protoc_insertion_point(builder_scope:hirundo.test.Person) + } + + // @@protoc_insertion_point(class_scope:hirundo.test.Person) + private static final hirundo.test.Person DEFAULT_INSTANCE; + static { + DEFAULT_INSTANCE = new hirundo.test.Person(); + } + + public static hirundo.test.Person getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private static final com.google.protobuf.Parser + PARSER = new com.google.protobuf.AbstractParser() { + @java.lang.Override + public Person parsePartialFrom( + com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + Builder builder = newBuilder(); + try { + builder.mergeFrom(input, extensionRegistry); + } catch (com.google.protobuf.InvalidProtocolBufferException e) { + throw e.setUnfinishedMessage(builder.buildPartial()); + } catch (com.google.protobuf.UninitializedMessageException e) { + throw e.asInvalidProtocolBufferException().setUnfinishedMessage(builder.buildPartial()); + } catch (java.io.IOException e) { + throw new com.google.protobuf.InvalidProtocolBufferException(e) + .setUnfinishedMessage(builder.buildPartial()); + } + return builder.buildPartial(); + } + }; + + public static com.google.protobuf.Parser parser() { + return PARSER; + } + + @java.lang.Override + public com.google.protobuf.Parser getParserForType() { + return PARSER; + } + + @java.lang.Override + public hirundo.test.Person getDefaultInstanceForType() { + return DEFAULT_INSTANCE; + } + +} + diff --git a/test/java/hirundo/test/PersonOrBuilder.java b/test/java/hirundo/test/PersonOrBuilder.java new file mode 100644 index 0000000..352ab73 --- /dev/null +++ b/test/java/hirundo/test/PersonOrBuilder.java @@ -0,0 +1,39 @@ +// Generated by the protocol buffer compiler. DO NOT EDIT! +// source: person.proto + +package hirundo.test; + +public interface PersonOrBuilder extends + // @@protoc_insertion_point(interface_extends:hirundo.test.Person) + com.google.protobuf.MessageOrBuilder { + + /** + * int32 id = 1; + * @return The id. + */ + int getId(); + + /** + * string name = 2; + * @return The name. + */ + java.lang.String getName(); + /** + * string name = 2; + * @return The bytes for name. + */ + com.google.protobuf.ByteString + getNameBytes(); + + /** + * string email = 3; + * @return The email. + */ + java.lang.String getEmail(); + /** + * string email = 3; + * @return The bytes for email. + */ + com.google.protobuf.ByteString + getEmailBytes(); +} diff --git a/test/java/hirundo/test/PersonOuterClass.java b/test/java/hirundo/test/PersonOuterClass.java new file mode 100644 index 0000000..5902fda --- /dev/null +++ b/test/java/hirundo/test/PersonOuterClass.java @@ -0,0 +1,48 @@ +// Generated by the protocol buffer compiler. DO NOT EDIT! +// source: person.proto + +package hirundo.test; + +public final class PersonOuterClass { + private PersonOuterClass() {} + public static void registerAllExtensions( + com.google.protobuf.ExtensionRegistryLite registry) { + } + + public static void registerAllExtensions( + com.google.protobuf.ExtensionRegistry registry) { + registerAllExtensions( + (com.google.protobuf.ExtensionRegistryLite) registry); + } + static final com.google.protobuf.Descriptors.Descriptor + internal_static_hirundo_test_Person_descriptor; + static final + com.google.protobuf.GeneratedMessageV3.FieldAccessorTable + internal_static_hirundo_test_Person_fieldAccessorTable; + + public static com.google.protobuf.Descriptors.FileDescriptor + getDescriptor() { + return descriptor; + } + private static com.google.protobuf.Descriptors.FileDescriptor + descriptor; + static { + java.lang.String[] descriptorData = { + "\n\014person.proto\022\014hirundo.test\"1\n\006Person\022\n" + + "\n\002id\030\001 \001(\005\022\014\n\004name\030\002 \001(\t\022\r\n\005email\030\003 \001(\tB" + + "\020\n\014hirundo.testP\001b\006proto3" + }; + descriptor = com.google.protobuf.Descriptors.FileDescriptor + .internalBuildGeneratedFileFrom(descriptorData, + new com.google.protobuf.Descriptors.FileDescriptor[] { + }); + internal_static_hirundo_test_Person_descriptor = + getDescriptor().getMessageTypes().get(0); + internal_static_hirundo_test_Person_fieldAccessorTable = new + com.google.protobuf.GeneratedMessageV3.FieldAccessorTable( + internal_static_hirundo_test_Person_descriptor, + new java.lang.String[] { "Id", "Name", "Email", }); + } + + // @@protoc_insertion_point(outer_class_scope) +} diff --git a/test/proto/person.proto b/test/proto/person.proto new file mode 100644 index 0000000..77a6ec8 --- /dev/null +++ b/test/proto/person.proto @@ -0,0 +1,12 @@ +syntax = "proto3"; + +package hirundo.test; + +option java_package = "hirundo.test"; +option java_multiple_files = true; + +message Person { + int32 id = 1; + string name = 2; + string email = 3; +} diff --git a/test/s_exp/hirundo/grpc/grpc_test.clj b/test/s_exp/hirundo/grpc/grpc_test.clj new file mode 100644 index 0000000..539e3c9 --- /dev/null +++ b/test/s_exp/hirundo/grpc/grpc_test.clj @@ -0,0 +1,250 @@ +(ns s-exp.hirundo.grpc.grpc-test + (:require [clojure.core.async :as async] + [clojure.test :refer [deftest is]] + [s-exp.hirundo :as m] + [s-exp.hirundo.grpc :as grpc]) + (:import (com.google.protobuf DescriptorProtos$FileDescriptorProto + DescriptorProtos$MethodDescriptorProto + DescriptorProtos$ServiceDescriptorProto + Descriptors$FileDescriptor) + (hirundo.test Person PersonOuterClass) + (io.grpc CallOptions ManagedChannel ManagedChannelBuilder + MethodDescriptor MethodDescriptor$MethodType) + (io.grpc.protobuf ProtoUtils) + (io.grpc.stub ClientCalls StreamObserver))) + +;;; ---- Proto / service descriptor setup ------------------------------------- + +(defn- service-file-descriptor + "Builds a FileDescriptor with PersonService (Echo, StreamEcho, BidiEcho) + layered on top of person.proto." + ^Descriptors$FileDescriptor [] + (Descriptors$FileDescriptor/buildFrom + (-> (DescriptorProtos$FileDescriptorProto/newBuilder) + (.setName "person_service.proto") + (.setSyntax "proto3") + (.addDependency "person.proto") + (.addService + (-> (DescriptorProtos$ServiceDescriptorProto/newBuilder) + (.setName "PersonService") + (.addMethod (-> (DescriptorProtos$MethodDescriptorProto/newBuilder) + (.setName "Echo") + (.setInputType ".hirundo.test.Person") + (.setOutputType ".hirundo.test.Person") + .build)) + (.addMethod (-> (DescriptorProtos$MethodDescriptorProto/newBuilder) + (.setName "StreamEcho") + (.setInputType ".hirundo.test.Person") + (.setOutputType ".hirundo.test.Person") + (.setServerStreaming true) + .build)) + (.addMethod (-> (DescriptorProtos$MethodDescriptorProto/newBuilder) + (.setName "BidiEcho") + (.setInputType ".hirundo.test.Person") + (.setOutputType ".hirundo.test.Person") + (.setClientStreaming true) + (.setServerStreaming true) + .build)) + .build)) + .build) + (into-array Descriptors$FileDescriptor [(PersonOuterClass/getDescriptor)]))) + +(def ^Descriptors$FileDescriptor svc-fd (service-file-descriptor)) + +;;; ---- gRPC client helpers -------------------------------------------------- + +(defn- method-descriptor + ^MethodDescriptor [service-name method-name method-type] + (let [marshaller (ProtoUtils/marshaller (Person/getDefaultInstance))] + (-> (MethodDescriptor/newBuilder marshaller marshaller) + (.setType method-type) + (.setFullMethodName (MethodDescriptor/generateFullMethodName service-name method-name)) + .build))) + +(defn- unary-call ^Person [^ManagedChannel ch service-name method-name ^Person request] + (ClientCalls/blockingUnaryCall ch + (method-descriptor service-name method-name + MethodDescriptor$MethodType/UNARY) + (CallOptions/DEFAULT) + request)) + +(defmacro with-server [options & body] + `(let [~'server (m/start! ~options)] + (try + ~@body + (finally (m/stop! ~'server))))) + +(defmacro with-channel [port & body] + `(let [~'ch (-> (ManagedChannelBuilder/forAddress "localhost" ~port) + .usePlaintext + .build)] + (try + ~@body + (finally (.shutdownNow ~'ch))))) + +(defn- person [name id] (-> (Person/newBuilder) (.setName name) (.setId id) .build)) + +;;; ---- Tests ----------------------------------------------------------------- + +(deftest test-unary-echo + (with-server + {:grpc-services + [{:proto svc-fd + :name "PersonService" + :methods {"Echo" {:type :unary + :handler (fn [^Person req observer] + (grpc/complete! observer req))}}}]} + (with-channel (.port server) + (let [result (unary-call ch "PersonService" "Echo" (person "Alice" 1))] + (is (= "Alice" (.getName result))) + (is (= 1 (.getId result))))))) + +(deftest test-unary-transform + (with-server + {:grpc-services + [{:proto svc-fd + :name "PersonService" + :methods {"Echo" {:type :unary + :handler (fn [^Person req observer] + (grpc/complete! + observer + (-> (Person/newBuilder) + (.setName (str "hello:" (.getName req))) + (.setId (.getId req)) + .build)))}}}]} + (with-channel (.port server) + (let [result (unary-call ch "PersonService" "Echo" (person "Alice" 42))] + (is (= "hello:Alice" (.getName result))) + (is (= 42 (.getId result))))))) + +(deftest test-server-streaming + (with-server + {:grpc-services + [{:proto svc-fd + :name "PersonService" + :methods {"StreamEcho" {:type :server-stream + :handler (fn [^Person req observer] + (dotimes [i 3] + (grpc/send! observer (person (str (.getName req) "-" i) i))) + (grpc/complete! observer))}}}]} + (with-channel (.port server) + (let [md (method-descriptor "PersonService" "StreamEcho" + MethodDescriptor$MethodType/SERVER_STREAMING) + results (iterator-seq + (ClientCalls/blockingServerStreamingCall ch md (CallOptions/DEFAULT) + (person "Bob" 0))) + names (mapv #(.getName ^Person %) results)] + (is (= ["Bob-0" "Bob-1" "Bob-2"] names)))))) + +(deftest test-bidi-streaming + (with-server + {:grpc-services + [{:proto svc-fd + :name "PersonService" + :methods {"BidiEcho" {:type :bidi + :handler (fn [response-observer] + (grpc/stream-observer + {:on-next #(grpc/send! response-observer %) + :on-error #(grpc/error! response-observer %) + :on-completed #(grpc/complete! response-observer)}))}}}]} + (with-channel (.port server) + (let [received (atom []) + done (promise) + md (method-descriptor "PersonService" "BidiEcho" + MethodDescriptor$MethodType/BIDI_STREAMING) + response-obs (reify StreamObserver + (onNext [_ msg] + (swap! received conj (.getName ^Person msg))) + (onError [_ _] + (deliver done :error)) + (onCompleted [_] + (deliver done :ok))) + request-obs (ClientCalls/asyncBidiStreamingCall + (.newCall ch md (CallOptions/DEFAULT)) + response-obs)] + (doseq [n ["Alice" "Bob" "Carol"]] + (.onNext request-obs (person n 0))) + (.onCompleted request-obs) + (is (= :ok (deref done 5000 :timeout))) + (is (= ["Alice" "Bob" "Carol"] @received)))))) + +(deftest test-unary-async-handler + (with-server + {:grpc-services + [{:proto svc-fd + :name "PersonService" + :methods {"Echo" {:type :unary + :handler (grpc/unary-async-handler + (fn [^Person req out-ch] + (async/>!! out-ch req) + (async/close! out-ch)))}}}]} + (with-channel (.port server) + (let [result (unary-call ch "PersonService" "Echo" (person "Alice" 1))] + (is (= "Alice" (.getName result))) + (is (= 1 (.getId result))))))) + +(deftest test-server-stream-async-handler + (with-server + {:grpc-services + [{:proto svc-fd + :name "PersonService" + :methods {"StreamEcho" {:type :server-stream + :handler (grpc/server-stream-async-handler + (fn [^Person req out-ch] + (dotimes [i 3] + (async/>!! out-ch (person (str (.getName req) "-" i) i))) + (async/close! out-ch)))}}}]} + (with-channel (.port server) + (let [md (method-descriptor "PersonService" "StreamEcho" + MethodDescriptor$MethodType/SERVER_STREAMING) + results (iterator-seq + (ClientCalls/blockingServerStreamingCall ch md (CallOptions/DEFAULT) + (person "Bob" 0))) + names (mapv #(.getName ^Person %) results)] + (is (= ["Bob-0" "Bob-1" "Bob-2"] names)))))) + +(deftest test-bidi-async-handler + (with-server + {:grpc-services + [{:proto svc-fd + :name "PersonService" + :methods {"BidiEcho" {:type :bidi + :handler (grpc/bidi-async-handler + (fn [in-ch out-ch] + (async/go-loop [] + (if-some [msg (async/! out-ch msg) + (recur)) + (async/close! out-ch)))))}}}]} + (with-channel (.port server) + (let [received (atom []) + done (promise) + md (method-descriptor "PersonService" "BidiEcho" + MethodDescriptor$MethodType/BIDI_STREAMING) + response-obs (reify StreamObserver + (onNext [_ msg] + (swap! received conj (.getName ^Person msg))) + (onError [_ _] + (deliver done :error)) + (onCompleted [_] + (deliver done :ok))) + request-obs (ClientCalls/asyncBidiStreamingCall + (.newCall ch md (CallOptions/DEFAULT)) + response-obs)] + (doseq [n ["Alice" "Bob" "Carol"]] + (.onNext request-obs (person n 0))) + (.onCompleted request-obs) + (is (= :ok (deref done 5000 :timeout))) + (is (= ["Alice" "Bob" "Carol"] @received)))))) + +(deftest test-service-as-fn + (with-server + {:grpc-services + [(fn [] + {:proto svc-fd + :name "PersonService" + :methods {"Echo" {:type :unary + :handler (fn [req observer] + (grpc/complete! observer req))}}})]} + (with-channel (.port server) + (is (= "Dave" (.getName (unary-call ch "PersonService" "Echo" (person "Dave" 0))))))))