Merge branch 'main' into scala-3

Author: johanandren

LICENSE.txt

@@ -15,7 +15,7 @@ Additional Use Grant:
     Connecting to a Play Framework websocket and/or Play Framework
     request/response bodies for server and play-ws client.
 
-Change Date:          TBD ( 3 years after 10.4.0 release )
+Change Date:          2025-10-25
 
 Change License:       Apache License, Version 2.0
 

akka-http-core/src/main/java/akka/http/javadsl/model/StatusCodes.java

@@ -57,10 +57,16 @@ private StatusCodes() {}
     public static final StatusCode GONE = akka.http.scaladsl.model.StatusCodes.Gone();
     public static final StatusCode LENGTH_REQUIRED = akka.http.scaladsl.model.StatusCodes.LengthRequired();
     public static final StatusCode PRECONDITION_FAILED = akka.http.scaladsl.model.StatusCodes.PreconditionFailed();
+    public static final StatusCode CONTENT_TOO_LARGE = akka.http.scaladsl.model.StatusCodes.ContentTooLarge();
+
+    /**
+     * @deprecated deprecated in favor of CONTENT_TOO_LARGE
+     */
+    @Deprecated
     public static final StatusCode PAYLOAD_TOO_LARGE = akka.http.scaladsl.model.StatusCodes.PayloadTooLarge();
 
     /**
-     * @deprecated deprecated in favor of PAYLOAD_TOO_LARGE
+     * @deprecated deprecated in favor of CONTENT_TOO_LARGE
      */
     @Deprecated
     public static final StatusCode REQUEST_ENTITY_TOO_LARGE = akka.http.scaladsl.model.StatusCodes.RequestEntityTooLarge();
@@ -83,6 +89,12 @@ private StatusCodes() {}
     public static final StatusCode IM_A_TEAPOT = akka.http.scaladsl.model.StatusCodes.ImATeapot();
     public static final StatusCode ENHANCE_YOUR_CALM = akka.http.scaladsl.model.StatusCodes.EnhanceYourCalm();
     public static final StatusCode MISDIRECTED_REQUEST = akka.http.scaladsl.model.StatusCodes.MisdirectedRequest();
+    public static final StatusCode UNPROCESSABLE_CONTENT = akka.http.scaladsl.model.StatusCodes.UnprocessableContent();
+
+    /**
+     * @deprecated deprecated in favor of UNPROCESSABLE_CONTENT
+     */
+    @Deprecated
     public static final StatusCode UNPROCESSABLE_ENTITY = akka.http.scaladsl.model.StatusCodes.UnprocessableEntity();
     public static final StatusCode LOCKED = akka.http.scaladsl.model.StatusCodes.Locked();
     public static final StatusCode FAILED_DEPENDENCY = akka.http.scaladsl.model.StatusCodes.FailedDependency();

akka-http-core/src/main/scala/akka/http/impl/engine/http2/ProtocolSwitch.scala

@@ -60,11 +60,7 @@ private[http] object ProtocolSwitch {
                 }
             })
 
