Skip to content
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

Attempt fix to StreamingForm blocking due to unsafe.run #2845

Merged
merged 4 commits into from
May 14, 2024
Merged

Attempt fix to StreamingForm blocking due to unsafe.run #2845

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bb673a3
Attempt fix to StreamingForm blocking
kyri-petrou May 14, 2024
b1c2bdd
Run stream in blocking threadpool
kyri-petrou May 14, 2024
7a3aa0a
Add FIXME comment
kyri-petrou May 14, 2024
e7c5bb7
Disable flaky test for now
kyri-petrou May 14, 2024
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
2 changes: 1 addition & 1 deletion zio-http/jvm/src/test/scala/zio/http/ClientHttpsSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ object ClientHttpsSpec extends ZIOHttpSpec {
),
),
)
} @@ nonFlaky(20),
} @@ nonFlaky(20) @@ ignore,
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The server the test uses seems to be not working at the moment which is causing CI to fail so I disabled the test. We should open a separate ticket to use a local server instance (perhaps even zio-http now that it supports SSL) instead of relying on a remote server

)
.provideSomeLayer[Client](Scope.default)
.provideShared(
Expand Down
2 changes: 1 addition & 1 deletion zio-http/jvm/src/test/scala/zio/http/FormSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -314,5 +314,5 @@ object FormSpec extends ZIOHttpSpec {
) @@ sequential

def spec =
suite("FormSpec")(urlEncodedSuite, multiFormSuite, multiFormStreamingSuite) @@ blocking
suite("FormSpec")(urlEncodedSuite, multiFormSuite, multiFormStreamingSuite)
}
2 changes: 1 addition & 1 deletion zio-http/jvm/src/test/scala/zio/http/endpoint/MultipartSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,6 @@ object MultipartSpec extends ZIOHttpSpec {
)
}
},
) @@ TestAspect.blocking,
),
)
}
15 changes: 9 additions & 6 deletions zio-http/shared/src/main/scala/zio/http/FormField.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
package zio.http
import java.nio.charset._

import scala.util.Try

import zio._
import zio.stacktracer.TracingImplicits.disableAutoTrace

Expand Down Expand Up @@ -145,7 +147,7 @@ object FormField {
private[http] def fromFormAST(
ast: Chunk[FormAST],
defaultCharset: Charset = StandardCharsets.UTF_8,
)(implicit trace: Trace): ZIO[Any, FormDecodingError, FormField] = {
): Either[FormDecodingError, FormField] = {
val extract =
ast.foldLeft(
(
Expand All @@ -167,11 +169,12 @@ object FormField {
}

for {
disposition <- ZIO.fromOption(extract._1).orElseFail(FormDataMissingContentDisposition)
name <- ZIO.fromOption(extract._1.flatMap(_.fields.get("name"))).orElseFail(ContentDispositionMissingName)
charset <- ZIO
.attempt(extract._2.flatMap(x => x.fields.get("charset").map(Charset.forName)).getOrElse(defaultCharset))
.mapError(e => InvalidCharset(e.getMessage))
disposition <- extract._1.toRight(FormDataMissingContentDisposition)
name <- extract._1.flatMap(_.fields.get("name")).toRight(ContentDispositionMissingName)
charset <-
Try {
extract._2.flatMap(x => x.fields.get("charset").map(Charset.forName)).getOrElse(defaultCharset)
}.toEither.left.map(e => InvalidCharset(e.getMessage))
contentParts = extract._4.tail // Skip the first empty line
content = contentParts.foldLeft(Chunk.empty[Byte])(_ ++ _.bytes)
contentType = extract._2
Expand Down
62 changes: 22 additions & 40 deletions zio-http/shared/src/main/scala/zio/http/StreamingForm.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,19 @@ final case class StreamingForm(source: ZStream[Any, Throwable, Byte], boundary:
reader =
source
.mapAccumImmediate(initialState) { (state, byte) =>
def handleBoundary(ast: Chunk[FormAST]): (StreamingForm.State, Option[FormField]) =
if (state.inNonStreamingPart) {
FormField.fromFormAST(ast, charset) match {
case Right(formData) =>
buffer.reset()
(state.reset, Some(formData))
case Left(e) => throw e.asException
}
} else {
buffer.reset()
(state.reset, None)
}

state.formState match {
case formState: FormState.FormStateBuffer =>
val nextFormState = formState.append(byte)
Expand Down Expand Up @@ -94,54 +107,23 @@ final case class StreamingForm(source: ZStream[Any, Throwable, Byte], boundary:
(state, None)
}
case FormState.BoundaryEncapsulated(ast) =>
if (state.inNonStreamingPart) {
runtime.unsafe.run {
FormField
.fromFormAST(ast, charset)
.mapBoth(
_.asException,
{ formData =>
buffer.reset()
(state.reset, Some(formData))
},
)
}.getOrThrowFiberFailure()
} else {
buffer.reset()
(state.reset, None)
}
handleBoundary(ast)
case FormState.BoundaryClosed(ast) =>
if (state.inNonStreamingPart) {
runtime.unsafe.run {
FormField
.fromFormAST(ast, charset)
.mapBoth(
_.asException,
{ formData =>
buffer.reset()
(state.reset, Some(formData))
},
)
}
.getOrThrowFiberFailure()
} else {
buffer.reset()
(state.reset, None)
}
handleBoundary(ast)
}

case _ =>
case _ =>
(state, None)
}
}
.mapZIO { field =>
fieldQueue.offer(Take.single(field))
}
_ <- reader.runDrain.catchAllCause { cause =>
fieldQueue.offer(Take.failCause(cause))
}.ensuring(
fieldQueue.offer(Take.end),
).forkScoped
// FIXME: .blocking here is temporary until we figure out a better way to avoid running effects within mapAccumImmediate
_ <- ZIO
.blocking(reader.runDrain)
.catchAllCause(cause => fieldQueue.offer(Take.failCause(cause)))
.ensuring(fieldQueue.offer(Take.end))
.forkScoped
.interruptible
_ <- Scope.addFinalizerExit { exit =>
// If the fieldStream fails, we need to make sure the reader stream can be interrupted, as it may be blocked
Expand Down
Loading