src/main/java/edu/ucla/library/bucketeer/verticles/S3BucketVerticle.java
@SuppressWarnings("Indentation") // Checkstyle's indentation check doesn't work with multiple lambdas
    private void upload(final Message<JsonObject> aMessage, final JsonObject aConfig) {
        final boolean deletable = aMessage.headers().contains(Constants.DERIVATIVE_IMAGE);
        final JsonObject storageRequest = aMessage.body();

        // If an S3 bucket isn't being supplied to us, use the one in our application configuration
        if (!storageRequest.containsKey(Config.S3_BUCKET)) {
            storageRequest.mergeIn(aConfig);
        }

        final String filePath = storageRequest.getString(Constants.FILE_PATH);
        final String jobName = storageRequest.getString(Constants.JOB_NAME);
        final String s3Bucket = storageRequest.getString(Config.S3_BUCKET);
        final String imageID = storageRequest.getString(Constants.IMAGE_ID);
        final String imageIDSansExt = FileUtils.stripExt(imageID);

        LOGGER.debug(MessageCodes.BUCKETEER_010, imageID, filePath, s3Bucket);

        vertx.fileSystem().open(filePath, new OpenOptions().setWrite(false), open -> {
            if (open.succeeded()) {
                final AsyncFile asyncFile = open.result();
                final UserMetadata metadata = new UserMetadata(Constants.IMAGE_ID, imageIDSansExt);

                // If there is a job name, supply it in user metadata on S3 upload
                if (StringUtils.trimToNull(jobName) != null) {
                    metadata.add(Constants.JOB_NAME, jobName);
                }

                LOGGER.debug(MessageCodes.BUCKETEER_044, imageID, filePath, s3Bucket);

                // If our connection pool is full, drop back and try resubmitting the request
                try {
                    myS3Client.put(s3Bucket, imageID, asyncFile, metadata, response -> {
                        final int statusCode = response.statusCode();

                        response.exceptionHandler(exception -> {
                            final String details = exception.getMessage();

                            LOGGER.error(exception, details);

                            closeFile(asyncFile, filePath, false);
                            sendReply(aMessage, CodeUtils.getInt(MessageCodes.BUCKETEER_500), details);
                        });

                        // If we get a successful upload response code, note this in our final results map
                        if (statusCode == HTTP.OK) {
                            LOGGER.info(MessageCodes.BUCKETEER_026, imageID);

                            vertx.sharedData().getLocalMap(Constants.RESULTS_MAP).put(imageIDSansExt, true);

                            // Close the file, send the success result, and decrement the S3 request counter
                            closeFile(asyncFile, filePath, deletable);
                            sendReply(aMessage, 0, Op.SUCCESS);
                        } else {
                            LOGGER.error(MessageCodes.BUCKETEER_014, statusCode, response.statusMessage());

                            // Log the detailed reason we failed so we can track down the issue
                            response.bodyHandler(body -> {
                                LOGGER.error(MessageCodes.BUCKETEER_500, body.getString(0, body.length()));
                            });

                            // If there is some internal S3 server error, let's try again
                            if (statusCode == HTTP.INTERNAL_SERVER_ERROR) {
                                closeFile(asyncFile, filePath, false);
                                sendReply(aMessage, 0, Op.RETRY);
                            } else {
                                final String errorMessage = statusCode + " - " + response.statusMessage();

                                LOGGER.error(MessageCodes.BUCKETEER_156, errorMessage);
                                closeFile(asyncFile, filePath, false);
                                retryUpload(imageID, aMessage);
                            }
                        }
                    }, exception -> {
                        LOGGER.error(MessageCodes.BUCKETEER_156, exception.getMessage());
                        closeFile(asyncFile, filePath, false);
                        retryUpload(imageID, aMessage);
                    });
                } catch (final ConnectionPoolTooBusyException details) {
                    LOGGER.debug(MessageCodes.BUCKETEER_046, imageID);
                    closeFile(asyncFile, filePath, false);
                    sendReply(aMessage, 0, Op.RETRY);
                }
            } else {
                LOGGER.error(open.cause(), LOGGER.getMessage(MessageCodes.BUCKETEER_043, filePath));
                sendReply(aMessage, CodeUtils.getInt(MessageCodes.BUCKETEER_043), filePath);
            }
        });
    }
