Sample http4s REST client/server with client macro derivation

build.sbt

@@ -189,7 +189,6 @@ lazy val `dropwizard-client` = project
 lazy val `http` = project
   .in(file("modules/http"))
   .dependsOn(common % "compile->compile;test->test")
-  .dependsOn(channel % "compile->compile;test->test")
   .dependsOn(server % "compile->compile;test->test")
   .settings(moduleName := "mu-rpc-http")
   .settings(httpSettings)

modules/http/src/main/scala/higherkindness/mu/http/Utils.scala → modules/http/src/main/scala/higherkindness/mu/http/implicits.scala

@@ -19,7 +19,6 @@ package higherkindness.mu.http
 import cats.ApplicativeError
 import cats.effect._
 import cats.implicits._
-import fs2.interop.reactivestreams._
 import fs2.{RaiseThrowable, Stream}
 import io.grpc.Status.Code._
 import org.typelevel.jawn.ParseException
@@ -29,24 +28,22 @@ import io.circe.jawn.CirceSupportParser.facade
 import io.circe.syntax._
 import io.grpc.{Status => _, _}
 import jawnfs2._
-import monix.execution._
-import monix.reactive.Observable
 import org.http4s._
 import org.http4s.dsl.Http4sDsl
-import scala.concurrent.ExecutionContext
+import org.http4s.Status.Ok
 import scala.util.control.NoStackTrace
 
