Merge pull request #583 from danicheg/propagate-features-from-main

Propagate the recent features from the `0.5` series

Author: danicheg

CONTRIBUTING.md

@@ -10,3 +10,4 @@ request signifies your consent to license your contributions under the
 Apache License 2.0.
 
 [contributors' guide]: https://http4s.org/contributing/
+[Apache License 2.0]: ./LICENSE

README.md

@@ -18,6 +18,12 @@
     - Distributed tracing
     - [and](https://armeria.dev/docs/server-docservice) [so](https://armeria.dev/docs/server-thrift) [on](https://armeria.dev/docs/advanced-metrics)
 
+## Current status
+
+Two series are currently under active development: the `0.x` and `1.0-x` release milestone series.
+The first depends on the `http4s-core`'s `0.23` series and belongs to the [main branch].
+The latter is for the cutting-edge `http4s-core`'s `1.0-x` release milestone series and belongs to the [series/1.x branch].
+
 ## Installation
 
 Add the following dependencies to `build.sbt`
@@ -162,3 +168,5 @@ Visit [examples](./examples) to find a fully working example.
 
 [http4s]: https://http4s.org/
 [armeria]: https://armeria.dev/
+[main branch]: https://github.com/http4s/http4s-armeria/tree/main
+[series/1.x branch]: https://github.com/http4s/http4s-armeria/tree/series/1.x

examples/armeria-fs2grpc/src/main/scala/com/example/fs2grpc/armeria/Main.scala

@@ -61,6 +61,7 @@ object Main extends IOApp {
       .bindHttp(httpPort)
       .withIdleTimeout(Duration.Zero)
       .withRequestTimeout(Duration.Zero)
+      .withMaxRequestLength(0L)
       .withHttpServiceUnder("/grpc", grpcService)
       .withHttpRoutes("/rest", ExampleService[IO].routes())
       .withDecorator(LoggingService.newDecorator())

server/src/main/scala/org/http4s/armeria/server/ArmeriaHttp4sHandler.scala

@@ -42,14 +42,8 @@ import fs2._
 import fs2.interop.reactivestreams._
 import ArmeriaHttp4sHandler.{RightUnit, canHasBody, defaultVault, toHttp4sMethod}
 import com.comcast.ip4s.SocketAddress
-import org.http4s.server.{
-  DefaultServiceErrorHandler,
-  SecureSession,
-  ServerRequestKeys,
-  ServiceErrorHandler
-}
+import org.http4s.server.{SecureSession, ServerRequestKeys, ServiceErrorHandler}
 import org.typelevel.ci.CIString
-import org.typelevel.log4cats.LoggerFactory
 import scodec.bits.ByteVector
 
 import scala.jdk.CollectionConverters._
@@ -257,11 +251,12 @@ private[armeria] class ArmeriaHttp4sHandler[F[_]](
 }
 
 private[armeria] object ArmeriaHttp4sHandler {
-  def apply[F[_]: Async: LoggerFactory](
+  def apply[F[_]: Async](
       prefix: String,
       service: HttpApp[F],
+      serviceErrorHandler: ServiceErrorHandler[F],
       dispatcher: Dispatcher[F]): ArmeriaHttp4sHandler[F] =
-    new ArmeriaHttp4sHandler(prefix, service, DefaultServiceErrorHandler, dispatcher)
+    new ArmeriaHttp4sHandler(prefix, service, serviceErrorHandler, dispatcher)
 
   private val serverSoftware: ServerSoftware =
     ServerSoftware("armeria", Some(Version.get("armeria").artifactVersion()))

server/src/main/scala/org/http4s/armeria/server/ArmeriaServerBuilder.scala

@@ -16,12 +16,25 @@
 
 package org.http4s.armeria.server
 
+import java.io.{File, InputStream}
+import java.net.InetSocketAddress
+import java.security.PrivateKey
+import java.security.cert.X509Certificate
+import java.util.function.{Function => JFunction}
+import javax.net.ssl.KeyManagerFactory
+
+import cats.Monad
 import cats.effect.{Async, Resource}
-import cats.syntax.applicative._
-import cats.syntax.flatMap._
-import cats.syntax.functor._
+import cats.effect.std.Dispatcher
+import cats.syntax.all._
 import com.linecorp.armeria.common.util.Version
-import com.linecorp.armeria.common.{HttpRequest, HttpResponse, SessionProtocol, TlsKeyPair}
+import com.linecorp.armeria.common.{
+  ContentTooLargeException,
+  HttpRequest,
+  HttpResponse,
+  SessionProtocol,
+  TlsKeyPair
+}
 import com.linecorp.armeria.server.{
   HttpService,
   HttpServiceWithRoutes,
@@ -33,18 +46,10 @@ import com.linecorp.armeria.server.{
 import io.micrometer.core.instrument.MeterRegistry
 import io.netty.channel.ChannelOption
 import io.netty.handler.ssl.SslContextBuilder
-
-import java.io.{File, InputStream}
-import java.net.InetSocketAddress
-import java.security.PrivateKey
-import java.security.cert.X509Certificate
-import java.util.function.{Function => JFunction}
-import cats.effect.std.Dispatcher
 import com.comcast.ip4s
-
-import javax.net.ssl.KeyManagerFactory
 import org.http4s.armeria.server.ArmeriaServerBuilder.AddServices
-import org.http4s.{BuildInfo, HttpApp, HttpRoutes}
+import org.http4s.headers.{Connection, `Content-Length`}
+import org.http4s.{BuildInfo, Headers, HttpApp, HttpRoutes, Request, Response, Status}
 import org.http4s.server.{
   DefaultServiceErrorHandler,
   Server,
@@ -169,7 +174,9 @@ sealed class ArmeriaServerBuilder[F[_]] private (
   def withHttpApp(prefix: String, service: HttpApp[F]): Self =
     copy(addServices = (ab, dispatcher) =>
       addServices(ab, dispatcher).map(
-        _.serviceUnder(prefix, ArmeriaHttp4sHandler(prefix, service, dispatcher))))
+        _.serviceUnder(
+          prefix,
+          ArmeriaHttp4sHandler(prefix, service, serviceErrorHandler, dispatcher))))
 
   /** Decorates all HTTP services with the specified [[DecoratingFunction]]. */
   def withDecorator(decorator: DecoratingFunction): Self =
@@ -205,6 +212,14 @@ sealed class ArmeriaServerBuilder[F[_]] private (
   def withIdleTimeout(idleTimeout: FiniteDuration): Self =
     atBuild(_.idleTimeoutMillis(idleTimeout.toMillis))
 
+  /** Sets the maximum allowed length of the content decoded at the session layer.
+    *
+    * @param limit
+    *   the maximum allowed length. {@code 0} disables the length limit.
+    */
+  def withMaxRequestLength(limit: Long): Self =
+    atBuild(_.maxRequestLength(limit))
+
   /** Sets the timeout of a request.
     *
     * @param requestTimeout
@@ -359,7 +374,29 @@ object ArmeriaServerBuilder {
     new ArmeriaServerBuilder(
       (armeriaBuilder, _) => armeriaBuilder.pure,
       socketAddress = defaults.IPv4SocketAddress.toInetSocketAddress,
-      serviceErrorHandler = DefaultServiceErrorHandler,
+      serviceErrorHandler = defaultServiceErrorHandler[F],
       banner = defaults.Banner
     )
+
+  /** Incorporates the default service error handling from Http4s'
+    * [[org.http4s.server.DefaultServiceErrorHandler DefaultServiceErrorHandler]] and adds handling
+    * for some errors propagated from the Armeria side.
+    */
+  def defaultServiceErrorHandler[F[_]](implicit
+      F: Monad[F],
+      LF: LoggerFactory[F]): Request[F] => PartialFunction[Throwable, F[Response[F]]] = {
+    val contentLengthErrorHandler: Request[F] => PartialFunction[Throwable, F[Response[F]]] =
+      req => { case _: ContentTooLargeException =>
+        Response[F](
+          Status.PayloadTooLarge,
+          req.httpVersion,
+          Headers(
+            Connection.close,
+            `Content-Length`.zero
+          )
+        ).pure[F]
+      }
+
+    req => contentLengthErrorHandler(req).orElse(DefaultServiceErrorHandler(LF, F)(req))
+  }
 }

server/src/test/scala/org/http4s/armeria/server/ArmeriaServerBuilderSuite.scala

@@ -24,7 +24,7 @@ import cats.implicits._
 import com.linecorp.armeria.client.logging.LoggingClient
 import com.linecorp.armeria.client.{ClientFactory, WebClient}
 import com.linecorp.armeria.common.{HttpData, HttpStatus}
-import com.linecorp.armeria.server.logging.{ContentPreviewingService, LoggingService}
+import com.linecorp.armeria.server.logging.LoggingService
 import fs2._
 import munit.CatsEffectSuite
 import org.http4s.dsl.io._
@@ -50,7 +50,9 @@ class ArmeriaServerBuilderSuite extends CatsEffectSuite with ServerFixture {
       IO(Thread.currentThread.getName).flatMap(Ok(_))
 
     case req @ POST -> Root / "echo" =>
-      Ok(req.body)
+      req.decode[IO, String] { r =>
+        Ok(r)
+      }
 
     case GET -> Root / "trailers" =>
       Ok("Hello").map(response =>
@@ -72,11 +74,11 @@ class ArmeriaServerBuilderSuite extends CatsEffectSuite with ServerFixture {
 
   protected def configureServer(serverBuilder: ArmeriaServerBuilder[IO]): ArmeriaServerBuilder[IO] =
     serverBuilder
-      .withDecorator(ContentPreviewingService.newDecorator(Int.MaxValue))
       .withDecorator(LoggingService.newDecorator())
       .bindAny()
       .withRequestTimeout(10.seconds)
       .withGracefulShutdownTimeout(0.seconds, 0.seconds)
+      .withMaxRequestLength(1024 * 1024)
       .withHttpRoutes("/service", service)
 
   lazy val client: WebClient = WebClient
@@ -151,6 +153,16 @@ class ArmeriaServerBuilderSuite extends CatsEffectSuite with ServerFixture {
     assertEquals(postChunkedMultipart("/service/issue2610", "aa", body), "a")
   }
 
+  test("reliably handle entity length limiting") {
+    val input = List.fill(1024 * 1024 + 1)("F").mkString
+
+    val statusIO = IO(
+      postLargeBody("/service/echo", input)
+    )
+
+    assertIO(statusIO, HttpStatus.REQUEST_ENTITY_TOO_LARGE.code())
+  }
+
   test("stream") {
     val response = client.get("/service/stream")
     val deferred = Deferred.unsafe[IO, Boolean]
@@ -176,6 +188,19 @@ class ArmeriaServerBuilderSuite extends CatsEffectSuite with ServerFixture {
     } yield ()
   }
 
+  private def postLargeBody(path: String, body: String): Int = {
+    val url = new URL(s"http://127.0.0.1:${httpPort.get}$path")
+    val conn = url.openConnection().asInstanceOf[HttpURLConnection]
+    val bytes = body.getBytes(StandardCharsets.UTF_8)
+    conn.setRequestMethod("POST")
+    conn.setRequestProperty("Content-Type", "text/html; charset=utf-8")
+    conn.setDoOutput(true)
+    conn.getOutputStream.write(bytes)
+    val code = conn.getResponseCode
+    conn.disconnect()
+    code
+  }
+
   private def postChunkedMultipart(path: String, boundary: String, body: String): String = {
     val url = new URL(s"http://127.0.0.1:${httpPort.get}$path")
     val conn = url.openConnection().asInstanceOf[HttpURLConnection]