/**
     * Upload the requested file to S3.
     *
     * @param aMessage The message containing the S3 upload request
     * @param aConfig The verticle's configuration
     */
Upload the requested file to S3.
@param aMessage The message containing the S3 upload request
@param aConfig The verticle's configuration
Upload the requested file to S3 . @param aMessage The message containing the S3 upload request @param aConfig The verticle ' s configuration
@SuppressWarnings("Indentation") 
    private void upload(final Message<JsonObject> aMessage, final JsonObject aConfig) {
        final boolean deletable = aMessage.headers().contains(Constants.DERIVATIVE_IMAGE);
        final JsonObject storageRequest = aMessage.body();
        if (!storageRequest.containsKey(Config.S3_BUCKET)) {
            storageRequest.mergeIn(aConfig);
        }
        final String filePath = storageRequest.getString(Constants.FILE_PATH);
        final String jobName = storageRequest.getString(Constants.JOB_NAME);
        final String s3Bucket = storageRequest.getString(Config.S3_BUCKET);
        final String imageID = storageRequest.getString(Constants.IMAGE_ID);
        final String imageIDSansExt = FileUtils.stripExt(imageID);
        LOGGER.debug(MessageCodes.BUCKETEER_010, imageID, filePath, s3Bucket);
        vertx.fileSystem().open(filePath, new OpenOptions().setWrite(false), open -> {
            if (open.succeeded()) {
                final AsyncFile asyncFile = open.result();
                final UserMetadata metadata = new UserMetadata(Constants.IMAGE_ID, imageIDSansExt);
                if (StringUtils.trimToNull(jobName) != null) {
                    metadata.add(Constants.JOB_NAME, jobName);
                }
                LOGGER.debug(MessageCodes.BUCKETEER_044, imageID, filePath, s3Bucket);
                try {
                    myS3Client.put(s3Bucket, imageID, asyncFile, metadata, response -> {
                        final int statusCode = response.statusCode();
                        response.exceptionHandler(exception -> {
                            final String details = exception.getMessage();
                            LOGGER.error(exception, details);
                            closeFile(asyncFile, filePath, false);
                            sendReply(aMessage, CodeUtils.getInt(MessageCodes.BUCKETEER_500), details);
                        });
                        if (statusCode == HTTP.OK) {
                            LOGGER.info(MessageCodes.BUCKETEER_026, imageID);
                            vertx.sharedData().getLocalMap(Constants.RESULTS_MAP).put(imageIDSansExt, true);
                            closeFile(asyncFile, filePath, deletable);
                            sendReply(aMessage, 0, Op.SUCCESS);
                        } else {
                            LOGGER.error(MessageCodes.BUCKETEER_014, statusCode, response.statusMessage());
                            response.bodyHandler(body -> {
                                LOGGER.error(MessageCodes.BUCKETEER_500, body.getString(0, body.length()));
                            });
                            if (statusCode == HTTP.INTERNAL_SERVER_ERROR) {
                                closeFile(asyncFile, filePath, false);
                                sendReply(aMessage, 0, Op.RETRY);
                            } else {
                                final String errorMessage = statusCode + " - " + response.statusMessage();
                                LOGGER.error(MessageCodes.BUCKETEER_156, errorMessage);
                                closeFile(asyncFile, filePath, false);
                                retryUpload(imageID, aMessage);
                            }
                        }
                    }, exception -> {
                        LOGGER.error(MessageCodes.BUCKETEER_156, exception.getMessage());
                        closeFile(asyncFile, filePath, false);
                        retryUpload(imageID, aMessage);
                    });
                } catch (final ConnectionPoolTooBusyException details) {
                    LOGGER.debug(MessageCodes.BUCKETEER_046, imageID);
                    closeFile(asyncFile, filePath, false);
                    sendReply(aMessage, 0, Op.RETRY);
                }
            } else {
                LOGGER.error(open.cause(), LOGGER.getMessage(MessageCodes.BUCKETEER_043, filePath));
                sendReply(aMessage, CodeUtils.getInt(MessageCodes.BUCKETEER_043), filePath);
            }
        });
    }
