diff --git a/README.md b/README.md index 387109e..7b5bd71 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ Scala: * [ScalaPB](https://scalapb.github.io) services with `Future` monad * [fs2-grpc](https://github.com/typelevel/fs2-grpc), built on top of `cats-effect` and `fs2` -* [ZIO gRPC](https://scalapb.github.io/zio-grpc/), built on top of `ZIO` monad (the most feature-rich implementation) +* [ZIO gRPC](https://scalapb.github.io/zio-grpc/), built on top of `ZIO` *Note*: at the moment, only unary (non-streaming) methods are supported. @@ -71,7 +71,7 @@ supports_message_receive_limit: false ## Usage -Library is installed via SBT (you also need to install particular `http4s` server implementation): +Installing with SBT (you also need to install particular `http4s` server implementation): ```scala libraryDependencies ++= Seq( @@ -109,10 +109,10 @@ val httpServer: Resource[IO, org.http4s.server.Server] = { httpServer.use(_ => IO.never).unsafeRunSync() ``` -### Tip: GRPC OpenTelemetry integration +### Hint: GRPC OpenTelemetry integration Since the library creates a separate "fake" GRPC server, traffic going through it won't be captured by the -instrumentation of the main GRPC server. +instrumentation of your main GRPC server (if any). Here is how you can integrate OpenTelemetry with the Connect-RPC server: @@ -129,8 +129,11 @@ ConnectRouteBuilder.forServices[IO](grpcServices) .build ``` -This will make sure that all the traffic going through the Connect-RPC server will be captured by the same -opentelemetry. +### ZIO Interop + +Because the library uses the Tagless Final pattern, it is perfectly possible to use it with ZIO. You might check +`zio_interop` branch, where conformance is implemented with `ZIO` and `ZIO-gRPC`. +You can read [this](https://zio.dev/guides/interop/with-cats-effect/). ## Development @@ -147,14 +150,14 @@ Diagnostic data from the server itself is written to the log file `out/out.log`. ### Conformance tests status -Current status: 6/79 tests pass +Current status: 11/79 tests pass. Known issues: -* `fs2-grpc` server implementation doesn't support setting response headers (which is required by the tests): [#31](https://github.com/igor-vovk/connect-rpc-scala/issues/31) -* `google.protobuf.Any` serialization doesn't follow Connect-RPC spec: [#32](https://github.com/igor-vovk/connect-rpc-scala/issues/32) +* `google.protobuf.Any` serialization doesn't follow Connect-RPC + spec: [#32](https://github.com/igor-vovk/connect-rpc-scala/issues/32) ## Future improvements -* Support GET-requests -* Support non-unary (streaming) methods +[x] Support GET-requests +[ ] Support non-unary (streaming) methods diff --git a/conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/ConformanceServiceImpl.scala b/conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/ConformanceServiceImpl.scala index 6886b3e..5cce47f 100644 --- a/conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/ConformanceServiceImpl.scala +++ b/conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/ConformanceServiceImpl.scala @@ -75,7 +75,10 @@ class ConformanceServiceImpl[F[_] : Async] extends ConformanceServiceFs2GrpcTrai throw new StatusRuntimeException(status) } - val trailers = mkMetadata(responseDefinition.responseTrailers) + val trailers = mkMetadata(Seq.concat( + responseDefinition.responseHeaders, + responseDefinition.responseTrailers.map(h => h.copy(name = s"trailer-${h.name}")), + )) Async[F].sleep(Duration(responseDefinition.responseDelayMs, TimeUnit.MILLISECONDS)) *> Async[F].pure(UnaryHandlerResponse( diff --git a/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectHandler.scala b/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectHandler.scala index 32d3c89..ea09a6c 100644 --- a/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectHandler.scala +++ b/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectHandler.scala @@ -27,6 +27,7 @@ class ConnectHandler[F[_] : Async]( methodRegistry: MethodRegistry, channel: Channel, httpDsl: Http4sDsl[F], + treatTrailersAsHeaders: Boolean, ) { import httpDsl.* @@ -119,12 +120,11 @@ class ConnectHandler[F[_] : Async]( message ) }).map { response => - val headers = org.http4s.Headers.empty ++ - responseHeaderMetadata.get.toHeaders ++ - responseTrailerMetadata.get.toTrailingHeaders + val headers = responseHeaderMetadata.get.toHeaders() ++ + responseTrailerMetadata.get.toHeaders(trailing = !treatTrailersAsHeaders) if (logger.isTraceEnabled) { - logger.trace(s"<<< Headers: ${headers.redactSensitive}") + logger.trace(s"<<< Headers: ${headers.redactSensitive()}") } Response(Ok, headers = headers).withEntity(response) diff --git a/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectRouteBuilder.scala b/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectRouteBuilder.scala index 24860b3..fb48233 100644 --- a/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectRouteBuilder.scala +++ b/core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectRouteBuilder.scala @@ -35,6 +35,7 @@ case class ConnectRouteBuilder[F[_] : Async] private( channelConfigurator: Endo[ManagedChannelBuilder[_]] = identity, executor: Executor = ExecutionContext.global, waitForShutdown: Duration = 5.seconds, + treatTrailersAsHeaders: Boolean = true, ) { import Mappings.* @@ -54,6 +55,17 @@ case class ConnectRouteBuilder[F[_] : Async] private( def withWaitForShutdown(duration: Duration): ConnectRouteBuilder[F] = copy(waitForShutdown = duration) + /** + * If enabled, trailers will be treated as headers (no "trailer-" prefix). + * + * Both `fs2-grpc` and `zio-grpc` support trailing headers only, so enabling this option is a single way to + * send headers from the server to the client. + * + * Enabled by default. + */ + def withTreatTrailersAsHeaders(enabled: Boolean): ConnectRouteBuilder[F] = + copy(treatTrailersAsHeaders = enabled) + /** * Method can be used if you want to add additional routes to the server. * Otherwise, it is preferred to use the [[build]] method. @@ -86,6 +98,7 @@ case class ConnectRouteBuilder[F[_] : Async] private( methodRegistry, channel, httpDsl, + treatTrailersAsHeaders, ) HttpRoutes.of[F] { diff --git a/core/src/main/scala/org/ivovk/connect_rpc_scala/Mappings.scala b/core/src/main/scala/org/ivovk/connect_rpc_scala/Mappings.scala index a40a075..62170e7 100644 --- a/core/src/main/scala/org/ivovk/connect_rpc_scala/Mappings.scala +++ b/core/src/main/scala/org/ivovk/connect_rpc_scala/Mappings.scala @@ -40,9 +40,12 @@ trait HeaderMappings { new Headers(b.result()) } - def toHeaders: Headers = headers() + def toHeaders(trailing: Boolean = false): Headers = { + val prefix = if trailing then "trailer-" else "" + + headers(prefix) + } - def toTrailingHeaders: Headers = headers("trailer-") } }