support multipart body with streaming content #9852

[smadarasmi]

This PR is not finished. I'm hoping to get some early feedback that I am on the right track before I continue and also to get some questions answered. I will leave the questions in line.

Things left to do aside from the comments I left in this PR:

1. Add to Scala and Java documentation for uploading files
2. Integration test

Pull Request Checklist

• Have you read How to write the perfect pull request?
• Have you read through the contributor guidelines?
• Have you signed the Lightbend CLA?
• Have you referenced any issues you're fixing using commit message keywords?
• Have you added copyright headers to new files?
• Have you checked that both Scala and Java APIs are updated?
• Have you updated the documentation for both Scala and Java sections?
• Have you added tests for any changed functionality?

Helpful things

Fixes

Fixes #xxxx

Purpose

What does this PR do?

Background Context

Why did you take this approach?

References

Are there any relevant issues / PRs / mailing lists discussions?

[lightbend-cla-validator]

Hi @smadarasmi,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Lightbend Contributors License Agreement:

http://www.lightbend.com/contribute/cla

[smadarasmi]

I had to make this fanout, otherwise my unit test was failing with

Right(MultipartFormData(HashMap(),Vector(FilePart(file1,file1.txt,Some(text/plain),Source(SourceShape(PublisherSource.out(1840397757))),-1,form-data), FilePart(file2,file2.txt,Some(text/plain),Source(SourceShape(PublisherSource.out(276528415))),-1,form-data)),Vector())) is Right but java.lang.IllegalStateException: Sink.asPublisher(fanout = false) only supports one subscriber (which is allowed, see reactive-streams specification, rule 1.12)
java.lang.Exception: Right(MultipartFormData(HashMap(),Vector(FilePart(file1,file1.txt,Some(text/plain),Source(SourceShape(PublisherSource.out(1840397757))),-1,form-data), FilePart(file2,file2.txt,Some(text/plain),Source(SourceShape(PublisherSource.out(276528415))),-1,form-data)),Vector())) is Right but java.lang.IllegalStateException: Sink.asPublisher(fanout = false) only supports one subscriber (which is allowed, see reactive-streams specification, rule 1.12)
	at org.specs2.matcher.MatchResultStackTrace.setStacktrace(Expectations.scala:57)
	at org.specs2.matcher.MatchResultStackTrace.setStacktrace$(Expectations.scala:55)
	at play.api.mvc.MultipartBodyParserSpec.setStacktrace(MultipartBodyParserSpec.scala:21)

I'm not sure where the other subscriber is coming from.

[arteme]

The error message Sink.asPublisher(fanout = false) only supports one subscriber is not correct, see: akka/akka#30908 In fact, if you say fanout = false, the subscription times out and the created source is cancelled.

[smadarasmi]

Still need to figure out how to determine the file size since this is streaming content. Or could we just leave it as -1?

[smadarasmi]

I had to add akka-stream-testkit for this test: https://github.com/playframework/playframework/pull/10309/files#diff-353e7858dc8605c961746609570dc2c9R65

When I constructed the test with a Source and running through my parser, which returns FilePartHandler[Source[ByteString, NotUsed]] and I collect and do assertion on that, I get an error similar to this:

Processor actor [Actor[akka://test-server/system/Materializers/StreamSupervisor-1/flow-0-1-fanoutPublisherSink#622867610]] terminated abruptly.

Any ideas what's causing this? I suspect this is an issue with the way my test was constructed.

[smadarasmi]

@renatocaval Hey, this is regarding #9852. It's still a WIP and a lot of work left to do but I was hoping to get some help and see if I am on the right track.
Look forward to your reply :)