Trying to use form field and file upload directive together in Akka-Http?

Table of contents

Reading Time: 2 minutes

If you are reading this then probably you have encountered the issue that comes while accessing fields with uploading a file, and if not then let me first tell you what’s the situation that I’m talking about here,

Let’s see the code where we try to use fileUpload directive and formFields together

path ("uploadFile") {
        post {
          fileUpload("file") {
            (fileInfo: FileInfo, file: File) =>
              formFields ('pictureName.as[String] ) {
                pictureName =>
                  println (s"pictureName is: $pictureName")
                  complete ("file uploaded")
              }
          }
        }
      }

So, the above code seems fine right? and you are wondering what is the problem here?
The problem is that if you’ll try to use it, it will throw an exception –
Substream Source cannot be materialized more than once, as we are trying to unmarshal the entity two times.

Note: It can work to upload a small file (<1mb) if the entity is small enough that it fits in the internal buffer.

I encountered this issue while working on some project as fortunately I tested my code with a large file and found the issue before it became big.

Some workarounds over this that I found are –

Use toStrict
We can use toStrict to make sure that entire entity is buffered in memory, as otherwise it will be drained by the first unmarshalling process and not available for the second.

path("uploadPicture") {
      (post & withSizeLimit(10 * 1024 * 1024) & toStrictEntity(60.seconds)) {
        fileUpload("file") { (fileInfo: FileInfo, file: File) =>
          formFields('filename.as[String]) { pictureName =>
            println (s"pictureName is: $pictureName")
            complete ("file uploaded")
          }
       }
    }

Custom Implementation to extract file and form fields

path("uploadPicture") {
        (post & withoutSizeLimit) {
          extractRequestContext { ctx =>
            import ctx.{ executionContext, materializer }
            entity(as[Multipart.FormData]) { data =>

              val extractedData: Future[Map[String, Any]] = data.parts.mapAsyncUnordered(1) {
                case picture: BodyPart if picture.name == "picture" =>
                  val extension = picture.entity.getContentType().toString.split("/")
                  val tempFile: File = File.createTempFile("picture", "." + extension.apply(1))
                  picture.entity.dataBytes.runWith(FileIO.toPath(tempFile.toPath)).map { _ =>
                    "picture" -> tempFile
                  }
                case expectedPart if expectedPart.name == "pictureName" =>
                  expectedPart.toStrict(2.seconds).map(field => Some(field.name -> field.entity.data.utf8String))

              }.runFold(Map.empty[String, Any])((map, tuple) => map + tuple)
              complete(
                extractedData.map { data =>
                  // do some processing and send response
                }.recover {
                  // handle exception in case of missing fields, or any other exception that can occure
                }
              )
            }
          }
        }
      }

In above code, I’m explicitly storing the file in memory, but here we can do any processing according to our use case like If I want to upload that stream directly to S3 rather than storing it to memory.
Using the above mechanism we are directly accessing fields, instead of using any directive.

So, Both of the above solutions can be used depending on the use case that you encountered.

I hope this was helpful to those who were stuck in the similar situation, and I was able to provide them with some input so that they can proceed further.