AWS Lambda byte streaming client in Java

Question

I am trying to write a byte streaming client in Java for an AWS Lambda function. I have created the Lambda function as an implementation of a RequestStreamHandler. The basis for this project is outlined in the docs here.

public class LambdaFunctionHandler implements RequestStreamHandler {
    static final String bucket = "anS3bucket";
    static final String key = "anS3KeyToAJpegFile";

    @Override
    public void  handleRequest(InputStream input, OutputStream output, 
            Context context) throws IOException {

        AmazonS3 s3Client = new AmazonS3Client(
                new EnvironmentVariableCredentialsProvider());
        try {
            context.getLogger().log("Downloading an object\n");
            S3Object s3object = s3Client.getObject(new GetObjectRequest(
                    bucket, key));
            context.getLogger().log("Content-Type: "  + 
                    s3object.getObjectMetadata().getContentType() 
                       + "\n");
            InputStream in = s3object.getObjectContent();
            int b = 0;
            context.getLogger().log("Writing jpeg on output\n");
            while ((b = in.read()) > -1) {
                output.write(b);
            }

        } catch (AmazonServiceException e) {
            System.out.println("Error Message: " + e.getMessage());
        }   
    }
}

This hardcoded example works fine on the Lambda test console. I can upload the JAR and run the lambda function (by clicking on "Test"). What the function does is that it retrives a jpeg file contents and writes the byte stream on the OutputStream. I can see the binary output in the test console as the function result. So far all is fine. Eventually I will run ImageMagick on the jpeg and resize it - that is the goal of this project.

My client code looks like:

public interface ImageService {
    @LambdaFunction(functionName="ImageProcessing")
    OutputStream getImageStream(InputStream data);
}

public class LambdaImageTest {

public static void main(String[] args) throws IOException {
    AWSLambdaClient lambda = new AWSLambdaClient(new  ProfileCredentialsProvider());
    lambda.configureRegion(Regions.EU_WEST_1);

    ImageService service = LambdaInvokerFactory.build(ImageService.class, lambda);

    // Call lambda function, receive byte stream
    OutputStream  out = service.getImageStream(null); 
    System.out.println(out); // This code is not complete
}

When I try to recieve the byte stream in a Java client, I fail. There does not seem to be a way to recieve the byte stream. The client seems to try to read the reault as a json data, which is not what I want here. I want to read the byte stream directly (the jpeg binary contents). The error I get is:

Exception in thread "main" com.amazonaws.services.lambda.invoke.LambdaSerializationException: Failed to parse Lambda function result
    at com.amazonaws.services.lambda.invoke.LambdaInvokerFactory$LambdaInvocationHandler.getObjectFromPayload(LambdaInvokerFactory.java:210)
    at com.amazonaws.services.lambda.invoke.LambdaInvokerFactory$LambdaInvocationHandler.processInvokeResult(LambdaInvokerFactory.java:189)
    at com.amazonaws.services.lambda.invoke.LambdaInvokerFactory$LambdaInvocationHandler.invoke(LambdaInvokerFactory.java:106)
    at com.sun.proxy.$Proxy3.getImageStream(Unknown Source)
    at se.devo.lambda.image.LambdaImageTest.main(LambdaImageTest.java:33)
Caused by: com.fasterxml.jackson.core.JsonParseException: Invalid UTF-8 middle byte 0xff
 at [Source: [B@42257bdd; line: 1, column: 4]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1487)

How do I properly receive byte stream data in an AWS Lambda java client?

Answer 1

I have found a solution. The LambdaInvokerFactory class will ALWAYS handle request and response data as JSON and therefore serialize and deserialize which is the problem. The source code holds the clue to the answer though, and I have ripped out the part that does the invocation of the lambda function, but I bypass the JSON deserialization and access the payload directly. Simple, but it really should have been in the LambdaInvokerFactory class already...

Here is my fully working solution. The Lambda function code:

public class LambdaFunctionHandler implements RequestStreamHandler  {
    public void handleRequest(InputStream input, OutputStream output, Context context) throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        AmazonS3 s3Client = new AmazonS3Client(
                new EnvironmentVariableCredentialsProvider());
        try {

            // Need to deserialize JSON data ourselves in Lambda streaming mode
            String data = getJSONInputStream(input);
            context.getLogger().log("JSON data:\n'" + data + "'\n");            
            context.getLogger().log("Deserialize JSON data to object\n");
            ImageRequest request  = mapper.readValue(data, ImageRequest.class);

            context.getLogger().log(String.format("Downloading S3 object: %s %s\n", 
                    request.getBucket(), request.getKey()));
            S3Object s3object = s3Client.getObject(new GetObjectRequest(
                    request.getBucket(), request.getKey()));
            context.getLogger().log("Content-Type: "  + 
                    s3object.getObjectMetadata().getContentType() + "\n");
            InputStream in = s3object.getObjectContent();
            int b = 0;
            byte[] buf = new byte[2048];
            context.getLogger().log("Writing image on output\n");
            while ((b = in.read(buf)) > -1) {
                output.write(buf, 0, b);
            }

        } catch (AmazonServiceException e) {
            System.out.println("Error Message: " + e.getMessage());
        }   
    }

  private String getJSONInputStream(InputStream input) throws IOException {
      BufferedReader reader = new BufferedReader(new InputStreamReader(input));
      String data = "";
      String line;
      while ((line = reader.readLine()) != null) {
                data += line;
      }
      return data;
  }
}

The client code:

public class LambdaImageTest {
    private static final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) throws IOException {
        String bucketName = args[0]; 
        String key        = args[1]; 

        // Lambda client proxy
        AWSLambdaClient lambda = new AWSLambdaClient(new ProfileCredentialsProvider());
        lambda.configureRegion(Regions.EU_WEST_1);

        // Build InvokeRequest
        InvokeRequest invokeRequest = buildInvokeRequest("ImageProcessing",  
                new ImageRequest(bucketName, key));

        // Invoke and get result payload as ByteBuffer. Note error handling should be done here
        InvokeResult invokeResult = lambda.invoke(invokeRequest);
        ByteBuffer byteBuffer = invokeResult.getPayload();

        // Write payload to file. Output hardcoded...
        FileChannel out = new FileOutputStream("D:/test.jpg").getChannel(); 
        out.write(byteBuffer);
        out.close();
    }

    private static InvokeRequest buildInvokeRequest(String functionName, Object input) {

        InvokeRequest invokeRequest = new InvokeRequest();
        invokeRequest.setFunctionName(functionName); // Lambda function name identifier
        invokeRequest.setInvocationType(InvocationType.RequestResponse);
        invokeRequest.setLogType(LogType.None);

        if (input != null) {
            try {

                String payload = MAPPER.writer().writeValueAsString(input);
                invokeRequest.setPayload(payload);

            } catch (JsonProcessingException ex) {
                throw new LambdaSerializationException("Failed to serialize request object to JSON", ex);
            }
        }

        return invokeRequest;
    }
}

One should note that error handling needs to be improved here . The source in LambdaInvokerFactory holds the missing pieces.