java - Java 中的 AWS Lambda 字节流客户端

Question

我正在尝试用 Java 为 AWS Lambda 函数编写字节流客户端。我创建了 Lambda 函数作为 RequestStreamHandler 的实现。文档中概述了该项目的基础 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());
        }   
    }
}

这个硬编码示例在 Lambda 测试控制台上运行良好。我可以上传 JAR 并运行 lambda 函数(通过单击“测试”)。该函数的作用是检索 jpeg 文件内容并将字节流写入 OutputStream。我可以在测试控制台中看到二进制输出作为函数结果。到目前为止一切都很好。最终我将在 jpeg 上运行 ImageMagick 并调整它的大小——这是这个项目的目标。

我的客户端代码如下:

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
}

当我尝试在 Java 客户端中接收字节流时，我失败了。似乎没有办法接收字节流。客户端似乎试图将 reault 读取为 json 数据，这不是我想要的。我想直接读取字节流(jpeg 二进制内容)。我得到的错误是:

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)

如何在 AWS Lambda Java 客户端中正确接收字节流数据？

Answer 1

我找到了解决办法。 LambdaInvokerFactory 类将始终将请求和响应数据作为 JSON 处理，因此序列化和反序列化就是问题所在。尽管源代码包含了答案的线索，我已经删除了执行 lambda 函数调用的部分，但我绕过了 JSON 反序列化并直接访问了有效负载。很简单，但它确实应该已经在 LambdaInvokerFactory 类中了......

这是我的完整解决方案。 Lambda 函数代码:

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;
  }
}

客户端代码:

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;
    }
}

应该注意这里的错误处理需要改进。 LambdaInvokerFactory 中的源包含缺失的部分。