Skip to content

Add extensible content encoding filter #14199

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
arnabnandy7 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add extensible content encoding filter #14199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions documentation/manual/working/commonGuide/filters/GzipEncoding.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,17 @@ Scala

Java
: @[gzip-filter](code/detailedtopics/configuration/gzipencoding/CustomFilters.java)

## Supporting other content encodings

The response handling used by `GzipFilter` is available separately as `ContentEncodingFilter`. You can supply an
encoding name and a factory for a Pekko Streams `Flow`, allowing an application to add encodings such as Brotli or
Zstandard using an encoder library of its choice:

Scala
: @[custom-content-encoding](code/GzipEncoding.scala)

The flow factory must return a new encoding flow for each response stream. The filter handles `Accept-Encoding`
negotiation, sets `Content-Encoding` and `Vary`, preserves chunk trailers, and avoids encoding responses that must not
contain a body. The optional `shouldTranscode`, `chunkedThreshold`, and `threshold` parameters provide the same
selection and buffering controls used by the gzip filter.
15 changes: 15 additions & 0 deletions documentation/manual/working/commonGuide/filters/code/GzipEncoding.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,21 @@ class GzipEncoding extends PlaySpecification {
import play.api.http.DefaultHttpFilters
import play.filters.gzip.GzipFilter

// #custom-content-encoding
object CustomContentEncoding {
import org.apache.pekko.stream.scaladsl.Flow
import org.apache.pekko.stream.Materializer
import org.apache.pekko.util.ByteString
import play.filters.encoding.ContentEncodingFilter

class BrotliFilter(createBrotliFlow: () => Flow[ByteString, ByteString, ?])(implicit mat: Materializer)
extends ContentEncodingFilter(
encodingName = "br",
createFlow = createBrotliFlow
)
}
// #custom-content-encoding

class Filters @Inject() (gzipFilter: GzipFilter) extends DefaultHttpFilters(gzipFilter)

"gzip filter" should {
Expand Down
141 changes: 141 additions & 0 deletions web/play-filters-helpers/src/main/scala/play/filters/encoding/ContentEncodingFilter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
/*
* Copyright (C) from 2022 The Play Framework Contributors <https://github.com/playframework>, 2011-2021 Lightbend Inc. <https://www.lightbend.com>
*/

package play.filters.encoding

import scala.concurrent.ExecutionContext
import scala.concurrent.Future

import org.apache.pekko.stream.scaladsl._
import org.apache.pekko.stream.FlowShape
import org.apache.pekko.stream.Materializer
import org.apache.pekko.stream.OverflowStrategy
import org.apache.pekko.util.ByteString
import play.api.http._
import play.api.mvc._
import play.api.mvc.RequestHeader.acceptHeader

/**
* A filter that encodes response bodies with a supplied stream transformation.
*
* The response is encoded when the request accepts the configured encoding and the response is suitable for encoding.
* Responses to HEAD requests, responses without content, responses that already have a content encoding, and responses
* at or below the configured threshold are not encoded.
*
* @param encodingName The content coding token used in the Accept-Encoding and Content-Encoding headers.
* @param createFlow Creates a new flow that encodes a response body.
* @param shouldTranscode Whether the given request and result should be encoded.
* @param chunkedThreshold The content length threshold after which the filter switches to chunked encoding.
* @param threshold The response body size threshold below which responses are not encoded.
*/
class ContentEncodingFilter(
encodingName: String,
createFlow: () => Flow[ByteString, ByteString, ?],
shouldTranscode: (RequestHeader, Result) => Boolean = (_, _) => true,
chunkedThreshold: Int = 102400,
threshold: Int = 0
)(implicit mat: Materializer)
extends EssentialFilter {
import play.api.http.HeaderNames._

def apply(next: EssentialAction): EssentialAction = new EssentialAction {
implicit val ec: ExecutionContext = mat.executionContext

def apply(request: RequestHeader) = {
if (mayEncode(request)) {
next(request).mapFuture(result => handleResult(request, result))
} else {
next(request)
}
}
}

private def handleResult(request: RequestHeader, result: Result): Future[Result] = {
implicit val ec: ExecutionContext = mat.executionContext
if (shouldEncode(result) && shouldTranscode(request, result)) {
val header = result.header.copy(headers = setupHeader(result.header))

result.body match {
case HttpEntity.Strict(data, contentType) =>
encodeStrictEntity(Source.single(data), contentType)
.map(entity => result.copy(header = header, body = entity))

case entity @ HttpEntity.Streamed(_, Some(contentLength), contentType) if contentLength <= chunkedThreshold =>
// It's below the chunked threshold, so buffer then encode and send.
encodeStrictEntity(entity.data, contentType)
.map(strictEntity => result.copy(header = header, body = strictEntity))

case HttpEntity.Streamed(data, _, contentType) if request.version == HttpProtocol.HTTP_1_0 =>
// HTTP 1.0 cannot use chunked encoding, so use a close-delimited body without a content length.
val encoded = data.via(createFlow())
Future.successful(result.copy(header = header, body = HttpEntity.Streamed(encoded, None, contentType)))

case HttpEntity.Streamed(data, _, contentType) =>
// It's above the chunked threshold, so stream the encoded body as chunks.
val encoded = data.via(createFlow()).map(d => HttpChunk.Chunk(d))
Future.successful(result.copy(header = header, body = HttpEntity.Chunked(encoded, contentType)))

case HttpEntity.Chunked(chunks, contentType) =>
val encodingFlow = Flow.fromGraph(GraphDSL.create[FlowShape[HttpChunk, HttpChunk]]() { implicit builder =>
import GraphDSL.Implicits._

val extractChunks = Flow[HttpChunk].collect { case HttpChunk.Chunk(data) => data }
val createChunks = Flow[ByteString].map[HttpChunk](HttpChunk.Chunk.apply)
val filterLastChunk = Flow[HttpChunk]
.filter(_.isInstanceOf[HttpChunk.LastChunk])
// Concat does not demand the last chunk until the encoding flow completes. Buffering here lets the
// broadcast start while preserving the last chunk and its trailers.
.buffer(1, OverflowStrategy.backpressure)

val broadcast = builder.add(Broadcast[HttpChunk](2))
val concat = builder.add(Concat[HttpChunk]())

broadcast.out(0) ~> extractChunks ~> createFlow() ~> createChunks ~> concat.in(0)
broadcast.out(1) ~> filterLastChunk ~> concat.in(1)

new FlowShape(broadcast.in, concat.out)
})

Future.successful(
result.copy(header = header, body = HttpEntity.Chunked(chunks.via(encodingFlow), contentType))
)
}
} else {
Future.successful(result)
}
}

private def encodeStrictEntity(source: Source[ByteString, Any], contentType: Option[String])(
implicit ec: ExecutionContext
) = {
val encoded = source.via(createFlow()).runFold(ByteString.empty)(_ ++ _)
encoded.map(data => HttpEntity.Strict(data, contentType))
}

private def mayEncode(request: RequestHeader) =
request.method != "HEAD" && encodingIsAcceptedAndPreferredBy(request)

private def encodingIsAcceptedAndPreferredBy(request: RequestHeader) = {
val codings = acceptHeader(request.headers, ACCEPT_ENCODING)
def explicitQValue(coding: String) = codings.collectFirst { case (q, c) if c.equalsIgnoreCase(coding) => q }
def defaultQValue(coding: String) = if (coding == "identity") 0.001d else 0d
def qvalue(coding: String) = explicitQValue(coding).orElse(explicitQValue("*")).getOrElse(defaultQValue(coding))

qvalue(encodingName) > 0d && qvalue(encodingName) >= qvalue("identity")
}

private def shouldEncode(result: Result) =
isAllowedContent(result.header) &&
isNotAlreadyEncoded(result.header) &&
!result.body.isKnownEmpty &&
result.body.contentLength.forall(_ > threshold)

private def isAllowedContent(header: ResponseHeader) =
header.status != Status.NO_CONTENT && header.status != Status.NOT_MODIFIED

private def isNotAlreadyEncoded(header: ResponseHeader) = header.headers.get(CONTENT_ENCODING).isEmpty

private def setupHeader(rh: ResponseHeader): Map[String, String] =
rh.headers + (CONTENT_ENCODING -> encodingName) + rh.varyWith(ACCEPT_ENCODING)
}
146 changes: 9 additions & 137 deletions web/play-filters-helpers/src/main/scala/play/filters/gzip/GzipFilter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,26 +7,20 @@ package play.filters.gzip
import java.util.function.BiFunction
import java.util.zip.Deflater

import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.jdk.FunctionConverters._

import com.typesafe.config.ConfigMemorySize
import jakarta.inject.Inject
import jakarta.inject.Provider
import jakarta.inject.Singleton
import org.apache.pekko.stream.scaladsl._
import org.apache.pekko.stream.FlowShape
import org.apache.pekko.stream.Materializer
import org.apache.pekko.stream.OverflowStrategy
import org.apache.pekko.util.ByteString
import play.api.http._
import play.api.inject._
import play.api.libs.streams.GzipFlow
import play.api.mvc._
import play.api.mvc.RequestHeader.acceptHeader
import play.api.Configuration
import play.api.Logger
import play.filters.encoding.ContentEncodingFilter

/**
* A gzip filter.
Expand All @@ -49,7 +43,13 @@ import play.api.Logger
*/
@Singleton
class GzipFilter @Inject() (config: GzipFilterConfig)(implicit mat: Materializer) extends EssentialFilter {
import play.api.http.HeaderNames._
private val contentEncodingFilter = new ContentEncodingFilter(
encodingName = "gzip",
createFlow = () => GzipFlow.gzip(config.bufferSize, config.compressionLevel),
shouldTranscode = config.shouldGzip,
chunkedThreshold = config.chunkedThreshold,
threshold = config.threshold
)

def this(
bufferSize: Int = 8192,
Expand All @@ -60,135 +60,7 @@ class GzipFilter @Inject() (config: GzipFilterConfig)(implicit mat: Materializer
)(implicit mat: Materializer) =
this(GzipFilterConfig(bufferSize, chunkedThreshold, threshold, shouldGzip, compressionLevel))

def apply(next: EssentialAction): EssentialAction = new EssentialAction {
implicit val ec: ExecutionContext = mat.executionContext

def apply(request: RequestHeader) = {
if (mayCompress(request)) {
next(request).mapFuture(result => handleResult(request, result))
} else {
next(request)
}
}
}

private def createGzipFlow: Flow[ByteString, ByteString, ?] =
GzipFlow.gzip(config.bufferSize, config.compressionLevel)

private def handleResult(request: RequestHeader, result: Result): Future[Result] = {
implicit val ec = mat.executionContext
if (shouldCompress(result) && config.shouldGzip(request, result)) {
val header = result.header.copy(headers = setupHeader(result.header))

result.body match {
case HttpEntity.Strict(data, contentType) =>
compressStrictEntity(Source.single(data), contentType)
.map(entity => result.copy(header = header, body = entity))

case entity @ HttpEntity.Streamed(_, Some(contentLength), contentType)
if contentLength <= config.chunkedThreshold =>
// It's below the chunked threshold, so buffer then compress and send
compressStrictEntity(entity.data, contentType)
.map(strictEntity => result.copy(header = header, body = strictEntity))

case HttpEntity.Streamed(data, _, contentType) if request.version == HttpProtocol.HTTP_1_0 =>
// It's above the chunked threshold, but we can't chunk it because we're using HTTP 1.0.
// Instead, we use a close delimited body (ie, regular body with no content length)
val gzipped = data.via(createGzipFlow)
Future.successful(
result.copy(header = header, body = HttpEntity.Streamed(gzipped, None, contentType))
)

case HttpEntity.Streamed(data, _, contentType) =>
// It's above the chunked threshold, compress through the gzip flow, and send as chunked
val gzipped = data.via(createGzipFlow).map(d => HttpChunk.Chunk(d))
Future.successful(
result.copy(header = header, body = HttpEntity.Chunked(gzipped, contentType))
)

case HttpEntity.Chunked(chunks, contentType) =>
val gzipFlow = Flow.fromGraph(GraphDSL.create[FlowShape[HttpChunk, HttpChunk]]() { implicit builder =>
import GraphDSL.Implicits._

val extractChunks = Flow[HttpChunk].collect { case HttpChunk.Chunk(data) => data }
val createChunks = Flow[ByteString].map[HttpChunk](HttpChunk.Chunk.apply)
val filterLastChunk = Flow[HttpChunk]
.filter(_.isInstanceOf[HttpChunk.LastChunk])
// Since we're doing a merge by concatenating, the filter last chunk won't receive demand until the gzip
// flow is finished. But the broadcast won't start broadcasting until both flows start demanding. So we
// put a buffer of one in to ensure the filter last chunk flow demands from the broadcast.
.buffer(1, OverflowStrategy.backpressure)

val broadcast = builder.add(Broadcast[HttpChunk](2))
val concat = builder.add(Concat[HttpChunk]())

// Broadcast the stream through two separate flows, one that collects chunks and turns them into
// ByteStrings, sends those ByteStrings through the Gzip flow, and then turns them back into chunks,
// the other that just allows the last chunk through. Then concat those two flows together.
broadcast.out(0) ~> extractChunks ~> createGzipFlow ~> createChunks ~> concat.in(0)
broadcast.out(1) ~> filterLastChunk ~> concat.in(1)

new FlowShape(broadcast.in, concat.out)
})

Future.successful(
result.copy(header = header, body = HttpEntity.Chunked(chunks.via(gzipFlow), contentType))
)
}
} else {
Future.successful(result)
}
}

private def compressStrictEntity(source: Source[ByteString, Any], contentType: Option[String])(
implicit ec: ExecutionContext
) = {
val compressed = source.via(createGzipFlow).runFold(ByteString.empty)(_ ++ _)
compressed.map(data => HttpEntity.Strict(data, contentType))
}

/**
* Whether this request may be compressed.
*/
private def mayCompress(request: RequestHeader) =
request.method != "HEAD" && gzipIsAcceptedAndPreferredBy(request)

private def gzipIsAcceptedAndPreferredBy(request: RequestHeader) = {
val codings = acceptHeader(request.headers, ACCEPT_ENCODING)
def explicitQValue(coding: String) = codings.collectFirst { case (q, c) if c.equalsIgnoreCase(coding) => q }
def defaultQValue(coding: String) = if (coding == "identity") 0.001d else 0d
def qvalue(coding: String) = explicitQValue(coding).orElse(explicitQValue("*")).getOrElse(defaultQValue(coding))

qvalue("gzip") > 0d && qvalue("gzip") >= qvalue("identity")
}

/**
* Whether this response should be compressed. Responses that may not contain content won't be compressed, nor will
* responses that already define a content encoding. Empty responses also shouldn't be compressed, as they will
* actually always get bigger. Also responses whose body size are equal or lower than the given byte threshold won't
* be compressed, because it's assumed they end up being bigger than the original body.
*/
private def shouldCompress(result: Result) =
isAllowedContent(result.header) &&
isNotAlreadyCompressed(result.header) &&
!result.body.isKnownEmpty &&
result.body.contentLength.forall(_ > config.threshold)

/**
* Certain response codes are forbidden by the HTTP spec to contain content, but a gzipped response always contains
* a minimum of 20 bytes, even for empty responses.
*/
private def isAllowedContent(header: ResponseHeader) =
header.status != Status.NO_CONTENT && header.status != Status.NOT_MODIFIED

/**
* Of course, we don't want to double compress responses
*/
private def isNotAlreadyCompressed(header: ResponseHeader) = header.headers.get(CONTENT_ENCODING).isEmpty

private def setupHeader(rh: ResponseHeader): Map[String, String] = {
rh.headers + (CONTENT_ENCODING -> "gzip") + rh.varyWith(ACCEPT_ENCODING)
}
def apply(next: EssentialAction): EssentialAction = contentEncodingFilter(next)
}

/**
Expand Down
Loading
Loading