@ SuppressWarnings ( "Indentation" ) private void upload ( final Message < JsonObject > aMessage , final JsonObject aConfig ) { final boolean deletable = aMessage . headers ( ) . contains ( Constants . DERIVATIVE_IMAGE ) ; final JsonObject storageRequest = aMessage . body ( ) ; if ( ! storageRequest . containsKey ( Config . S3_BUCKET ) ) { storageRequest . mergeIn ( aConfig ) ; } final String filePath = storageRequest . getString ( Constants . FILE_PATH ) ; final String jobName = storageRequest . getString ( Constants . JOB_NAME ) ; final String s3Bucket = storageRequest . getString ( Config . S3_BUCKET ) ; final String imageID = storageRequest . getString ( Constants . IMAGE_ID ) ; final String imageIDSansExt = FileUtils . stripExt ( imageID ) ; LOGGER . debug ( MessageCodes . BUCKETEER_010 , imageID , filePath , s3Bucket ) ; vertx . fileSystem ( ) . open ( filePath , new OpenOptions ( ) . setWrite ( false ) , open -> { if ( open . succeeded ( ) ) { final AsyncFile asyncFile = open . result ( ) ; final UserMetadata metadata = new UserMetadata ( Constants . IMAGE_ID , imageIDSansExt ) ; if ( StringUtils . trimToNull ( jobName ) != null ) { metadata . add ( Constants . JOB_NAME , jobName ) ; } LOGGER . debug ( MessageCodes . BUCKETEER_044 , imageID , filePath , s3Bucket ) ; try { myS3Client . put ( s3Bucket , imageID , asyncFile , metadata , response -> { final int statusCode = response . statusCode ( ) ; response . exceptionHandler ( exception -> { final String details = exception . getMessage ( ) ; LOGGER . error ( exception , details ) ; closeFile ( asyncFile , filePath , false ) ; sendReply ( aMessage , CodeUtils . getInt ( MessageCodes . BUCKETEER_500 ) , details ) ; } ) ; if ( statusCode == HTTP . OK ) { LOGGER . info ( MessageCodes . BUCKETEER_026 , imageID ) ; vertx . sharedData ( ) . getLocalMap ( Constants . RESULTS_MAP ) . put ( imageIDSansExt , true ) ; closeFile ( asyncFile , filePath , deletable ) ; sendReply ( aMessage , 0 , Op . SUCCESS ) ; } else { LOGGER . error ( MessageCodes . BUCKETEER_014 , statusCode , response . statusMessage ( ) ) ; response . bodyHandler ( body -> { LOGGER . error ( MessageCodes . BUCKETEER_500 , body . getString ( 0 , body . length ( ) ) ) ; } ) ; if ( statusCode == HTTP . INTERNAL_SERVER_ERROR ) { closeFile ( asyncFile , filePath , false ) ; sendReply ( aMessage , 0 , Op . RETRY ) ; } else { final String errorMessage = statusCode + " - " + response . statusMessage ( ) ; LOGGER . error ( MessageCodes . BUCKETEER_156 , errorMessage ) ; closeFile ( asyncFile , filePath , false ) ; retryUpload ( imageID , aMessage ) ; } } } , exception -> { LOGGER . error ( MessageCodes . BUCKETEER_156 , exception . getMessage ( ) ) ; closeFile ( asyncFile , filePath , false ) ; retryUpload ( imageID , aMessage ) ; } ) ; } catch ( final ConnectionPoolTooBusyException details ) { LOGGER . debug ( MessageCodes . BUCKETEER_046 , imageID ) ; closeFile ( asyncFile , filePath , false ) ; sendReply ( aMessage , 0 , Op . RETRY ) ; } } else { LOGGER . error ( open . cause ( ) , LOGGER . getMessage ( MessageCodes . BUCKETEER_043 , filePath ) ) ; sendReply ( aMessage , CodeUtils . getInt ( MessageCodes . BUCKETEER_043 ) , filePath ) ; } } ) ; }
Upload the requested file to S3.
Upload the requested file to S3 .
// Checkstyle's indentation check doesn't work with multiple lambdas // If an S3 bucket isn't being supplied to us, use the one in our application configuration // If there is a job name, supply it in user metadata on S3 upload // If our connection pool is full, drop back and try resubmitting the request // If we get a successful upload response code, note this in our final results map // Close the file, send the success result, and decrement the S3 request counter // Log the detailed reason we failed so we can track down the issue // If there is some internal S3 server error, let's try again