Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-45: Fix S3 hierarchy support

Browse Source 
Fixes GH-45 (https://github.com/spring-projects/spring-integration-aws/issues/45)

S3InboundFileSynchronizer failing when processing through S3 directories - modified approach for splitPathToBucketAndKey to account for the fact that there will be more than one forward slash in the path when subdirectories are used in an S3 bucket. Set up split operation to use the limit parameter to force the bucket identifier into the first element and the remainder of the text (the key) into the second element.

* modified unit test to include subdirectories.
*fixed additional usages of the split operation without the match limit parameter. DRY'd up some code. Cleaned up unit tests.

* added @author annotation at project lead's request.

Add some polishing
This commit is contained in:
Jim Krygowski
2016-08-04 11:46:17 -04:00
committed by Artem Bilan
parent 6a79933266
commit 72821bfb8c
3 changed files with 63 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
src/main/java/org/springframework/integration/aws/support/S3Session.java
View File

@@ -43,6 +43,7 @@ import com.amazonaws.services.s3.model.S3ObjectSummary;
* An Amazon S3 {@link Session} implementation.
*
* @author Artem Bilan
* @author Jim Krygowski
*/
public class S3Session implements Session<S3ObjectSummary> {
@@ -62,15 +63,10 @@ public class S3Session implements Session<S3ObjectSummary> {
@Override
public S3ObjectSummary[] list(String path) throws IOException {
Assert.hasText(path, "'path' must not be empty String.");
String[] bucketPrefix = path.split("/");
Assert.state(bucketPrefix.length > 0 && bucketPrefix[0].length() >= 3,
"S3 bucket name must be at least 3 characters long.");
String bucket = resolveBucket(bucketPrefix[0]);
String[] bucketPrefix = splitPathToBucketAndKey(path, false);
ListObjectsRequest listObjectsRequest = new ListObjectsRequest()
.withBucketName(bucket);
.withBucketName(bucketPrefix[0]);
if (bucketPrefix.length > 1) {
listObjectsRequest.setPrefix(bucketPrefix[1]);
}
@@ -103,14 +99,10 @@ public class S3Session implements Session<S3ObjectSummary> {
@Override
public String[] listNames(String path) throws IOException {
String[] bucketPrefix = path.split("/");
Assert.state(bucketPrefix.length > 0 && bucketPrefix[0].length() >= 3,
"S3 bucket name must be at least 3 characters long.");
String bucket = resolveBucket(bucketPrefix[0]);
String[] bucketPrefix = splitPathToBucketAndKey(path, false);
ListObjectsRequest listObjectsRequest = new ListObjectsRequest()
.withBucketName(bucket);
.withBucketName(bucketPrefix[0]);
if (bucketPrefix.length > 1) {
listObjectsRequest.setPrefix(bucketPrefix[1]);
}
@@ -136,15 +128,15 @@ public class S3Session implements Session<S3ObjectSummary> {
@Override
public boolean remove(String path) throws IOException {
String[] bucketKey = splitPathToBucketAndKey(path);
String[] bucketKey = splitPathToBucketAndKey(path, true);
this.amazonS3.deleteObject(bucketKey[0], bucketKey[1]);
return true;
}
@Override
public void rename(String pathFrom, String pathTo) throws IOException {
String[] bucketKeyFrom = splitPathToBucketAndKey(pathFrom);
String[] bucketKeyTo = splitPathToBucketAndKey(pathTo);
String[] bucketKeyFrom = splitPathToBucketAndKey(pathFrom, true);
String[] bucketKeyTo = splitPathToBucketAndKey(pathTo, true);
CopyObjectRequest copyRequest = new CopyObjectRequest(bucketKeyFrom[0], bucketKeyFrom[1],
bucketKeyTo[0], bucketKeyTo[1]);
this.amazonS3.copyObject(copyRequest);
@@ -155,7 +147,7 @@ public class S3Session implements Session<S3ObjectSummary> {
@Override
public void read(String source, OutputStream outputStream) throws IOException {
String[] bucketKey = splitPathToBucketAndKey(source);
String[] bucketKey = splitPathToBucketAndKey(source, true);
S3Object s3Object = this.amazonS3.getObject(bucketKey[0], bucketKey[1]);
S3ObjectInputStream objectContent = s3Object.getObjectContent();
try {
@@ -169,7 +161,7 @@ public class S3Session implements Session<S3ObjectSummary> {
@Override
public void write(InputStream inputStream, String destination) throws IOException {
Assert.notNull(inputStream, "'inputStream' must not be null.");
String[] bucketKey = splitPathToBucketAndKey(destination);
String[] bucketKey = splitPathToBucketAndKey(destination, true);
this.amazonS3.putObject(bucketKey[0], bucketKey[1], inputStream, new ObjectMetadata());
}
@@ -192,7 +184,7 @@ public class S3Session implements Session<S3ObjectSummary> {
@Override
public boolean exists(String path) throws IOException {
String[] bucketKey = splitPathToBucketAndKey(path);
String[] bucketKey = splitPathToBucketAndKey(path, true);
try {
this.amazonS3.getObjectMetadata(bucketKey[0], bucketKey[1]);
}
@@ -209,7 +201,7 @@ public class S3Session implements Session<S3ObjectSummary> {
@Override
public InputStream readRaw(String source) throws IOException {
String[] bucketKey = splitPathToBucketAndKey(source);
String[] bucketKey = splitPathToBucketAndKey(source, true);
S3Object s3Object = this.amazonS3.getObject(bucketKey[0], bucketKey[1]);
return s3Object.getObjectContent();
}
@@ -234,11 +226,19 @@ public class S3Session implements Session<S3ObjectSummary> {
return this.amazonS3;
}
private String[] splitPathToBucketAndKey(String path) {
private String[] splitPathToBucketAndKey(String path, boolean requireKey) {
Assert.hasText(path, "'path' must not be empty String.");
String[] bucketKey = path.split("/");
Assert.state(bucketKey.length == 2, "'path' must in pattern [BUCKET/KEY].");
Assert.state(bucketKey[0].length() >= 3, "S3 bucket name must be at least 3 characters long.");
String[] bucketKey = path.split("/", 2);
if (requireKey) {
Assert.state(bucketKey.length == 2, "'path' must in pattern [BUCKET/KEY].");
Assert.state(bucketKey[0].length() >= 3, "S3 bucket name must be at least 3 characters long.");
}
else {
Assert.state(bucketKey.length > 0 && bucketKey[0].length() >= 3,
"S3 bucket name must be at least 3 characters long.");
}
bucketKey[0] = resolveBucket(bucketKey[0]);
return bucketKey;
}

5
src/test/java/org/springframework/integration/aws/inbound/S3InboundChannelAdapterTests.java
View File

@@ -64,6 +64,7 @@ import com.amazonaws.services.s3.model.S3ObjectSummary;
/**
* @author Artem Bilan
* @autor Jim Krygowski
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@@ -100,7 +101,7 @@ public class S3InboundChannelAdapterTests {
for (File file : remoteFolder.listFiles()) {
S3Object s3Object = new S3Object();
s3Object.setBucketName(S3_BUCKET);
s3Object.setKey(file.getName());
s3Object.setKey("subdir/" + file.getName());
s3Object.setObjectContent(new FileInputStream(file));
S3_OBJECTS.add(s3Object);
}
@@ -191,7 +192,7 @@ public class S3InboundChannelAdapterTests {
synchronizer.setPreserveTimestamp(true);
synchronizer.setRemoteDirectory(S3_BUCKET);
synchronizer.setFilter(new S3RegexPatternFileListFilter(".*\\.test$"));
Expression expression = PARSER.parseExpression("#this.toUpperCase() + '.a'");
Expression expression = PARSER.parseExpression("(#this.contains('/') ? #this.substring(#this.lastIndexOf('/') + 1) : #this).toUpperCase() + '.a'");
synchronizer.setLocalFilenameGeneratorExpression(expression);
return synchronizer;
}

55
src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java
View File

@@ -102,6 +102,7 @@ import com.amazonaws.util.StringUtils;
/**
* @author Artem Bilan
* @author John Logan
* @author Jim Krygowski
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@@ -110,6 +111,13 @@ public class S3MessageHandlerTests {
private static SpelExpressionParser PARSER = new SpelExpressionParser();
// define the bucket and file names used throughout the test
private static final String S3_BUCKET_NAME = "myBucket";
private static final String S3_FILE_KEY_BAR = "subdir/bar";
private static final String S3_FILE_KEY_FOO = "subdir/foo";
@Rule
public TemporaryFolder temporaryFolder = new TemporaryFolder();
@@ -145,7 +153,7 @@ public class S3MessageHandlerTests {
verify(this.amazonS3, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());
PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue();
assertThat(putObjectRequest.getBucketName()).isEqualTo("myBucket");
assertThat(putObjectRequest.getBucketName()).isEqualTo(S3_BUCKET_NAME);
assertThat(putObjectRequest.getKey()).isEqualTo("foo.mp3");
assertThat(putObjectRequest.getFile()).isNotNull();
assertThat(putObjectRequest.getInputStream()).isNull();
@@ -167,7 +175,7 @@ public class S3MessageHandlerTests {
SetObjectAclRequest setObjectAclRequest = setObjectAclRequestArgumentCaptor.getValue();
assertThat(setObjectAclRequest.getBucketName()).isEqualTo("myBucket");
assertThat(setObjectAclRequest.getBucketName()).isEqualTo(S3_BUCKET_NAME);
assertThat(setObjectAclRequest.getKey()).isEqualTo("foo.mp3");
assertThat(setObjectAclRequest.getAcl()).isNull();
assertThat(setObjectAclRequest.getCannedAcl()).isEqualTo(CannedAccessControlList.PublicReadWrite);
@@ -188,7 +196,7 @@ public class S3MessageHandlerTests {
verify(this.amazonS3, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());
PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue();
assertThat(putObjectRequest.getBucketName()).isEqualTo("myBucket");
assertThat(putObjectRequest.getBucketName()).isEqualTo(S3_BUCKET_NAME);
assertThat(putObjectRequest.getKey()).isEqualTo("myStream");
assertThat(putObjectRequest.getFile()).isNull();
assertThat(putObjectRequest.getInputStream()).isNotNull();
@@ -234,7 +242,7 @@ public class S3MessageHandlerTests {
verify(this.amazonS3, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());
PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue();
assertThat(putObjectRequest.getBucketName()).isEqualTo("myBucket");
assertThat(putObjectRequest.getBucketName()).isEqualTo(S3_BUCKET_NAME);
assertThat(putObjectRequest.getKey()).isEqualTo("myStream");
assertThat(putObjectRequest.getFile()).isNull();
assertThat(putObjectRequest.getInputStream()).isNotNull();
@@ -255,7 +263,16 @@ public class S3MessageHandlerTests {
this.s3SendChannel.send(message);
File[] fileArray = directoryForDownload.listFiles();
// get the "root" directory
File[] directoryArray = directoryForDownload.listFiles();
assertThat(directoryArray).isNotNull();
assertThat(directoryArray.length).isEqualTo(1);
File subDirectory = directoryArray[0];
assertThat(subDirectory.getName()).isEqualTo("subdir");
// get the files we downloaded
File[] fileArray = subDirectory.listFiles();
assertThat(fileArray).isNotNull();
assertThat(fileArray.length).isEqualTo(2);
@@ -263,11 +280,11 @@ public class S3MessageHandlerTests {
Collections.sort(files, (o1, o2) -> o1.getName().compareTo(o2.getName()));
File file1 = files.get(0);
assertThat(file1.getName()).isEqualTo("bar");
assertThat(file1.getName()).isEqualTo(S3_FILE_KEY_BAR.split("/", 2)[1]);
assertThat(FileCopyUtils.copyToString(new FileReader(file1))).isEqualTo("bb");
File file2 = files.get(1);
assertThat(file2.getName()).isEqualTo("foo");
assertThat(file2.getName()).isEqualTo(S3_FILE_KEY_FOO.split("/", 2)[1]);
assertThat(FileCopyUtils.copyToString(new FileReader(file2))).isEqualTo("f");
}
@@ -308,14 +325,14 @@ public class S3MessageHandlerTests {
List<S3ObjectSummary> s3ObjectSummaries = new LinkedList<>();
S3ObjectSummary fileSummary1 = new S3ObjectSummary();
fileSummary1.setBucketName("myBucket");
fileSummary1.setKey("foo");
fileSummary1.setBucketName(S3_BUCKET_NAME);
fileSummary1.setKey(S3_FILE_KEY_FOO);
fileSummary1.setSize(1);
s3ObjectSummaries.add(fileSummary1);
S3ObjectSummary fileSummary2 = new S3ObjectSummary();
fileSummary2.setBucketName("myBucket");
fileSummary2.setKey("bar");
fileSummary2.setBucketName(S3_BUCKET_NAME);
fileSummary2.setKey(S3_FILE_KEY_BAR);
fileSummary2.setSize(2);
s3ObjectSummaries.add(fileSummary2);
@@ -323,8 +340,8 @@ public class S3MessageHandlerTests {
given(amazonS3.listObjects(any(ListObjectsRequest.class))).willReturn(objectListing);
final S3Object file1 = new S3Object();
file1.setBucketName("myBucket");
file1.setKey("foo");
file1.setBucketName(S3_BUCKET_NAME);
file1.setKey(S3_FILE_KEY_FOO);
try {
byte[] data = "f".getBytes(StringUtils.UTF8);
byte[] md5 = Md5Utils.computeMD5Hash(data);
@@ -338,8 +355,8 @@ public class S3MessageHandlerTests {
}
final S3Object file2 = new S3Object();
file2.setBucketName("myBucket");
file2.setKey("bar");
file2.setBucketName(S3_BUCKET_NAME);
file2.setKey(S3_FILE_KEY_BAR);
try {
byte[] data = "bb".getBytes(StringUtils.UTF8);
byte[] md5 = Md5Utils.computeMD5Hash(data);
@@ -355,10 +372,10 @@ public class S3MessageHandlerTests {
willAnswer(invocation -> {
GetObjectRequest getObjectRequest = (GetObjectRequest) invocation.getArguments()[0];
String key = getObjectRequest.getKey();
if ("foo".equals(key)) {
if (S3_FILE_KEY_FOO.equals(key)) {
return file1;
}
else if ("bar".equals(key)) {
else if (S3_FILE_KEY_BAR.equals(key)) {
return file2;
}
else {
@@ -411,7 +428,7 @@ public class S3MessageHandlerTests {
@Bean
@ServiceActivator(inputChannel = "s3SendChannel")
public MessageHandler s3MessageHandler() {
S3MessageHandler s3MessageHandler = new S3MessageHandler(amazonS3(), "myBucket");
S3MessageHandler s3MessageHandler = new S3MessageHandler(amazonS3(), S3_BUCKET_NAME);
s3MessageHandler.setCommandExpression(PARSER.parseExpression("headers.s3Command"));
Expression keyExpression =
PARSER.parseExpression("payload instanceof T(java.io.File) ? payload.name : headers.key");
@@ -436,7 +453,7 @@ public class S3MessageHandlerTests {
@Bean
@ServiceActivator(inputChannel = "s3ProcessChannel")
public MessageHandler s3ProcessMessageHandler() {
S3MessageHandler s3MessageHandler = new S3MessageHandler(amazonS3(), "myBucket", true);
S3MessageHandler s3MessageHandler = new S3MessageHandler(amazonS3(), S3_BUCKET_NAME, true);
s3MessageHandler.setOutputChannel(s3ReplyChannel());
s3MessageHandler.setCommand(S3MessageHandler.Command.COPY);
s3MessageHandler.setKeyExpression(PARSER.parseExpression("payload.key"));