Add multiple file upload directive #1033

adlawson · 2017-04-14T19:19:24Z

Adds support for multiple file uploads with the same field name.
I've followed the same implementation and testing approach as fileUpload. I did think about changing the fileUpload implementation to take the head of fileUploadAll but it would remove any benefit Sink.headOption gives you over Sink.seq.

// scaladsl
def fileUploadAll(fieldName: String): Directive1[immutable.Seq[(FileInfo, Source[ByteString, Any])]]

// javadsl
def fileUploadAll(fieldName: String, inner: JFunction[JList[JMap.Entry[FileInfo, Source[ByteString, Any]]], Route]): Route

Tasks

Following review at #1033 (review) there are a few outstanding pieces of work that need to be done before merge

Rename directives
Separate destination file creation
Refactor Source(files).map to just files.map
Java Documentation
Scala Documentation
~~Deprecate uploadedFile directive(?)~~ Todo in a following PR

lightbend-cla-validator · 2017-04-14T19:19:25Z

Hi @adlawson,

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

akka-ci · 2017-04-14T19:20:02Z

Can one of the repo owners verify this patch?

adlawson · 2017-04-14T19:20:45Z

akka-http/src/main/scala/akka/http/javadsl/server/directives/FileUploadDirectives.scala

+   * for streaming the file contents somewhere. If there is no such field the request will be rejected.
+   */
+  def fileMultiUpload(fieldName: String, inner: JFunction[JList[JMap.Entry[FileInfo, Source[ByteString, Any]]], Route]): Route = RouteAdapter {
+    def toJava[F <: FileInfo](info: F): FileInfo = info


I'm not super happy about this bit, but without it I'd need to asInstanceOf to make it work with Map.Entry.

jlprat · 2017-04-15T17:38:10Z

OK TO TEST

akka-ci · 2017-04-15T17:47:29Z

Test FAILed.

akka-ci · 2017-04-18T09:12:35Z

Test PASSed.

jrudolph · 2017-04-18T15:41:32Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+          .filter(part ⇒ part.filename.isDefined && part.name == fieldName)
+          .map(part ⇒ (FileInfo(part.name, part.filename.get, part.entity.contentType), part.entity.dataBytes))
+
+        val partF = partSource.runWith(Sink.seq[(FileInfo, Source[ByteString, Any])])


Unfortunately, this approach will not work if the file uploads are big enough. You try to collect ongoing sources for the parts into a seq. However, without reading the first source you will not get the second source, so that the seq will never complete and the stream will deadlock. This is an unfortunate consequence of the squential multipart format. It will work with smaller payloads that fit into all the buffers but if the whole request doesn't fit into the buffers it will deadlock.

I think we need to do the same that other web servers / frameworks are doing and buffer the incoming files on the disk so that we can drain the whole request body before giving out any data.

Yeah that's a real shame. I've been trying to get this to work with larger files today and hitting timeouts, but couldn't trace the root of the problem. Super obvious when you think about it.

If you'd like I can continue working on it and submit updated changes, since it's something I need anyway?

Yes, of course. We'll be glad if you keep working on it. I think buffering to disk isn't completely trivial to do but would definitely be a worthwhile addition.

jrudolph

Needs some solution to the deadlocking problem, commented above. (Just adding the "Changes Requested" marker for administrative reasons)

akka-ci · 2017-04-28T14:19:53Z

Test PASSed.

adlawson · 2017-04-28T14:23:06Z

I've updated the implementation to buffer to disk and squashed the commits into one. The changes now include 2 directives, where the first depends on the second.

def fileMultiUpload(fieldName: String): Directive1[immutable.Seq[(FileInfo, Source[ByteString, Any])]]

def uploadedFiles(fieldName: String): Directive1[immutable.Seq[(FileInfo, File)]]

I had started to go down the route of a custom GraphStage to buffer to disk, just like Alpakka's akka.stream.alpakka.s3.impl.DiskBuffer but it proved harder to implement and more complex than necessary. I also couldn't figure out a way to guarantee reading from a written-to file. Maybe chunking writes may have helped but I still don't see any guarantees.

Another solution I tried was Flow.fromSinkAndSource but it appears the Source doesn't wait for the Sink to fully complete, so this wouldn't work.

The final solution looks too simple to me, but I've tested it with multiple large files and they stream over perfectly fine. I suppose the only downside is that we effectively have to wait for all ByteString sinks to complete before we can begin sourcing again from disk.

patrox · 2017-07-10T13:14:05Z

hi @adlawson - I'm quite interested in the functionality from this PR, so I would like to ask if you are going to finish it ?

@jrudolph Can you please give me some overview of what has to be done in order to get this PR to a mergable state ?

jrudolph

This is looking quite good (even it seems to simple, streams should make things simple ;) ).