-object Utils {
+object implicits {
 
-  implicit class MessageOps[F[_]](val message: Message[F]) extends AnyVal {
+  implicit class MessageOps[F[_]](private val message: Message[F]) extends AnyVal {
 
     def jsonBodyAsStream[A](
         implicit decoder: Decoder[A],
         F: ApplicativeError[F, Throwable]): Stream[F, A] =
       message.body.chunks.parseJsonStream.map(_.as[A]).rethrow
   }
 
-  implicit class RequestOps[F[_]](val request: Request[F]) {
+  implicit class RequestOps[F[_]](private val request: Request[F]) {
 
     def asStream[A](implicit decoder: Decoder[A], F: ApplicativeError[F, Throwable]): Stream[F, A] =
       request
@@ -57,7 +54,7 @@ object Utils {
         }
   }
 
-  implicit class ResponseOps[F[_]](val response: Response[F]) {
+  implicit class ResponseOps[F[_]](private val response: Response[F]) {
 
     implicit private val throwableDecoder: Decoder[Throwable] =
       Decoder.decodeTuple2[String, String].map {
@@ -73,29 +70,21 @@ object Utils {
         implicit decoder: Decoder[A],
         F: ApplicativeError[F, Throwable],
         R: RaiseThrowable[F]): Stream[F, A] =
-      if (response.status.code != 200) Stream.raiseError(ResponseError(response.status))
+      if (response.status.code != Ok.code) Stream.raiseError(ResponseError(response.status))
       else response.jsonBodyAsStream[Either[Throwable, A]].rethrow
   }
 
-  implicit class Fs2StreamOps[F[_], A](stream: Stream[F, A]) {
+  implicit class Fs2StreamOps[F[_], A](private val stream: Stream[F, A]) {
 
-    implicit private val throwableEncoder: Encoder[Throwable] = new Encoder[Throwable] {
+    implicit val throwableEncoder: Encoder[Throwable] = new Encoder[Throwable] {
       def apply(ex: Throwable): Json = (ex.getClass.getName, ex.getMessage).asJson
     }
 
     def asJsonEither(implicit encoder: Encoder[A]): Stream[F, Json] = stream.attempt.map(_.asJson)
-
-    def toObservable(implicit F: ConcurrentEffect[F], ec: ExecutionContext): Observable[A] =
-      Observable.fromReactivePublisher(stream.toUnicastPublisher)
-  }
-
-  implicit class MonixStreamOps[A](val stream: Observable[A]) extends AnyVal {
-
-    def toFs2Stream[F[_]](implicit F: ConcurrentEffect[F], sc: Scheduler): Stream[F, A] =
-      stream.toReactivePublisher.toStream[F]()
   }
 
-  implicit class FResponseOps[F[_]: Sync](response: F[Response[F]]) extends Http4sDsl[F] {
+  implicit class FResponseOps[F[_]: Sync](private val response: F[Response[F]])
+      extends Http4sDsl[F] {
 
     def adaptErrors: F[Response[F]] = response.handleErrorWith {
       case se: StatusException         => errorFromStatus(se.getStatus, se.getMessage)
@@ -106,7 +95,7 @@ object Utils {
     private def errorFromStatus(status: io.grpc.Status, message: String): F[Response[F]] =
       status.getCode match {
         case INVALID_ARGUMENT  => BadRequest(message)
-        case UNAUTHENTICATED   => BadRequest(message)
+        case UNAUTHENTICATED   => Forbidden(message)
         case PERMISSION_DENIED => Forbidden(message)
         case NOT_FOUND         => NotFound(message)
         case UNAVAILABLE       => ServiceUnavailable(message)

modules/http/src/test/scala/higherkindness/mu/rpc/http/GreeterDerivedRestTests.scala

@@ -18,21 +18,21 @@ package higherkindness.mu.rpc.http
 
 import cats.effect.{IO, _}
 import fs2.Stream
+import fs2.interop.reactivestreams._
 import higherkindness.mu.http.{HttpServer, ResponseError, RouteMap}
 import higherkindness.mu.rpc.common.RpcBaseTestSuite
-import higherkindness.mu.http.Utils._
 import monix.reactive.Observable
 import org.http4s._
 import org.http4s.client.blaze.BlazeClientBuilder
 import org.http4s.server.blaze._
 import org.scalatest._
-import org.scalatest.prop.GeneratorDrivenPropertyChecks
+import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks
 
 import scala.concurrent.duration._
 
 class GreeterDerivedRestTests
     extends RpcBaseTestSuite
-    with GeneratorDrivenPropertyChecks
+    with ScalaCheckDrivenPropertyChecks
     with BeforeAndAfter {
 
   val host            = "localhost"
@@ -142,7 +142,7 @@ class GreeterDerivedRestTests
     "serve a POST request with Observable streaming response" in {
       val request = HelloRequest("hey")
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixClient.sayHelloAll(request)(_).toFs2Stream[IO])
+        .flatMap(monixClient.sayHelloAll(request)(_).toReactivePublisher.toStream[IO])
       responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) shouldBe List(HelloResponse("hey"), HelloResponse("hey"))
@@ -159,7 +159,7 @@ class GreeterDerivedRestTests
     "handle errors with Observable streaming response" in {
       val request = HelloRequest("")
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixClient.sayHelloAll(request)(_).toFs2Stream[IO])
+        .flatMap(monixClient.sayHelloAll(request)(_).toReactivePublisher.toStream[IO])
       the[IllegalArgumentException] thrownBy responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) should have message "empty greeting"
@@ -183,7 +183,7 @@ class GreeterDerivedRestTests
     "serve a POST request with bidirectional Observable streaming" in {
       val requests = Observable(HelloRequest("hey"), HelloRequest("there"))
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixClient.sayHellosAll(requests)(_).toFs2Stream[IO])
+        .flatMap(monixClient.sayHellosAll(requests)(_).toReactivePublisher.toStream[IO])
       responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) shouldBe List(HelloResponse("hey"), HelloResponse("there"))
@@ -192,7 +192,7 @@ class GreeterDerivedRestTests
     "serve an empty POST request with bidirectional Observable streaming" in {
       val requests = Observable.empty
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixClient.sayHellosAll(requests)(_).toFs2Stream[IO])
+        .flatMap(monixClient.sayHellosAll(requests)(_).toReactivePublisher.toStream[IO])
       responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) shouldBe Nil
@@ -202,7 +202,7 @@ class GreeterDerivedRestTests
       forAll { strings: List[String] =>
         val requests = Observable.fromIterable(strings.map(HelloRequest))
         val responses = BlazeClientBuilder[IO](ec).stream
-          .flatMap(monixClient.sayHellosAll(requests)(_).toFs2Stream[IO])
+          .flatMap(monixClient.sayHellosAll(requests)(_).toReactivePublisher.toStream[IO])
         responses.compile.toList
           .unsafeRunTimed(10.seconds)
           .getOrElse(sys.error("Stuck!")) shouldBe strings.map(HelloResponse)

modules/http/src/test/scala/higherkindness/mu/rpc/http/GreeterRestClients.scala

@@ -18,9 +18,10 @@ package higherkindness.mu.rpc.http
 
 import cats.effect._
 import fs2.Stream
+import fs2.interop.reactivestreams._
 import io.circe.generic.auto._
 import io.circe.syntax._
-import higherkindness.mu.http.Utils._
+import higherkindness.mu.http.implicits._
 import org.http4s._
 import org.http4s.circe._
 import org.http4s.client._
@@ -67,28 +68,33 @@ class MonixGreeterRestClient[F[_]: ConcurrentEffect](uri: Uri)(
     implicit sc: monix.execution.Scheduler) {
 
   import monix.reactive.Observable
-  import higherkindness.mu.http.Utils._
+  import higherkindness.mu.http.implicits._
 
   private implicit val responseDecoder: EntityDecoder[F, HelloResponse] = jsonOf[F, HelloResponse]
 
   def sayHellos(arg: Observable[HelloRequest])(implicit client: Client[F]): F[HelloResponse] = {
     val request = Request[F](Method.POST, uri / "sayHellos")
-    client.expectOr[HelloResponse](request.withEntity(arg.toFs2Stream.map(_.asJson)))(
-      handleResponseError)
+    client.expectOr[HelloResponse](
+      request.withEntity(arg.toReactivePublisher.toStream.map(_.asJson)))(handleResponseError)
   }
 
   def sayHelloAll(arg: HelloRequest)(implicit client: Client[F]): Observable[HelloResponse] = {
     val request = Request[F](Method.POST, uri / "sayHelloAll")
-    client.stream(request.withEntity(arg.asJson)).flatMap(_.asStream[HelloResponse]).toObservable
+    Observable.fromReactivePublisher(
+      client
+        .stream(request.withEntity(arg.asJson))
+        .flatMap(_.asStream[HelloResponse])
+        .toUnicastPublisher)
   }
 
   def sayHellosAll(arg: Observable[HelloRequest])(
       implicit client: Client[F]): Observable[HelloResponse] = {
     val request = Request[F](Method.POST, uri / "sayHellosAll")
-    client
-      .stream(request.withEntity(arg.toFs2Stream.map(_.asJson)))
-      .flatMap(_.asStream[HelloResponse])
-      .toObservable
+    Observable.fromReactivePublisher(
+      client
+        .stream(request.withEntity(arg.toReactivePublisher.toStream.map(_.asJson)))
+        .flatMap(_.asStream[HelloResponse])
+        .toUnicastPublisher)
   }
 
 }

modules/http/src/test/scala/higherkindness/mu/rpc/http/GreeterRestServices.scala

@@ -21,7 +21,9 @@ import cats.syntax.flatMap._
 import cats.syntax.functor._
 import io.circe.generic.auto._
 import io.circe.syntax._
-import higherkindness.mu.http.Utils._
+import higherkindness.mu.http.implicits._
+import fs2.interop.reactivestreams._
+import monix.reactive.Observable
 import org.http4s._
 import org.http4s.circe._
 import org.http4s.dsl.Http4sDsl
@@ -77,16 +79,24 @@ class MonixGreeterRestService[F[_]: ConcurrentEffect](
 
     case msg @ POST -> Root / "sayHellos" =>
       val requests = msg.asStream[HelloRequest]
-      Ok(handler.sayHellos(requests.toObservable).map(_.asJson))
+      Ok(
+        handler
+          .sayHellos(Observable.fromReactivePublisher(requests.toUnicastPublisher))
+          .map(_.asJson))
 
     case msg @ POST -> Root / "sayHelloAll" =>
       for {
         request   <- msg.as[HelloRequest]
-        responses <- Ok(handler.sayHelloAll(request).toFs2Stream.asJsonEither)
+        responses <- Ok(handler.sayHelloAll(request).toReactivePublisher.toStream.asJsonEither)
       } yield responses
 
     case msg @ POST -> Root / "sayHellosAll" =>
       val requests = msg.asStream[HelloRequest]
-      Ok(handler.sayHellosAll(requests.toObservable).toFs2Stream.asJsonEither)
+      Ok(
+        handler
+          .sayHellosAll(Observable.fromReactivePublisher(requests.toUnicastPublisher))
+          .toReactivePublisher
+          .toStream
+          .asJsonEither)
   }
 }

modules/http/src/test/scala/higherkindness/mu/rpc/http/GreeterRestTests.scala

@@ -18,9 +18,10 @@ package higherkindness.mu.rpc.http
 
 import cats.effect.{IO, _}
 import fs2.Stream
+import fs2.interop.reactivestreams._
 import higherkindness.mu.http.ResponseError
 import higherkindness.mu.rpc.common.RpcBaseTestSuite
-import higherkindness.mu.http.Utils._
+import higherkindness.mu.http.implicits._
 import io.circe.Json
 import io.circe.generic.auto._
 import io.circe.syntax._
@@ -31,15 +32,15 @@ import org.http4s.client.UnexpectedStatus
 import org.http4s.client.blaze.BlazeClientBuilder
 import org.http4s.server.blaze._
 import org.scalatest._
-import org.scalatest.prop.GeneratorDrivenPropertyChecks
 import org.http4s.implicits._
 import org.http4s.server.Router
+import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks
 
 import scala.concurrent.duration._
 
 class GreeterRestTests
     extends RpcBaseTestSuite
-    with GeneratorDrivenPropertyChecks
+    with ScalaCheckDrivenPropertyChecks
     with BeforeAndAfter {
 
   val Hostname = "localhost"
@@ -59,7 +60,6 @@ class GreeterRestTests
   implicit val fs2HandlerIO   = new Fs2GreeterHandler[IO]
   implicit val monixHandlerIO = new MonixGreeterHandler[IO]
 
-  //TODO: add Logger middleware
   val unaryService: HttpRoutes[IO] = new UnaryGreeterRestService[IO].service
   val fs2Service: HttpRoutes[IO]   = new Fs2GreeterRestService[IO].service
   val monixService: HttpRoutes[IO] = new MonixGreeterRestService[IO].service
@@ -198,7 +198,7 @@ class GreeterRestTests
     "serve a POST request with Observable streaming response" in {
       val request = HelloRequest("hey")
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixServiceClient.sayHelloAll(request)(_).toFs2Stream[IO])
+        .flatMap(monixServiceClient.sayHelloAll(request)(_).toReactivePublisher.toStream[IO])
       responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) shouldBe List(HelloResponse("hey"), HelloResponse("hey"))
@@ -215,7 +215,7 @@ class GreeterRestTests
     "handle errors with Observable streaming response" in {
       val request = HelloRequest("")
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixServiceClient.sayHelloAll(request)(_).toFs2Stream[IO])
+        .flatMap(monixServiceClient.sayHelloAll(request)(_).toReactivePublisher.toStream[IO])
       the[IllegalArgumentException] thrownBy responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) should have message "empty greeting"
@@ -239,7 +239,7 @@ class GreeterRestTests
     "serve a POST request with bidirectional Observable streaming" in {
       val requests = Observable(HelloRequest("hey"), HelloRequest("there"))
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixServiceClient.sayHellosAll(requests)(_).toFs2Stream[IO])
+        .flatMap(monixServiceClient.sayHellosAll(requests)(_).toReactivePublisher.toStream[IO])
       responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) shouldBe List(HelloResponse("hey"), HelloResponse("there"))
@@ -248,7 +248,7 @@ class GreeterRestTests
     "serve an empty POST request with bidirectional Observable streaming" in {
       val requests = Observable.empty
       val responses = BlazeClientBuilder[IO](ec).stream
-        .flatMap(monixServiceClient.sayHellosAll(requests)(_).toFs2Stream[IO])
+        .flatMap(monixServiceClient.sayHellosAll(requests)(_).toReactivePublisher.toStream[IO])
       responses.compile.toList
         .unsafeRunTimed(10.seconds)
         .getOrElse(sys.error("Stuck!")) shouldBe Nil
@@ -258,7 +258,7 @@ class GreeterRestTests
       forAll { strings: List[String] =>
         val requests = Observable.fromIterable(strings.map(HelloRequest))
         val responses = BlazeClientBuilder[IO](ec).stream
-          .flatMap(monixServiceClient.sayHellosAll(requests)(_).toFs2Stream[IO])
+          .flatMap(monixServiceClient.sayHellosAll(requests)(_).toReactivePublisher.toStream[IO])
         responses.compile.toList
           .unsafeRunTimed(10.seconds)
           .getOrElse(sys.error("Stuck!")) shouldBe strings.map(HelloResponse)

modules/internal/src/main/scala/higherkindness/mu/rpc/internal/serviceImpl.scala

@@ -441,7 +441,7 @@ object serviceImpl {
 
         val executionClient: Tree = response match {
           case MonixObservableTpe(_, _) =>
-            q"client.stream(request).flatMap(_.asStream[${response.safeInner}]).toObservable"
+            q"Observable.fromReactivePublisher(client.stream(request).flatMap(_.asStream[${response.safeInner}]).toUnicastPublisher)"
           case Fs2StreamTpe(_, _) =>
             q"client.stream(request).flatMap(_.asStream[${response.safeInner}])"
           case _ =>
@@ -454,7 +454,7 @@ object serviceImpl {
           case _: Fs2StreamTpe =>
             q"val request = Request[F](Method.$method, uri / ${uri.replace("\"", "")}).withEntity(req.map(_.asJson))"
           case _: MonixObservableTpe =>
-            q"val request = Request[F](Method.$method, uri / ${uri.replace("\"", "")}).withEntity(req.toFs2Stream.map(_.asJson))"
+            q"val request = Request[F](Method.$method, uri / ${uri.replace("\"", "")}).withEntity(req.toReactivePublisher.toStream.map(_.asJson))"
           case _ =>
             q"val request = Request[F](Method.$method, uri / ${uri.replace("\"", "")})"
         }
