diff --git a/pom.xml b/pom.xml index 2ec00ac..2a00876 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.amazonaws amazon-sqs-java-extended-client-lib - 1.0.2 + 1.1.0-rc1 jar Amazon SQS Extended Client Library for Java An extension to the Amazon SQS client that enables sending and receiving messages up to 2GB via Amazon S3. diff --git a/src/main/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClient.java b/src/main/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClient.java index 8200d7b..1745789 100644 --- a/src/main/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClient.java +++ b/src/main/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClient.java @@ -24,11 +24,14 @@ import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.UUID; import java.util.Map.Entry; import com.amazonaws.AmazonClientException; import com.amazonaws.AmazonServiceException; +import com.amazonaws.services.s3.model.AccessControlList; +import com.amazonaws.services.s3.model.CannedAccessControlList; import com.amazonaws.services.s3.model.GetObjectRequest; import com.amazonaws.services.s3.model.ObjectMetadata; import com.amazonaws.services.s3.model.PutObjectRequest; @@ -1328,12 +1331,27 @@ private String getJSONFromS3Pointer(MessageS3Pointer s3Pointer) { return s3PointerStr; } + private void addAcl(PutObjectRequest request) { + // Add ACLs in request if provided in configuration. + AccessControlList acl = clientConfiguration.getS3Acl(); + if (null != acl) { + request.setAccessControlList(acl); + } + + CannedAccessControlList cannedAcl = clientConfiguration.getS3CannedAcl(); + if (null != cannedAcl) { + request.setCannedAcl(cannedAcl); + } + } + private void storeTextInS3(String s3Key, String messageContentStr, Long messageContentSize) { InputStream messageContentStream = new ByteArrayInputStream(messageContentStr.getBytes(StandardCharsets.UTF_8)); ObjectMetadata messageContentStreamMetadata = new ObjectMetadata(); messageContentStreamMetadata.setContentLength(messageContentSize); PutObjectRequest putObjectRequest = new PutObjectRequest(clientConfiguration.getS3BucketName(), s3Key, messageContentStream, messageContentStreamMetadata); + addAcl(putObjectRequest); // Add any ACL policy that might been set in the config to the request. + try { clientConfiguration.getAmazonS3Client().putObject(putObjectRequest); } catch (AmazonServiceException e) { diff --git a/src/main/java/com/amazon/sqs/javamessaging/ExtendedClientConfiguration.java b/src/main/java/com/amazon/sqs/javamessaging/ExtendedClientConfiguration.java index 4852178..47d2fee 100644 --- a/src/main/java/com/amazon/sqs/javamessaging/ExtendedClientConfiguration.java +++ b/src/main/java/com/amazon/sqs/javamessaging/ExtendedClientConfiguration.java @@ -17,6 +17,8 @@ import com.amazonaws.AmazonClientException; import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.model.AccessControlList; +import com.amazonaws.services.s3.model.CannedAccessControlList; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import com.amazonaws.annotation.NotThreadSafe; @@ -35,6 +37,9 @@ public class ExtendedClientConfiguration { private String s3BucketName; private boolean largePayloadSupport = false; private boolean alwaysThroughS3 = false; + private AccessControlList acl; + private CannedAccessControlList cannedAcl; + private int messageSizeThreshold = SQSExtendedClientConstants.DEFAULT_MESSAGE_SIZE_THRESHOLD; public ExtendedClientConfiguration() { @@ -48,6 +53,8 @@ public ExtendedClientConfiguration(ExtendedClientConfiguration other) { this.largePayloadSupport = other.largePayloadSupport; this.alwaysThroughS3 = other.alwaysThroughS3; this.messageSizeThreshold = other.messageSizeThreshold; + this.acl = other.acl; + this.cannedAcl = other.cannedAcl; } /** @@ -214,4 +221,70 @@ public ExtendedClientConfiguration withAlwaysThroughS3(boolean alwaysThroughS3) public boolean isAlwaysThroughS3() { return alwaysThroughS3; } + + /** + * Sets an ACL Policy for S3 to be applied with PutObject operations. + * + * @param s3Acl + * ACL Policy to apply when putting a large object in S3 + */ + public void setS3Acl(AccessControlList s3Acl) { + acl = s3Acl; + LOG.info("ACL Policy for S3 put object operations set."); + } + + /** + * Sets an ACL Policy for S3 to be applied with PutObject operations. + * + * @param s3Acl + * ACL Policy to apply when putting a large object in S3 + */ + public ExtendedClientConfiguration withS3ACL(AccessControlList s3Acl) { + setS3Acl(s3Acl); + return this; + } + + /** + * Gets S3 ACL Policy configured + * + * @return ACL Policy configured + * S3. Default: Null. + */ + public AccessControlList getS3Acl() { + return acl; + } + + + /** + * Sets a Canned ACL Policy for S3 to be applied with PutObject operations. + * + * @param s3CannedAcl + * ACL Policy to apply when putting a large object in S3 + */ + public void setS3CannedAcl(CannedAccessControlList s3CannedAcl) { + cannedAcl = s3CannedAcl; + LOG.info("ACL Policy for S3 put object operations set."); + } + + /** + * Sets a Canned ACL Policy for S3 to be applied with PutObject operations. + * + * @param s3CannedAcl + * ACL Policy to apply when putting a large object in S3 + */ + public ExtendedClientConfiguration withS3CannedACL(CannedAccessControlList s3CannedAcl) { + setS3CannedAcl(s3CannedAcl); + return this; + } + + /** + * Gets S3 Canned ACL Policy configured + * + * @return ACL Canned Policy configured + * S3. Default: Null. + */ + public CannedAccessControlList getS3CannedAcl() { + return cannedAcl; + } + } diff --git a/src/test/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClientTest.java b/src/test/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClientTest.java index 3dd70c3..ae3e04f 100644 --- a/src/test/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClientTest.java +++ b/src/test/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClientTest.java @@ -21,6 +21,8 @@ import java.util.Map; import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.model.AccessControlList; +import com.amazonaws.services.s3.model.CannedAccessControlList; import com.amazonaws.services.s3.model.PutObjectRequest; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClient; @@ -224,6 +226,26 @@ public void testWhenLargeMessgaeIsSentThenAttributeWithPayloadSizeIsAdded() { Assert.assertEquals(messageLength, (int)Integer.valueOf(attributes.get(SQSExtendedClientConstants.RESERVED_ATTRIBUTE_NAME).getStringValue())); } + @Test + public void testACLsAreAddedToRequestIfSetInConfig() { + AccessControlList acl = new AccessControlList(); + ExtendedClientConfiguration config = new ExtendedClientConfiguration() + .withLargePayloadSupportEnabled(mockS3, S3_BUCKET_NAME) + .withAlwaysThroughS3(true) + .withS3CannedACL(CannedAccessControlList.BucketOwnerFullControl) + .withS3ACL(acl); + + AmazonSQS sqsExtended = spy(new AmazonSQSExtendedClient(mock(AmazonSQSClient.class), config)); + SendMessageRequest messageRequest = new SendMessageRequest(SQS_QUEUE_URL, "someBody"); + sqsExtended.sendMessage(messageRequest); + + ArgumentCaptor putObjectRequestCaptor = ArgumentCaptor.forClass(PutObjectRequest.class); + verify(mockS3).putObject(putObjectRequestCaptor.capture()); + PutObjectRequest captured = putObjectRequestCaptor.getValue(); + Assert.assertEquals(acl, captured.getAccessControlList()); + Assert.assertEquals(CannedAccessControlList.BucketOwnerFullControl, captured.getCannedAcl()); + + } private String generateStringWithLength(int messageLength) { char[] charArray = new char[messageLength]; Arrays.fill(charArray, 'x'); diff --git a/src/test/java/com/amazon/sqs/javamessaging/ExtendedClientConfigurationTest.java b/src/test/java/com/amazon/sqs/javamessaging/ExtendedClientConfigurationTest.java index 9949853..fceb548 100644 --- a/src/test/java/com/amazon/sqs/javamessaging/ExtendedClientConfigurationTest.java +++ b/src/test/java/com/amazon/sqs/javamessaging/ExtendedClientConfigurationTest.java @@ -16,6 +16,8 @@ package com.amazon.sqs.javamessaging; import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.model.AccessControlList; +import com.amazonaws.services.s3.model.CannedAccessControlList; import com.amazonaws.services.s3.model.PutObjectRequest; import junit.framework.Assert; import org.junit.Before; @@ -116,5 +118,31 @@ public void testMessageSizeThreshold() { } + @Test + public void testS3ACL() { + ExtendedClientConfiguration extendedClientConfiguration = new ExtendedClientConfiguration(); + + Assert.assertNull(extendedClientConfiguration.getS3Acl()); + AccessControlList acl = new AccessControlList(); + extendedClientConfiguration.setS3Acl(acl); + Assert.assertEquals(acl, extendedClientConfiguration.getS3Acl()); + + extendedClientConfiguration = new ExtendedClientConfiguration() + .withS3ACL(acl); + Assert.assertEquals(acl, extendedClientConfiguration.getS3Acl()); + } + @Test + public void testS3CannedACL() { + ExtendedClientConfiguration extendedClientConfiguration = new ExtendedClientConfiguration(); + + Assert.assertNull(extendedClientConfiguration.getS3Acl()); + CannedAccessControlList acl = CannedAccessControlList.BucketOwnerFullControl; + extendedClientConfiguration.setS3CannedAcl(acl); + Assert.assertEquals(acl, extendedClientConfiguration.getS3CannedAcl()); + + extendedClientConfiguration = new ExtendedClientConfiguration() + .withS3CannedACL(acl); + Assert.assertEquals(acl, extendedClientConfiguration.getS3CannedAcl()); + } }