S3BackedStreamPayloadStore.java

package software.amazon.payloadoffloading;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.core.ResponseInputStream;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import java.io.InputStream;

public class S3BackedStreamPayloadStore extends S3BackedPayloadStore implements StreamPayloadStore {
    private static final Logger LOG = LoggerFactory.getLogger(S3BackedStreamPayloadStore.class);
    private final S3Dao s3Dao;
    private final String s3BucketName;

    public S3BackedStreamPayloadStore(S3Dao s3Dao, String s3BucketName) {
        super(s3Dao, s3BucketName);
        this.s3Dao = s3Dao;
        this.s3BucketName = s3BucketName;
    }

    @Override
    public String storeOriginalPayloadStream(InputStream payloadStream, String s3Key) {
        s3Dao.storeTextStreamInS3(s3BucketName, s3Key, payloadStream);
        LOG.info("S3 stream object created from InputStream, Bucket name: " + s3BucketName + ", Object key: " + s3Key + ".");
        return new PayloadS3Pointer(s3BucketName, s3Key).toJson();
    }

    @Override
    public ResponseInputStream<GetObjectResponse> getOriginalPayloadStreamStream(String payloadPointer) {
        PayloadS3Pointer s3Pointer = PayloadS3Pointer.fromJson(payloadPointer);
        String s3BucketName = s3Pointer.getS3BucketName();
        String s3Key = s3Pointer.getS3Key();

        ResponseInputStream<GetObjectResponse> stream = s3Dao.getTextStreamFromS3(s3BucketName, s3Key);
        LOG.info("S3 object stream retrieved, Bucket name: " + s3BucketName + ", Object key: " + s3Key);
        return stream;
    }
}