Play Framework 2.2 Java Iteratee - Reactive Upload

Question

Is it possible to use Play's Iteratees from within Java? I have not been able to find any examples nor doco on using Iteratee in Java, only Scala. I'm guessing getting Iteratees working in Java with the PLay API is a little more messy code wise (lots of anon Funtion1<?,>s)...

If it is possible I would like to create an App controller that can accept multi-part file uploads uploaded via HTTPs chunked transfer encoding and parse these message chunks downstream to an S3 store. Any ideas on how I can approach this in Java?

Cheers.

Answer 1

I think it is possible to implement iteratees in Java.

There is an example of doing this in Scala written by Sadache in this question: Play 2.x : Reactive file upload with Iteratees

Note though that there is no async api library available for S3, so you will be blocking in that end of the upload if you for example use the official amazon api java libraries.

Answer 2

The Java SDK contains the class TransferManager to perform asynchronous uploads. It contains an own configurable ThreadPool.

Iteratees written in Java might be able to push the bytes of the uploaded file directly to S3 but the code would looking akward and hard to configure. For a lot of use cases it would be good enough to stream a file from the browser to a temporary file (so it is not fully in memory) and then stream it to S3.

I created an example project on Github how to do that:

package controllers;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.event.ProgressEvent;
import com.amazonaws.event.ProgressListener;
import com.amazonaws.services.s3.transfer.TransferManager;
import com.amazonaws.services.s3.transfer.Upload;
import com.amazonaws.services.s3.transfer.model.UploadResult;
import play.*;
import play.libs.F.Function;
import play.libs.F.Promise;
import play.mvc.*;
import views.html.index;
import scala.concurrent.Promise$;

public class Application extends Controller {
    //don't forget to tm.shutdownNow() on application termination
    //you can configure a Thread pool http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/transfer/TransferManager.html#TransferManager(com.amazonaws.services.s3.AmazonS3, java.util.concurrent.ThreadPoolExecutor)
    private static TransferManager tm;
    private static final String BUCKET_NAME = "your-bucket";

    //this is bad style, use a plugin!
    static {
        final String accessKey = System.getenv("AWS_ACCESS");
        final String secretKey = System.getenv("AWS_SECRET");
        final AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretKey);
        tm = new TransferManager(credentials);
    }

    /** shows a form to upload a file */
    public static Result index() {
        return ok(index.render("Your new application is ready."));
    }

    /** uploads a file to Amazon S3. */
    public static Promise<Result> upload() {
        final Http.MultipartFormData.FilePart meta = request().body().asMultipartFormData().getFile("picture");
        final String key = meta.getFilename();
        final Upload upload = tm.upload(BUCKET_NAME, key, meta.getFile());
        Logger.info("start upload " + meta.getFilename());
        return asPromise(meta.getFilename(), upload).map(new Function<UploadResult, Result>() {
            @Override
            public Result apply(UploadResult uploadResult) throws Throwable {
                Logger.info("finished " + meta.getFilename());
                return ok(asString(uploadResult));
            }
        });
    }

    private static String asString(UploadResult result) {
        return "UploadResult{bucketName=" + result.getBucketName() + ", key=" + result.getKey() + ", version=" + result.getVersionId() + ", ETag=" + result.getETag() + "}";
    }

    private static Promise<UploadResult> asPromise(final String filename, final Upload upload) {
        final scala.concurrent.Promise<UploadResult> scalaPromise = Promise$.MODULE$.apply();
        upload.addProgressListener(new ProgressListener() {
            @Override
            public void progressChanged(ProgressEvent progressEvent) {
                if (progressEvent.getEventCode() == ProgressEvent.CANCELED_EVENT_CODE) {
                    scalaPromise.failure(new RuntimeException("canceled " + filename));
                } else if (progressEvent.getEventCode() == ProgressEvent.FAILED_EVENT_CODE) {
                    scalaPromise.failure(new RuntimeException("failed " + filename));
                } else if(progressEvent.getEventCode() == ProgressEvent.COMPLETED_EVENT_CODE) {
                    Logger.info("done " + filename);
                    try {
                        scalaPromise.success(upload.waitForUploadResult());
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        });
        return Promise.wrap(scalaPromise.future());
    }

}

An example log output for uploading two files in different browser windows:

[info] play - play-internal-execution-context-1 Application started (Dev)
[info] application - play-akka.actor.default-dispatcher-3 start upload file5.ext
[info] application - play-akka.actor.default-dispatcher-2 start upload file1.ext
[info] application - java-sdk-progress-listener-callback-thread done file1.ext
[info] application - play-akka.actor.default-dispatcher-5 finished file1.ext
[info] application - java-sdk-progress-listener-callback-thread done file5.ext
[info] application - play-akka.actor.default-dispatcher-5 finished file5.ext