Merge pull request #317 from newrelic/fix/akka-body-NR-299709

NR-299709: Workaround for issue #310 (#310)
newrelic · Aug 13, 2024 · e800262 · e800262
2 parents 9ec9995 + 689dc34
commit e800262
Show file tree

Hide file tree

Showing 9 changed files with 68 additions and 45 deletions.
diff --git a/...tp-2.11_10.0.0/src/main/scala/akka/http/scaladsl/server/CsecAkkaHttpContextFunction.scala b/...tp-2.11_10.0.0/src/main/scala/akka/http/scaladsl/server/CsecAkkaHttpContextFunction.scala
@@ -8,6 +8,7 @@
 package akka.http.scaladsl.server
 
 import akka.Done
+import akka.http.scaladsl.model.HttpEntity
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
 import akka.util.ByteString
@@ -55,15 +56,17 @@ class CsecContextWrapper(original: Function1[RequestContext, Future[RouteResult]
   override def apply(ctx: RequestContext): Future[RouteResult] = {
     try {
 
-      var httpRequest = ctx.request;
+      val httpRequest = ctx.request;
       val body: lang.StringBuilder = new lang.StringBuilder();
       val dataBytes: Source[ByteString, AnyRef] = httpRequest.entity.getDataBytes()
       val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-        val chunk = byteString.utf8String
-        body.append(chunk)
+      if (!httpRequest.entity.isInstanceOf[HttpEntity.Chunked]) {
+        val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+          val chunk = byteString.utf8String
+          body.append(chunk)
+        }
+        val processingResult: Future[Done] = dataBytes.runWith(sink, ctx.materializer)
       }
-      val processingResult: Future[Done] = dataBytes.runWith(sink, ctx.materializer)
       AkkaCoreUtils.preProcessHttpRequest(isLockAquired, httpRequest, body, NewRelic.getAgent.getTransaction.getToken);
       original.apply(ctx)
     } catch {

diff --git a/...urity/akka-http-core-10.0/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala b/...urity/akka-http-core-10.0/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -26,11 +26,14 @@ class AkkaAsyncRequestHandler(handler: HttpRequest ⇒ Future[HttpResponse])(imp
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
+
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val futureResponse: Future[HttpResponse] = handler.apply(param)
     futureResponse.flatMap(ResponseFutureHelper.wrapResponseAsync(NewRelic.getAgent.getTransaction.getToken, materializer))

diff --git a/...curity/akka-http-core-10.0/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala b/...curity/akka-http-core-10.0/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -26,11 +26,14 @@ class AkkaSyncRequestHandler(handler: HttpRequest ⇒ HttpResponse)(implicit mat
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val response: HttpResponse = handler.apply(param)
     ResponseFutureHelper.wrapResponseSync(response, materializer)

diff --git a/...ka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala b/...ka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -26,11 +26,13 @@ class AkkaAsyncRequestHandler(handler: HttpRequest ⇒ Future[HttpResponse])(imp
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val futureResponse: Future[HttpResponse] = handler.apply(param)
     futureResponse.flatMap(ResponseFutureHelper.wrapResponseAsync(NewRelic.getAgent.getTransaction.getToken, materializer))

diff --git a/...kka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala b/...kka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -26,11 +26,13 @@ class AkkaSyncRequestHandler(handler: HttpRequest ⇒ HttpResponse)(implicit mat
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val response: HttpResponse = handler.apply(param)
     ResponseFutureHelper.wrapResponseSync(response, materializer)

diff --git a/...kka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala b/...kka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -26,11 +26,13 @@ class AkkaAsyncRequestHandler(handler: HttpRequest ⇒ Future[HttpResponse])(imp
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val futureResponse: Future[HttpResponse] = handler.apply(param)
     futureResponse.flatMap(ResponseFutureHelper.wrapResponseAsync(NewRelic.getAgent.getTransaction.getToken, materializer))

diff --git a/...akka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala b/...akka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -26,11 +26,13 @@ class AkkaSyncRequestHandler(handler: HttpRequest ⇒ HttpResponse)(implicit mat
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val response: HttpResponse = handler.apply(param)
     ResponseFutureHelper.wrapResponseSync(response, materializer)

diff --git a/...kka-http-core-2.13_10.2.0/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala b/...kka-http-core-2.13_10.2.0/src/main/scala/akka/http/scaladsl/AkkaAsyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -26,11 +26,14 @@ class AkkaAsyncRequestHandler(handler: HttpRequest ⇒ Future[HttpResponse])(imp
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val futureResponse: Future[HttpResponse] = handler.apply(param)
     futureResponse.flatMap(ResponseFutureHelper.wrapResponseAsync(NewRelic.getAgent.getTransaction.getToken, materializer))

diff --git a/...akka-http-core-2.13_10.2.0/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala b/...akka-http-core-2.13_10.2.0/src/main/scala/akka/http/scaladsl/AkkaSyncRequestHandler.scala
@@ -8,7 +8,7 @@
 package akka.http.scaladsl
 
 import akka.Done
-import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
+import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 import akka.stream.javadsl.Source
 import akka.stream.scaladsl.Sink
@@ -27,11 +27,14 @@ class AkkaSyncRequestHandler(handler: HttpRequest ⇒ HttpResponse)(implicit mat
     val body: lang.StringBuilder = new lang.StringBuilder();
     val dataBytes: Source[ByteString, AnyRef] = param.entity.getDataBytes()
     val isLockAquired = AkkaCoreUtils.acquireServletLockIfPossible();
-    val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
-      val chunk = byteString.utf8String
-      body.append(chunk)
+
+    if (!param.entity.isInstanceOf[HttpEntity.Chunked]) {
+      val sink: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString] { byteString =>
+        val chunk = byteString.utf8String
+        body.append(chunk)
+      }
+      val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     }
-    val processingResult: Future[Done] = dataBytes.runWith(sink, materializer)
     AkkaCoreUtils.preProcessHttpRequest(isLockAquired, param, body, NewRelic.getAgent.getTransaction.getToken);
     val response: HttpResponse = handler.apply(param)
     ResponseFutureHelper.wrapResponseSync(response, materializer)