Apart from the few comments, it also needs documentation before it can be merged.

jrudolph · 2017-07-17T11:50:35Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+   *
+   * @group fileupload
+   */
+  def uploadedFiles(fieldName: String): Directive1[immutable.Seq[(FileInfo, File)]] =


This needs a name which better describes what it does (maybe storeUploadedFilesToTempFiles? better ideas?). It should have a parameter to specify the directory to put the temporary files into.

jrudolph · 2017-07-17T11:51:45Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+          .filter(part ⇒ part.filename.isDefined && part.name == fieldName)
+          .mapAsync(1) { part ⇒
+            val fileInfo = FileInfo(part.name, part.filename.get, part.entity.contentType)
+            val dest = File.createTempFile("akka-http-upload", ".tmp")


Maybe this part should be completely abstracted away, so that the user can choose the destination freely based on the file info.

jrudolph · 2017-07-17T11:54:20Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+      implicit val ec = ctx.executionContext
+
+      uploadedFiles(fieldName).flatMap { files ⇒
+        val partsSource: Source[(FileInfo, Source[ByteString, Any]), Any] = Source(files).map {


This can be even simpler: no need for Source(files) when source and target datastructures are a Seq. You can just map the files datastructure.

jrudolph · 2017-07-17T11:55:13Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+   *
+   * @group fileupload
+   */
+  def fileMultiUpload(fieldName: String): Directive1[immutable.Seq[(FileInfo, Source[ByteString, Any])]] =


Should be marked as @ApiMayChange.

jrudolph · 2017-07-17T11:55:18Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+   *
+   * @group fileupload
+   */
+  def uploadedFiles(fieldName: String): Directive1[immutable.Seq[(FileInfo, File)]] =


Should be marked as @ApiMayChange.

jrudolph · 2017-07-17T11:56:56Z

(And sorry for keeping it around for so long. It's not because the PR wasn't good, we were just busy.)

adlawson · 2017-07-18T15:44:17Z

Thanks for the feedback. I stuck with similar names to the single file uploads, just for consistency, but I'm happy to change them to be more descriptive. I'll also get onto your other points like separating temp file construction etc. too.

Don't worry about the delay - I've been buried under a mountain of work too. I'll try to get this updated within the next week.

adlawson · 2017-07-30T23:15:51Z

Hey @jrudolph, I've made some updates since your last review

Renamed uploadedFiles to storeUploadedFiles to better describe what it does
Separated out the temporary file creation
Added @ApiMayChange where necessary, and removed the mima-filter exclusions

I've also taken a couple of further steps

Renamed fileMultiUpload to fileUploadAll so "multi" isn't confused with "multipart"
Added a storeUploadedFile (singular) to have a consistent API with the new directive. This should replace uploadedFile (which still exists and behaves the same as before, but could be deprecated in favour of this new function)

Let me know what you think, and whether you agree with storeUploadedFile or not.

akka-ci · 2017-09-09T17:47:44Z

Test PASSed.

- Add fileUploadAll directive, buffering multiple files on disk - Add storeUploadedFiles directive to stream multiple files to disk - Add mima-filter exclusions for new directives - Add destFn argument to directives to support custom destinations - Replace uploadedFile directive with storeUploadedFile

- Document fileUploadAll directive - Document storeUploadedFile directive - Document storeUploadedFiles directive

akka-ci · 2017-09-10T23:12:37Z

Test FAILed.

akka-ci · 2017-09-10T23:17:57Z

Test PASSed.

adlawson · 2017-09-10T23:29:09Z

I've added the discardBytes call you asked for, but I haven't been able to come up with a working test for it. The closest I had was observing completing promises when the streams are consumed, just like the tests in akka/http/scaladsl/model/EntityDiscardingSpec.scala#L44-L53, but the futures never complete. I also tried simply consuming streams again to trigger and catch an IllegalStateException but I was able to consume them all again fine, even the ones consumed to files. If you have any pointers I'll keep trying, but I'm at a loss right now.

In addition to the change, I've squashed the commits into more reasonable, safe, chunks of work.

storeUploadedFile has a more descriptive name and also adds the possibility to select the target file.

Directive1[T] = Directive[Tuple1[T]], so don't nest another Tuple2 inside. This also allows to get rid of the `case` syntax when using the directive. Slightly improved implementation as well .

Before only Strict BodyParts/HttpEntity were tested which are not susceptible to deadlocks. The new version should test the more common streaming uploads more comprehensively.

jrudolph · 2017-09-18T12:42:51Z

@adlawson looking very good now. I added a few commits with smallish cleanups and the deprecation of uploadedFile. Could you review and merge https://github.com/adlawson/akka-http/pull/1 into this PR (couldn't push to your branch)?

akka-ci · 2017-09-19T08:42:51Z

Test PASSed.

adlawson · 2017-09-19T09:08:09Z

Changes look good, @jrudolph. Thanks for the updates.
All merged and ready to go 👍

ktoso · 2017-09-19T09:08:44Z

Great, will give it a final read then :)

ktoso

LGTM, great effort, thanks a lot 👍

ktoso · 2017-09-19T09:12:23Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+   * @group fileupload
+   */
+  @ApiMayChange
+  def storeUploadedFile(fieldName: String, destFn: FileInfo ⇒ File): Directive[(FileInfo, File)] =


nice name btw 👍

Docs added and Johanness's PR also merged, everything looks good

ktoso · 2017-09-19T09:35:18Z

Resolved #1273 Thanks!

DanLin123 · 2017-10-06T23:26:10Z

akka-http/src/main/scala/akka/http/scaladsl/server/directives/FileUploadDirectives.scala

+  /**
+   * Streams the bytes of the file submitted using multipart with the given field name into designated files on disk.
+   * If there is an error writing to disk the request will be failed with the thrown exception, if there is no such
+   * field the request will be rejected. Stored files are cleaned up on exit but not on failure.


Does storeUploadedFiles Directive reject request if field is empty?
It seems there is no implementation of it

adlawson commented Apr 14, 2017

View reviewed changes

akka-ci added validating PR that is currently being validated by Jenkins needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed validating PR that is currently being validated by Jenkins labels Apr 15, 2017

akka-ci added validating PR that is currently being validated by Jenkins tested PR that was successfully built and tested by Jenkins and removed needs-attention Indicates a PR validation failure (set by CI infrastructure) validating PR that is currently being validated by Jenkins labels Apr 18, 2017

adlawson changed the title ~~[WIP] Add fileMultiUpload directive~~ Add fileMultiUpload directive Apr 18, 2017

jrudolph reviewed Apr 18, 2017

View reviewed changes

jrudolph suggested changes Apr 26, 2017

View reviewed changes

akka-ci added validating PR that is currently being validated by Jenkins tested PR that was successfully built and tested by Jenkins and removed tested PR that was successfully built and tested by Jenkins validating PR that is currently being validated by Jenkins labels Apr 28, 2017

jrudolph mentioned this pull request Jul 10, 2017

fileUpload & uploadedFile directives don't support multiple files upload #1273

Closed

jrudolph previously requested changes Jul 17, 2017

View reviewed changes

akka-ci added validating PR that is currently being validated by Jenkins and removed tested PR that was successfully built and tested by Jenkins labels Jul 30, 2017

akka-ci added validating PR that is currently being validated by Jenkins and removed tested PR that was successfully built and tested by Jenkins labels Sep 10, 2017

adlawson added 2 commits September 11, 2017 00:06

Add documentation for new file upload directives

850f7cc

- Document fileUploadAll directive - Document storeUploadedFile directive - Document storeUploadedFiles directive

akka-ci added validating PR that is currently being validated by Jenkins needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed validating PR that is currently being validated by Jenkins labels Sep 10, 2017

akka-ci added tested PR that was successfully built and tested by Jenkins and removed needs-attention Indicates a PR validation failure (set by CI infrastructure) labels Sep 10, 2017

jrudolph added 3 commits September 18, 2017 14:00

!htp deprecate uploadedFile directive in favor of new storeUploadedFile

3832032

storeUploadedFile has a more descriptive name and also adds the possibility to select the target file.

=htp slight improvement of storeUploadedFile signature

0a0976d

Directive1[T] = Directive[Tuple1[T]], so don't nest another Tuple2 inside. This also allows to get rid of the `case` syntax when using the directive. Slightly improved implementation as well .

=htp add additional streaming tests for new FileUploadDirectives

7126f29

Before only Strict BodyParts/HttpEntity were tested which are not susceptible to deadlocks. The new version should test the more common streaming uploads more comprehensively.

akka-ci added validating PR that is currently being validated by Jenkins tested PR that was successfully built and tested by Jenkins and removed tested PR that was successfully built and tested by Jenkins validating PR that is currently being validated by Jenkins labels Sep 19, 2017

ktoso approved these changes Sep 19, 2017

View reviewed changes

ktoso merged commit a1a9c27 into akka:master Sep 19, 2017

adlawson deleted the adlawson/directive-file-multiupload branch September 19, 2017 09:43

DanLin123 reviewed Oct 6, 2017

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add multiple file upload directive #1033

Add multiple file upload directive #1033

adlawson commented Apr 14, 2017 •

edited

lightbend-cla-validator commented Apr 14, 2017

akka-ci commented Apr 14, 2017

adlawson Apr 14, 2017

jlprat commented Apr 15, 2017

akka-ci commented Apr 15, 2017

akka-ci commented Apr 18, 2017

jrudolph Apr 18, 2017

adlawson Apr 18, 2017

jrudolph Apr 18, 2017

jrudolph left a comment

akka-ci commented Apr 28, 2017

adlawson commented Apr 28, 2017 •

edited

patrox commented Jul 10, 2017

jrudolph left a comment

jrudolph Jul 17, 2017

jrudolph Jul 17, 2017

jrudolph Jul 17, 2017

jrudolph Jul 17, 2017

jrudolph Jul 17, 2017

jrudolph commented Jul 17, 2017

adlawson commented Jul 18, 2017

adlawson commented Jul 30, 2017

akka-ci commented Sep 9, 2017

akka-ci commented Sep 10, 2017

akka-ci commented Sep 10, 2017

adlawson commented Sep 10, 2017 •

edited

jrudolph commented Sep 18, 2017

akka-ci commented Sep 19, 2017

adlawson commented Sep 19, 2017

ktoso commented Sep 19, 2017

ktoso left a comment

ktoso Sep 19, 2017

ktoso commented Sep 19, 2017

DanLin123 Oct 6, 2017

Add multiple file upload directive #1033

Add multiple file upload directive #1033

Conversation

adlawson commented Apr 14, 2017 • edited

Tasks

lightbend-cla-validator commented Apr 14, 2017

akka-ci commented Apr 14, 2017

Choose a reason for hiding this comment

jlprat commented Apr 15, 2017

akka-ci commented Apr 15, 2017

akka-ci commented Apr 18, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrudolph left a comment

Choose a reason for hiding this comment

akka-ci commented Apr 28, 2017

adlawson commented Apr 28, 2017 • edited

patrox commented Jul 10, 2017

jrudolph left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrudolph commented Jul 17, 2017

adlawson commented Jul 18, 2017

adlawson commented Jul 30, 2017

akka-ci commented Sep 9, 2017

akka-ci commented Sep 10, 2017

akka-ci commented Sep 10, 2017

adlawson commented Sep 10, 2017 • edited

jrudolph commented Sep 18, 2017

akka-ci commented Sep 19, 2017

adlawson commented Sep 19, 2017

ktoso commented Sep 19, 2017

ktoso left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ktoso commented Sep 19, 2017

Choose a reason for hiding this comment

adlawson commented Apr 14, 2017 •

edited

adlawson commented Apr 28, 2017 •

edited

adlawson commented Sep 10, 2017 •

edited