Sample http4s REST client/server with client macro derivation

build.sbt

@@ -160,6 +160,13 @@ lazy val `prometheus` = project
 //// DROPWIZARD ////
 ////////////////////
 
+lazy val `dropwizard` = project
+  .in(file("modules/metrics/dropwizard"))
+  .dependsOn(`internal-core`)
+  .dependsOn(testing % "test->test")
+  .settings(moduleName := "mu-rpc-dropwizard")
+  .settings(dropwizardMetricsSettings)
+
 lazy val `dropwizard-server` = project
   .in(file("modules/dropwizard/server"))
   .dependsOn(`prometheus-server` % "compile->compile;test->test")
@@ -175,12 +182,17 @@ lazy val `dropwizard-client` = project
   .settings(moduleName := "mu-rpc-dropwizard-client")
   .settings(dropwizardSettings)
 
-lazy val `dropwizard` = project
-  .in(file("modules/metrics/dropwizard"))
-  .dependsOn(`internal-core`)
-  .dependsOn(testing % "test->test")
-  .settings(moduleName := "mu-rpc-dropwizard")
-  .settings(dropwizardMetricsSettings)
+///////////////////
+//// HTTP/REST ////
+///////////////////
+
+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)
 
 ////////////////
 //// IDLGEN ////
@@ -395,6 +407,7 @@ lazy val allModules: Seq[ProjectReference] = Seq(
   testing,
   ssl,
   `idlgen-core`,
+  `http`,
   `marshallers-jodatime`,
   `example-routeguide-protocol`,
   `example-routeguide-common`,

modules/common/src/main/scala/higherkindness/mu/rpc/protocol/protocol.scala

@@ -37,6 +37,7 @@ class message                          extends StaticAnnotation
 class option(name: String, value: Any) extends StaticAnnotation
 class outputPackage(value: String)     extends StaticAnnotation
 class outputName(value: String)        extends StaticAnnotation
+class http                             extends StaticAnnotation
 
 @message
 object Empty

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

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2017-2019 47 Degrees, LLC. <http://www.47deg.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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
+import io.circe._
+import io.circe.generic.auto._
+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 scala.util.control.NoStackTrace
+
+object Utils {
+
+  implicit class MessageOps[F[_]](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]) {
+
+    def asStream[A](implicit decoder: Decoder[A], F: ApplicativeError[F, Throwable]): Stream[F, A] =
+      request
+        .jsonBodyAsStream[A]
+        .adaptError { // mimic behavior of MessageOps.as[T] in handling of parsing errors
+          case ex: ParseException =>
+            MalformedMessageBodyFailure(ex.getMessage, Some(ex)) // will return 400 instead of 500
+        }
+  }
+
+  implicit class ResponseOps[F[_]](val response: Response[F]) {
+
+    implicit private val throwableDecoder: Decoder[Throwable] =
+      Decoder.decodeTuple2[String, String].map {
+        case (cls, msg) =>
+          Class
+            .forName(cls)
+            .getConstructor(classOf[String])
+            .newInstance(msg)
+            .asInstanceOf[Throwable]
+      }
+
+    def asStream[A](
+        implicit decoder: Decoder[A],
+        F: ApplicativeError[F, Throwable],
+        R: RaiseThrowable[F]): Stream[F, A] =
+      if (response.status.code != 200) Stream.raiseError(ResponseError(response.status))
+      else response.jsonBodyAsStream[Either[Throwable, A]].rethrow
+  }
+
+  implicit class Fs2StreamOps[F[_], A](stream: Stream[F, A]) {
+
+    implicit private 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] {
+
+    def adaptErrors: F[Response[F]] = response.handleErrorWith {
+      case se: StatusException         => errorFromStatus(se.getStatus, se.getMessage)
+      case sre: StatusRuntimeException => errorFromStatus(sre.getStatus, sre.getMessage)
+      case other: Throwable            => InternalServerError(other.getMessage)
+    }
+
+    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 PERMISSION_DENIED => Forbidden(message)
+        case NOT_FOUND         => NotFound(message)
+        case UNAVAILABLE       => ServiceUnavailable(message)
+        case _                 => InternalServerError(message)
+      }
+  }
+
+  def handleResponseError[F[_]: Sync](errorResponse: Response[F]): F[Throwable] =
+    errorResponse.bodyAsText.compile.foldMonoid.map(body =>
+      ResponseError(errorResponse.status, Some(body).filter(_.nonEmpty)))
+}
+
+final case class ResponseError(status: Status, msg: Option[String] = None)
+    extends RuntimeException(status + msg.fold("")(": " + _))
+    with NoStackTrace

modules/http/src/main/scala/higherkindness/mu/http/protocol.scala

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2017-2019 47 Degrees, LLC. <http://www.47deg.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package higherkindness.mu.http
+
+import cats.effect.{ConcurrentEffect, Timer}
+import org.http4s.HttpRoutes
+import org.http4s.server.blaze.BlazeServerBuilder
+import org.http4s.implicits._
+import org.http4s.server.Router
+
+case class RouteMap[F[_]](prefix: String, route: HttpRoutes[F])
+
+object HttpServer {
+
+  def bind[F[_]: ConcurrentEffect: Timer](
+      port: Int,
+      host: String,
+      routes: RouteMap[F]*): BlazeServerBuilder[F] =
+    BlazeServerBuilder[F]
+      .bindHttp(port, host)
+      .withHttpApp(Router(routes.map(r => (s"/${r.prefix}", r.route)): _*).orNotFound)
+
+}

modules/http/src/test/resources/logback-test.xml

@@ -0,0 +1,11 @@
+<configuration debug="true">
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="debug">
+        <appender-ref ref="STDOUT" />
+    </root>
+</configuration>