java - 将 S3Client 升级到 S3AsyncClient 如何获取 Publisher？

Question

我正在将 aws 云服务中的 S3Client 升级到 S3AsyncClient。

我有这个函数可以转换为异步:

public PutObjectResponse uploadFileByUrl(String fileUrl, String builderId, PbModel category, String categoryId)

    URL url = new URL(fileUrl);
    String[] fileNameArray = url.getFile().split("\\.");
    var uniqueFileName = prepareFileName(fileNameArray[fileNameArray.length -1]);

    URLConnection connection = url.openConnection();

    long contentSize = connection.getContentLengthLong();
    InputStream inputStream = connection.getInputStream();

    return s3Client.putObject(myObjectRequestBuild, RequestBody.fromInputStream(inputStream, contentSize));
}

我有这个函数可以转换为异步:

public CompletableFuture<PutObjectResponse> uploadFileByUrl(String fileUrl, String builderId, PbModel category, String categoryId)

    URL url = new URL(fileUrl);
    String[] fileNameArray = url.getFile().split("\\.");
    var uniqueFileName = prepareFileName(fileNameArray[fileNameArray.length -1]);

    URLConnection connection = url.openConnection();

    long contentSize = connection.getContentLengthLong();
    InputStream inputStream = connection.getInputStream();

    return asyncClient.putObject(myObjectRequestBuild, AsyncRequestBody.fromPublisher(???));
}
    

正如您在上面的第二种方法中看到的那样，当我将上面的第一个函数转换为异步时，我需要使用 AsyncRequestBody 而不是 RequestBody。AsyncRequestBody 没有 fromInputStream 方法，但它有我想使用的 fromPublisher 方法，fromPublisher 方法作为 Publisher 的参数类型获取。

所以我的问题是如何将我的 inputStream 转换为 Publisher？

Answer 1

AsyncRequestBody doesn't have fromInputStream method

正确的，但是它有很多其他方法来创建 AsyncRequestBody :

1. fromByteBuffer(ByteBuffer byteBuffer)
2. fromBytes(byte[] bytes)
3. fromFile(File file)
4. fromFile(Path path)
5. fromPublisher(org.reactivestreams.Publisher<ByteBuffer> publisher)
6. fromString(String string)
7. fromString(String string, Charset cs)

考虑到以上情况，您有几个解决方案:

1. 转换 InputStream使用 IOUtils.toByteArray(inputStream) 到字节数组(或在 Java 9+ 中， inputStream.readAllBytes() )然后使用 fromBytes直接
2. 如上但随后转换 byte[]到 ByteBuffer使用 ByteBuffer.wrap(byteArray)然后使用 fromByteBuffer
3. 创建一个新的 File 指定文件名的对象，复制 InputStream 的内容到文件的 FileOutputStream使用 IOUtils.copy() , 然后使用 fromFile(File file)
4. 同上但不提供 File对象，提供它到 fromFile(Path path) 的路径在你写给它之后是FileOutputStream
5. 转换 InputSteam到 Publisher<ByteArray>使用 DataBufferUtils.readByteChannel 来自 Spring 框架，Akka StreamConverters 等，然后使用 fromPublisher
6. 转换 InputStream到 UTF-8 编码 String然后使用 fromString(String string) (如果是 UTF-8 编码，则无需指定 Charset)
7. 转换 InputStream到非 UTF-8 编码 String然后使用 fromString(String string, Charset cs) , 指定 CharSet

当然，上面的一些内容在你的情况下是多余的，例如fromFile(Path path)适用于您已经存储并转换 InputSteam 的文件到 Publisher<ByteArray>会很痛苦，但为了完整性，我已经包含了所有可能的解决方案。

---

我会使用解决方案 #1 来解决这个问题，从而生成上述代码中最干净、最简单的代码。

转换 InputStream至 byte[]使用 inputStream.readAllBytes()然后使用 AsyncRequestBody.fromBytes(...) .

这应该有效:

public CompletableFuture<PutObjectResponse> uploadFileByUrl(String fileUrl, String builderId, PbModel category, String categoryId)
    URL url = new URL(fileUrl);
    String[] fileNameArray = url.getFile().split("\\.");
    var uniqueFileName = prepareFileName(fileNameArray[fileNameArray.length -1]);

    URLConnection connection = url.openConnection();

    long contentSize = connection.getContentLengthLong();
    InputStream inputStream = connection.getInputStream();

    byte[] fileByteArray = inputStream.readAllBytes();

    return asyncClient.putObject(myObjectRequestBuild, AsyncRequestBody.fromBytes(fileByteArray));
}