Better body handling when logging requests. Handling for unconsumed r…

…equest body. Updated README.
theiterators · Oct 14, 2024 · 2e8dbef · 2e8dbef
1 parent 8403336
commit 2e8dbef
Show file tree

Hide file tree

Showing 3 changed files with 118 additions and 68 deletions.
diff --git a/README.md b/README.md
@@ -15,8 +15,8 @@ http4s-stir also furnishes a test kit akin to Pekko's (Akka's).
 In SBT:
 
 ```scala
-libraryDependencies += "pl.iterators" %% "http4s-stir" % "0.2"
-libraryDependencies += "pl.iterators" %% "http4s-stir-testkit" % "0.2" % Test // if you need this
+libraryDependencies += "pl.iterators" %% "http4s-stir" % "0.4.0"
+libraryDependencies += "pl.iterators" %% "http4s-stir-testkit" % "0.4.0" % Test // if you need this
 ```
 
 For `scala-cli` see [this example](#example).
@@ -25,15 +25,15 @@ For `scala-cli` see [this example](#example).
 
 Here's an example in Scala 3 that you can run using scala-cli:
 
-```scala
+```scala 3
 // Main.scala
-//> using dep org.typelevel::cats-effect:3.5.1
-//> using dep org.http4s::http4s-dsl:0.23.23
-//> using dep org.http4s::http4s-ember-server:0.23.23
-//> using dep org.http4s::http4s-circe:0.23.23
-//> using dep io.circe::circe-core:0.14.5
-//> using dep io.circe::circe-generic:0.14.5
-//> using dep pl.iterators::http4s-stir:0.2
+//> using dep org.typelevel::cats-effect::3.5.4
+//> using dep org.http4s::http4s-dsl::0.23.28
+//> using dep org.http4s::http4s-ember-server::0.23.28
+//> using dep org.http4s::http4s-circe::0.23.28
+//> using dep io.circe::circe-core::0.14.10
+//> using dep io.circe::circe-generic::0.14.10
+//> using dep pl.iterators::http4s-stir::0.4.0
 
 import org.http4s.Status
 import org.http4s.ember.server.EmberServerBuilder
@@ -94,17 +94,23 @@ val route: Route =
 
 object Main extends IOApp.Simple {
   val run = EmberServerBuilder
-    .default[IO]
-    .withHttpApp(route.toHttpRoutes.orNotFound)
-    .build
-    .use(_ => IO.never)
+          .default[IO]
+          .withHttpApp(route.toHttpRoutes.orNotFound)
+          .build
+          .use(_ => IO.never)
 }
+
 ```
 
+To run this service you can use `scala-cli run .`.
+
+Or maybe if you want, you can compile it to JS file: `scala-cli --power package --js --js-module-kind commonjs Main.scala`.
+
 ```scala 3
 // Main.test.scala
-//> using test.dep org.specs2::specs2-core:4.19.2
-//> using test.dep pl.iterators::http4s-stir-testkit:0.2
+//> using test.dep org.specs2::specs2-core:5.5.8
+//> using test.dep pl.iterators::http4s-stir-testkit:0.4.0
+//> using test.dep org.http4s::http4s-circe:0.23.28
 
 import org.http4s.Status
 import org.http4s.circe.CirceEntityEncoder.*
@@ -115,33 +121,36 @@ import org.specs2.mutable.Specification
 import pl.iterators.stir.testkit.Specs2RouteTest
 
 class MainRoutesSpec extends Specification with Specs2RouteTest {
-    override implicit val runtime: IORuntime = IORuntime.global
-
-    sequential
-    "The routes" should {
-        "create order" in {
-            Post("/create-order", Order(List(Item("foo", 42)))) ~> route ~> check {
-                responseAs[String] must contain("order created")
-                orders.head must beEqualTo(Item("foo", 42))
-            }
-        }
-        "retrieve an item if present" in {
-            orders = List(Item("foo", 42))
-            Get("/item/42") ~> route ~> check {
-                responseAs[Item] must beEqualTo(Item("foo", 42))
-            }
-        }
-        "return 404 if item is not present" in {
-            orders = List.empty
-            Get("/item/42") ~> route ~> check {
-                status must beEqualTo(Status.NotFound)
-            }
-        }
+  override implicit val runtime: IORuntime = IORuntime.global
+
+  sequential
+  "The routes" should {
+    "create order" in {
+      Post("/create-order", Order(List(Item("foo", 42)))) ~> route ~> check {
+        responseAs[String] must contain("order created")
+        orders.head must beEqualTo(Item("foo", 42))
+      }
+    }
+    "retrieve an item if present" in {
+      orders = List(Item("foo", 42))
+      Get("/item/42") ~> route ~> check {
+        responseAs[Item] must beEqualTo(Item("foo", 42))
+      }
     }
+    "return 404 if item is not present" in {
+      orders = List.empty
+      Get("/item/42") ~> route ~> check {
+        status must beEqualTo(Status.NotFound)
+      }
+    }
+  }
 }
+
 ```
 
-For a more comprehensive example showcasing additional directives see [examples](https://github.com/theiterators/http4s-stir/blob/master/examples/src/main/scala/Service.scala).
+To run the tests you can use `scala-cli test .`.
+
+For a more comprehensive example showcasing additional directives see [examples](https://github.com/theiterators/http4s-stir/blob/master/examples/src/main/scala/Service.scala). You can run it with `~examples/reStart`.
 
 ## Why this library?
 

diff --git a/core/src/main/scala/pl/iterators/stir/server/directives/DebuggingDirectives.scala b/core/src/main/scala/pl/iterators/stir/server/directives/DebuggingDirectives.scala
@@ -2,10 +2,9 @@ package pl.iterators.stir.server.directives
 
 import cats.effect.IO
 import cats.effect.std.Console
-import cats.implicits.toFlatMapOps
-import fs2.{ Chunk, Pull, Stream }
+import fs2.{ Pull, Stream }
 import org.http4s.server.middleware.Logger
-import org.http4s.{ EntityBody, Headers, Request, Response }
+import org.http4s.{ Headers, Request, Response }
 import org.typelevel.ci.CIString
 import pl.iterators.stir.server.{ Directive, Directive0, RouteResult }
 
@@ -25,26 +24,39 @@ trait DebuggingDirectives {
         DebuggingDirectives.logger(s)
       }
       val logWithTrimmingIndicator = indicateTrimming(maxBodyBytes, ctx.request.contentLength).andThen(log)
-      if (logBody && !ctx.request.isChunked) {
-        ctx.request.body.pull.unconsN(maxBodyBytes).flatMap {
-          case Some((head, tail)) =>
-            Pull.eval {
-              Logger.logMessage[IO, Request[IO]](ctx.request.withBodyStream(Stream.chunk(head)))(logHeaders,
-                logBody = true,
-                redactHeadersWhen)(logWithTrimmingIndicator).flatMap { _ =>
-                val newBody = Stream.chunk(head) ++ tail
-                val newRequest = ctx.request.withBodyStream(newBody)
-                val newCtx = ctx.copy(request = newRequest)
-                inner(())(newCtx)
+      val logWithBodyNotConsumedIndicator = indicateBodyNotConsumed(ctx.request.contentLength).andThen(log)
+
+      if (logBody && !ctx.request.isChunked && ctx.request.contentLength.exists(_ > 0)) {
+        IO.ref(false).flatMap { bodyConsumedRef =>
+          val newBody = ctx.request.body.pull.unconsN(maxBodyBytes, allowFewer = true).flatMap {
+            case Some((head, tail)) =>
+              Pull.output(head) >>
+              Pull.eval {
+                bodyConsumedRef.update(_ => true) *> Logger.logMessage[IO, Request[IO]](
+                  ctx.request.withBodyStream(Stream.chunk(head)))(logHeaders,
+                  logBody = true, redactHeadersWhen)(logWithTrimmingIndicator)
+              } >>
+              tail.pull.echo
+            case None =>
+              Pull.eval {
+                bodyConsumedRef.update(_ => true) *> Logger.logMessage[IO, Request[IO]](ctx.request)(logHeaders,
+                  logBody = false, redactHeadersWhen)(log)
+              }
+          }.stream
+          val newRequest = ctx.request.withBodyStream(newBody)
+          inner(())(ctx.copy(request = newRequest)).flatTap {
+            _ =>
+              bodyConsumedRef.get.flatMap {
+                bodyConsumed =>
+                  if (!bodyConsumed) {
+                    Logger.logMessage[IO, Request[IO]](newRequest)(logHeaders, logBody = false, redactHeadersWhen)(
+                      logWithBodyNotConsumedIndicator)
+                  } else {
+                    IO.unit
+                  }
               }
-            }.flatMap(r => Pull.output1(r))
-          case None =>
-            Pull.eval {
-              Logger.logMessage[IO, Request[IO]](ctx.request)(logHeaders, logBody = false, redactHeadersWhen)(
-                log).flatMap(_ =>
-                inner(())(ctx))
-            }.flatMap(r => Pull.output1(r))
-        }.stream.compile.onlyOrError
+          }
+        }
       } else {
         Logger.logMessage[IO, Request[IO]](ctx.request)(logHeaders, logBody = false, redactHeadersWhen)(log).flatMap(
           _ =>
@@ -70,11 +82,19 @@ trait DebuggingDirectives {
         case RouteResult.Complete(response) =>
           val logWithTrimmingIndicator = indicateTrimming(maxBodyBytes, response.contentLength).andThen(log)
           if (logBody && !response.isChunked) {
-            val bodyToLog = response.body.take(maxBodyBytes.toLong).chunks.flatMap(Stream.chunk)
-            Logger.logMessage[IO, Response[IO]](response.withBodyStream(bodyToLog))(
-              logHeaders,
-              logBody = true,
-              redactHeadersWhen)(logWithTrimmingIndicator).as(RouteResult.Complete(response))
+            val newBody = response.body.pull.unconsN(maxBodyBytes, allowFewer = true).flatMap {
+              case Some((head, tail)) =>
+                Pull.output(head) >>
+                Pull.eval {
+                  Logger.logMessage[IO, Response[IO]](response.withBodyStream(Stream.chunk(head)))(logHeaders,
+                    logBody = true, redactHeadersWhen)(logWithTrimmingIndicator)
+                } >>
+                tail.pull.echo
+              case None => Pull.eval {
+                  Logger.logMessage[IO, Response[IO]](response)(logHeaders, logBody = false, redactHeadersWhen)(log)
+                }
+            }.stream
+            IO.pure(RouteResult.Complete(response.copy(body = newBody)))
           } else {
             Logger.logMessage[IO, Response[IO]](response)(logHeaders, logBody = false, redactHeadersWhen)(log).as(
               RouteResult.Complete(response))
@@ -111,6 +131,15 @@ trait DebuggingDirectives {
         log
     }
   }
+
+  private def indicateBodyNotConsumed(contentLength: Option[Long]): String => String = { log =>
+    contentLength match {
+      case Some(length) =>
+        s"$log body=<not consumed> ($length bytes total)"
+      case None =>
+        s"$log body=<not consumed> (??? bytes total)"
+    }
+  }
 }
 
 object DebuggingDirectives extends DebuggingDirectives {

diff --git a/examples/src/main/scala/Service.scala b/examples/src/main/scala/Service.scala
@@ -101,15 +101,27 @@ object Main extends IOApp.Simple {
                   }
                 }
               }
+            } ~ (path("pipe") & extractRequest) { request =>
+              complete {
+                Status.Ok -> request.body
+              }
+            } ~ (path("empty") & extractRequest) { _ =>
+              complete {
+                Status.Ok
+              }
+            } ~ (post & path("empty") & extractRequest) { _ =>
+              complete {
+                Status.Ok
+              }
             } ~ (path("file-upload") & storeUploadedFiles("file", fi => new File("/tmp/" + fi.fileName))) { files =>
               complete {
                 Status.Ok -> s"File $files uploaded"
               }
             } ~ authenticateBasic("d-and-d-realm", authenticator) { _ =>
               path("file") {
-                getFromFile("project/plugins.sbt")
+                getFromFile("../src/main/scala/Service.scala")
               } ~ pathPrefix("dir") {
-                getFromDirectory("core/src/main")
+                getFromDirectory("../")
               }
             }
           } ~ path("ws") {