-            private val ignorePull = new OutHandler {
-              def onPull(): Unit = ()
-            }
-
-            setHandler(netOut, ignorePull)
+            setHandler(netOut, GraphStageLogic.EagerTerminateOutput)
 
             def install(serverImplementation: HttpImplementation, firstElement: SslTlsInbound): Unit = {
               val networkSide = Flow.fromSinkAndSource(serverDataIn.sink, serverDataOut.source)

akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpRequestParser.scala

@@ -209,7 +209,7 @@ private[http] final class HttpRequestParser(
             setCompletionHandling(HttpMessageParser.CompletionOk)
             startNewMessage(input, bodyStart)
           } else if (!method.isEntityAccepted) {
-            failMessageStart(UnprocessableEntity, s"${method.name} requests must not have an entity")
+            failMessageStart(UnprocessableContent, s"${method.name} requests must not have an entity")
           } else if (contentLength <= input.size - bodyStart) {
             val cl = contentLength.toInt
             emitRequestStart(strictEntity(cth, input, bodyStart, cl))
@@ -221,7 +221,7 @@ private[http] final class HttpRequestParser(
           }
         } else {
           if (!method.isEntityAccepted) {
-            failMessageStart(UnprocessableEntity, s"${method.name} requests must not have an entity")
+            failMessageStart(UnprocessableContent, s"${method.name} requests must not have an entity")
           } else {
             if (clh.isEmpty) {
               emitRequestStart(chunkedEntity(cth), headers)

akka-http-core/src/main/scala/akka/http/impl/engine/rendering/HttpResponseRendererFactory.scala

@@ -51,37 +51,36 @@ private[http] class HttpResponseRendererFactory(
     val shape: FlowShape[ResponseRenderingContext, ResponseRenderingOutput] = FlowShape(in, out)
 
     def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
-      new GraphStageLogic(shape) {
+      new GraphStageLogic(shape) with InHandler {
         var closeMode: CloseMode = DontClose // signals what to do after the current response
         def close: Boolean = closeMode != DontClose
         def closeIf(cond: Boolean): Unit = if (cond) closeMode = CloseConnection
         var transferSink: Option[SubSinkInlet[ByteString]] = None
         def transferring: Boolean = transferSink.isDefined
 
-        setHandler(in, new InHandler {
-          override def onPush(): Unit =
-            render(grab(in)) match {
-              case Strict(outElement) =>
-                push(out, outElement)
-                if (close) completeStage()
-              case HeadersAndStreamedEntity(headerData, outStream) =>
-                try transfer(headerData, outStream)
-                catch {
-                  case NonFatal(e) =>
-                    log.error(e, s"Rendering of response failed because response entity stream materialization failed with '${e.getMessage}'. Sending out 500 response instead.")
-                    push(out, render(ResponseRenderingContext(HttpResponse(500, entity = StatusCodes.InternalServerError.defaultMessage))).asInstanceOf[Strict].bytes)
-                }
-            }
+        override def onPush(): Unit =
+          render(grab(in)) match {
+            case Strict(outElement) =>
+              push(out, outElement)
+              if (close) completeStage()
+            case HeadersAndStreamedEntity(headerData, outStream) =>
+              try transfer(headerData, outStream)
+              catch {
+                case NonFatal(e) =>
+                  log.error(e, s"Rendering of response failed because response entity stream materialization failed with '${e.getMessage}'. Sending out 500 response instead.")
+                  push(out, render(ResponseRenderingContext(HttpResponse(500, entity = StatusCodes.InternalServerError.defaultMessage))).asInstanceOf[Strict].bytes)
+              }
+          }
 
-          override def onUpstreamFinish(): Unit =
-            if (transferring) closeMode = CloseConnection
-            else completeStage()
+        override def onUpstreamFinish(): Unit =
+          if (transferring) closeMode = CloseConnection
+          else completeStage()
+
+        override def onUpstreamFailure(ex: Throwable): Unit = {
+          stopTransfer()
+          failStage(ex)
+        }
 
-          override def onUpstreamFailure(ex: Throwable): Unit = {
-            stopTransfer()
-            failStage(ex)
-          }
-        })
         private val waitForDemandHandler = new OutHandler {
           def onPull(): Unit = if (!hasBeenPulled(in)) tryPull(in)
         }
@@ -92,6 +91,7 @@ private[http] class HttpResponseRendererFactory(
           transferSink = None
         }
 
+        setHandler(in, this)
         setHandler(out, waitForDemandHandler)
         def transfer(headerData: ByteString, outStream: Source[ByteString, Any]): Unit = {
           val sinkIn = new SubSinkInlet[ByteString]("RenderingSink")

akka-http-core/src/main/scala/akka/http/impl/engine/server/HttpServerBluePrint.scala

@@ -499,7 +499,7 @@ private[http] object HttpServerBluePrint {
                 case None     => s"Aggregated data length of request entity exceeds the configured limit of $limit bytes"
               }
               val info = ErrorInfo(summary, "Consider increasing the value of akka.http.server.parsing.max-content-length")
-              finishWithIllegalRequestError(StatusCodes.PayloadTooLarge, info)
+              finishWithIllegalRequestError(StatusCodes.ContentTooLarge, info)
 
             case IllegalUriException(errorInfo) =>
               finishWithIllegalRequestError(StatusCodes.BadRequest, errorInfo)

akka-http-core/src/main/scala/akka/http/impl/engine/ws/FrameEventRenderer.scala

@@ -33,7 +33,7 @@ private[http] final class FrameEventRenderer extends GraphStage[FlowShape[FrameE
 
           push(out, renderStart(start))
 
-        case f: FrameData =>
+        case _: FrameData =>
           fail(out, new IllegalStateException("unexpected FrameData (need FrameStart first)"))
       }
     }

akka-http-core/src/main/scala/akka/http/impl/engine/ws/WebSocket.scala

@@ -271,16 +271,14 @@ private[http] object WebSocket {
 
     val shape = new FlowShape(in, out)
 
-    def createLogic(effectiveAttributes: Attributes) = new GraphStageLogic(shape) {
-      setHandler(out, new OutHandler {
+    def createLogic(effectiveAttributes: Attributes): GraphStageLogic with InHandler with OutHandler =
+      new GraphStageLogic(shape) with InHandler with OutHandler {
         override def onPull(): Unit = pull(in)
-      })
-      setHandler(in, new InHandler {
         override def onPush(): Unit = push(out, grab(in))
         override def onUpstreamFinish(): Unit = emit(out, UserHandlerCompleted, () => completeStage())
         override def onUpstreamFailure(ex: Throwable): Unit = emit(out, UserHandlerErredOut(ex), () => completeStage())
-      })
-    }
+        setHandlers(in, out, this)
+      }
   }
 
   case object Tick

akka-http-core/src/main/scala/akka/http/impl/util/package.scala

@@ -111,22 +111,20 @@ package util {
 
     override val shape = FlowShape(byteStringIn, httpEntityOut)
 
-    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new TimerGraphStageLogic(shape) {
-      val bytes = ByteString.newBuilder
-      private var emptyStream = false
+    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+      new TimerGraphStageLogic(shape) with InHandler with OutHandler {
+        val bytes = ByteString.newBuilder
+        private var emptyStream = false
 
-      override def preStart(): Unit = scheduleOnce("ToStrictTimeoutTimer", timeout)
+        override def preStart(): Unit = scheduleOnce("ToStrictTimeoutTimer", timeout)
 
-      setHandler(httpEntityOut, new OutHandler {
         override def onPull(): Unit = {
           if (emptyStream) {
             push(httpEntityOut, HttpEntity.Strict(contentType, ByteString.empty))
             completeStage()
           } else pull(byteStringIn)
         }
-      })
 
-      setHandler(byteStringIn, new InHandler {
         override def onPush(): Unit = {
           bytes ++= grab(byteStringIn)
           maxBytes match {
@@ -136,18 +134,20 @@ package util {
               pull(byteStringIn)
           }
         }
+
         override def onUpstreamFinish(): Unit = {
           if (isAvailable(httpEntityOut)) {
             push(httpEntityOut, HttpEntity.Strict(contentType, bytes.result()))
             completeStage()
           } else emptyStream = true
         }
-      })
 
-      override def onTimer(key: Any): Unit =
-        failStage(new java.util.concurrent.TimeoutException(
-          s"HttpEntity.toStrict timed out after $timeout while still waiting for outstanding data"))
-    }
+        setHandlers(byteStringIn, httpEntityOut, this)
+
+        override def onTimer(key: Any): Unit =
+          failStage(new java.util.concurrent.TimeoutException(
+            s"HttpEntity.toStrict timed out after $timeout while still waiting for outstanding data"))
+      }
 
     override def toString = "ToStrict"
   }

akka-http-core/src/main/scala/akka/http/scaladsl/model/StatusCode.scala

@@ -141,9 +141,11 @@ object StatusCodes extends ObjectRegistry[Int, StatusCode] {
   val Gone                         = reg(c(410)("Gone", "The resource requested is no longer available and will not be available again."))
   val LengthRequired               = reg(c(411)("Length Required", "The request did not specify the length of its content, which is required by the requested resource."))
   val PreconditionFailed           = reg(c(412)("Precondition Failed", "The server does not meet one of the preconditions that the requester put on the request."))
-  val PayloadTooLarge              = reg(c(413)("Payload Too Large", "The request payload is larger than the server is willing or able to process."))
-  @deprecated("deprecated in favor of PayloadTooLarge", "10.1.11")
-  val RequestEntityTooLarge        = PayloadTooLarge
+  val ContentTooLarge              = reg(c(413)("Content Too Large", "The request content is larger than the server is willing or able to process."))
+  @deprecated("deprecated in favor of ContentTooLarge", "10.4.0")
+  val PayloadTooLarge              = ContentTooLarge
+  @deprecated("deprecated in favor of ContentTooLarge", "10.4.0")
+  val RequestEntityTooLarge        = ContentTooLarge
   val UriTooLong                   = reg(c(414)("URI Too Long", "The URI provided was too long for the server to process."))
   @deprecated("deprecated in favor of UriTooLong", "10.1.11")
   val RequestUriTooLong            = UriTooLong
@@ -155,7 +157,9 @@ object StatusCodes extends ObjectRegistry[Int, StatusCode] {
   val ImATeapot                    = reg(c(418)("I'm a teapot", "The resulting entity body MAY be short and stout."))
   val EnhanceYourCalm              = reg(c(420)("Enhance Your Calm", "You are being rate-limited.")) // Twitter only
   val MisdirectedRequest           = reg(c(421)("Misdirected Request", "The request was directed at a server that is not able to produce a response.")) // HTTP/2 only. https://tools.ietf.org/html/rfc7540#section-9.1.2
-  val UnprocessableEntity          = reg(c(422)("Unprocessable Entity", "The request was well-formed but was unable to be followed due to semantic errors."))
+  val UnprocessableContent          = reg(c(422)("Unprocessable Content", "The request was well-formed but was unable to be followed due to semantic errors."))
+  @deprecated("deprecated in favor of UnprocessableContent", "10.4.0")
+  val UnprocessableEntity          = UnprocessableContent
   val Locked                       = reg(c(423)("Locked", "The resource that is being accessed is locked."))
   val FailedDependency             = reg(c(424)("Failed Dependency", "The request failed due to failure of a previous request."))
   val TooEarly                     = reg(c(425)("Too Early", "The server is unwilling to risk processing a request that might be replayed.")) // RFC 8470