@@ -494,17 +494,17 @@ object serviceImpl {
 
           case (_: MonixObservableTpe, _: UnaryTpe) =>
             q"""val requests = msg.asStream[${operation.request.safeInner}]
-              Ok(handler.${operation.name}(requests.toObservable).map(_.asJson))"""
+              Ok(handler.${operation.name}(Observable.fromReactivePublisher(requests.toUnicastPublisher)).map(_.asJson))"""
 
           case (_: UnaryTpe, _: MonixObservableTpe) =>
             q"""for {
                 request   <- msg.as[${operation.request.safeInner}]
-                responses <- Ok(handler.${operation.name}(request).toFs2Stream.asJsonEither)
+                responses <- Ok(handler.${operation.name}(request).toReactivePublisher.toStream.asJsonEither)
               } yield responses"""
 
           case (_: MonixObservableTpe, _: MonixObservableTpe) =>
             q"""val requests = msg.asStream[${operation.request.safeInner}]
-              Ok(handler.${operation.name}(requests.toObservable).toFs2Stream.asJsonEither)"""
+              Ok(handler.${operation.name}(Observable.fromReactivePublisher(requests.toUnicastPublisher)).toReactivePublisher.toStream.asJsonEither)"""
 
           case (_: EmptyTpe, _) =>
             q"""Ok(handler.${operation.name}(_root_.higherkindness.mu.rpc.protocol.Empty).map(_.asJson))"""
@@ -549,7 +549,8 @@ object serviceImpl {
       }"""
 
       val httpImports: List[Tree] = List(
-        q"import _root_.higherkindness.mu.http.Utils._",
+        q"import _root_.higherkindness.mu.http.implicits._",
+        q"import _root_.fs2.interop.reactivestreams._",
         q"import _root_.cats.syntax.flatMap._",
         q"import _root_.cats.syntax.functor._",
         q"import _root_.org.http4s._",