diff --git a/README.md b/README.md index 94e4fd1..9e4e488 100644 --- a/README.md +++ b/README.md @@ -286,7 +286,7 @@ Other interested clients may subscribe using different protocols like HTTP/HTTPS receive the messages. Plus mobile devices can be registered as subscribers from the AWS Management Console. Unfortunately [Spring Cloud AWS][] doesn't provide flexible components which can be used from the channel adapter -implementations, but Amazon SNS API is pretty simple, from other side. Hence Spring Integration AWS SNS Support is +implementations, but Amazon SNS API is pretty simple, on the other hand. Hence Spring Integration AWS SNS Support is straightforward and just allows to provide channel adapter foundation for Spring Integration applications. Since e-mail, SMS and mobile devices subscription/unsubscription confirmation is out of the Spring Integration @@ -353,15 +353,15 @@ HTTP headers, populated by the `DefaultHttpHeaderMapper`, are available for the ###Outbound Channel Adapter The `SnsMessageHandler` (``) is a simple one-way Outbound Channel Adapter -to send Topic Notification using `AmasonSNS` service. +to send Topic Notification using `AmazonSNS` service. This Channel Adapter (`MessageHandler`) accepts these options: -- `topic-arn` (`topic-arn-expression`) - the SNS Topic to send notification for. The `ResourceIdResolver` can be used -from the SpEL definition to determine the target Topic Arn from the local logical name; +- `topic-arn` (`topic-arn-expression`) - the SNS Topic to send notification for. - `subject` (`subject-expression`) - the SNS Notification Subject; - `body-expression` - the SpEL expression to evaluate the `message` property for the `com.amazonaws.services.sns.model.PublishRequest`. +- `resource-id-resolver` - a `ResourceIdResolver` bean reference to resolve logical topic names to physical resource ids; See `SnsMessageHandler` JavaDocs for more information. diff --git a/src/main/java/org/springframework/integration/aws/config/xml/S3OutboundGatewayParser.java b/src/main/java/org/springframework/integration/aws/config/xml/S3OutboundGatewayParser.java index 68718a5..c47b96c 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/S3OutboundGatewayParser.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/S3OutboundGatewayParser.java @@ -97,6 +97,7 @@ public class S3OutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "resource-id-resolver"); return builder; } diff --git a/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java b/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java index 73a3e6e..0a3f7b4 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java @@ -65,6 +65,7 @@ public class SnsOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "resource-id-resolver"); return builder; } diff --git a/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java index c2f77f6..ddfe6de 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java @@ -23,6 +23,7 @@ import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; +import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; @@ -120,6 +121,8 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler { private UploadMetadataProvider uploadMetadataProvider; + private ResourceIdResolver resourceIdResolver; + public S3MessageHandler(AmazonS3 amazonS3, String bucket) { this(amazonS3, bucket, false); } @@ -232,6 +235,14 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler { this.uploadMetadataProvider = uploadMetadataProvider; } + /** + * Specify a {@link ResourceIdResolver} to resolve logical bucket names to physical resource ids. + * @param resourceIdResolver the {@link ResourceIdResolver} to use. + */ + public void setResourceIdResolver(ResourceIdResolver resourceIdResolver) { + this.resourceIdResolver = resourceIdResolver; + } + @Override protected void doInit() { Assert.notNull(this.bucketExpression, "The 'bucketExpression' must not be null"); @@ -470,6 +481,10 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler { String.class); } + if (this.resourceIdResolver != null) { + destinationBucketName = this.resourceIdResolver.resolveToPhysicalResourceId(destinationBucketName); + } + Assert.state(destinationBucketName != null, "The 'destinationBucketExpression' must not be null for 'copy' operation and can't evaluate to null. " + "Root object is: " + requestMessage); @@ -502,6 +517,10 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler { + this.bucketExpression.getExpressionString() + "] must not evaluate to null. Root object is: " + requestMessage); + if (this.resourceIdResolver != null) { + bucketName = this.resourceIdResolver.resolveToPhysicalResourceId(bucketName); + } + return bucketName; } diff --git a/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java index 71bf2b9..6841f26 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java @@ -16,6 +16,7 @@ package org.springframework.integration.aws.outbound; +import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.expression.TypeLocator; @@ -96,6 +97,7 @@ public class SnsMessageHandler extends AbstractReplyProducingMessageHandler { private Expression bodyExpression; + private ResourceIdResolver resourceIdResolver; public SnsMessageHandler(AmazonSNS amazonSns) { this(amazonSns, false); @@ -140,6 +142,14 @@ public class SnsMessageHandler extends AbstractReplyProducingMessageHandler { this.bodyExpression = bodyExpression; } + /** + * Specify a {@link ResourceIdResolver} to resolve logical topic names to physical resource ids. + * @param resourceIdResolver the {@link ResourceIdResolver} to use. + */ + public void setResourceIdResolver(ResourceIdResolver resourceIdResolver) { + this.resourceIdResolver = resourceIdResolver; + } + @Override protected void doInit() { super.doInit(); @@ -164,11 +174,13 @@ public class SnsMessageHandler extends AbstractReplyProducingMessageHandler { publishRequest = (PublishRequest) payload; } else { + Assert.state(this.topicArnExpression != null, "'topicArn' or 'topicArnExpression' must be specified."); publishRequest = new PublishRequest(); - if (this.topicArnExpression != null) { - String topicArn = this.topicArnExpression.getValue(this.evaluationContext, requestMessage, String.class); - publishRequest.setTopicArn(topicArn); + String topicArn = this.topicArnExpression.getValue(this.evaluationContext, requestMessage, String.class); + if (this.resourceIdResolver != null) { + topicArn = this.resourceIdResolver.resolveToPhysicalResourceId(topicArn); } + publishRequest.setTopicArn(topicArn); if (this.subjectExpression != null) { String subject = this.subjectExpression.getValue(this.evaluationContext, requestMessage, String.class); diff --git a/src/main/java/org/springframework/integration/aws/support/S3Session.java b/src/main/java/org/springframework/integration/aws/support/S3Session.java index 6b1d40c..0fe98a2 100644 --- a/src/main/java/org/springframework/integration/aws/support/S3Session.java +++ b/src/main/java/org/springframework/integration/aws/support/S3Session.java @@ -24,6 +24,7 @@ import java.util.List; import org.apache.http.HttpStatus; +import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.integration.file.remote.session.Session; import org.springframework.util.Assert; import org.springframework.util.StreamUtils; @@ -47,7 +48,14 @@ public class S3Session implements Session { private final AmazonS3 amazonS3; + private final ResourceIdResolver resourceIdResolver; + public S3Session(AmazonS3 amazonS3) { + this(amazonS3, null); + } + + public S3Session(AmazonS3 amazonS3, ResourceIdResolver resourceIdResolver) { + this.resourceIdResolver = resourceIdResolver; Assert.notNull(amazonS3, "'amazonS3' must not be null."); this.amazonS3 = amazonS3; } @@ -59,8 +67,10 @@ public class S3Session implements Session { Assert.state(bucketPrefix.length > 0 && bucketPrefix[0].length() >= 3, "S3 bucket name must be at least 3 characters long."); + String bucket = resolveBucket(bucketPrefix[0]); + ListObjectsRequest listObjectsRequest = new ListObjectsRequest() - .withBucketName(bucketPrefix[0]); + .withBucketName(bucket); if (bucketPrefix.length > 1) { listObjectsRequest.setPrefix(bucketPrefix[1]); } @@ -82,14 +92,25 @@ public class S3Session implements Session { return objectSummaries.toArray(new S3ObjectSummary[objectSummaries.size()]); } + private String resolveBucket(String bucket) { + if (this.resourceIdResolver != null) { + return this.resourceIdResolver.resolveToPhysicalResourceId(bucket); + } + else { + return bucket; + } + } + @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]); + ListObjectsRequest listObjectsRequest = new ListObjectsRequest() - .withBucketName(bucketPrefix[0]); + .withBucketName(bucket); if (bucketPrefix.length > 1) { listObjectsRequest.setPrefix(bucketPrefix[1]); } @@ -165,7 +186,7 @@ public class S3Session implements Session { @Override public boolean rmdir(String directory) throws IOException { - this.amazonS3.deleteBucket(directory); + this.amazonS3.deleteBucket(resolveBucket(directory)); return true; } @@ -213,11 +234,12 @@ public class S3Session implements Session { return this.amazonS3; } - private static String[] splitPathToBucketAndKey(String path) { + private String[] splitPathToBucketAndKey(String path) { 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."); + Assert.state(bucketKey[0].length() >= 3, "S3 bucket name must be at least 3 characters long."); + bucketKey[0] = resolveBucket(bucketKey[0]); return bucketKey; } diff --git a/src/main/java/org/springframework/integration/aws/support/S3SessionFactory.java b/src/main/java/org/springframework/integration/aws/support/S3SessionFactory.java index 26005aa..8eb4ac3 100644 --- a/src/main/java/org/springframework/integration/aws/support/S3SessionFactory.java +++ b/src/main/java/org/springframework/integration/aws/support/S3SessionFactory.java @@ -16,6 +16,7 @@ package org.springframework.integration.aws.support; +import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.remote.session.SharedSessionCapable; import org.springframework.util.Assert; @@ -40,8 +41,12 @@ public class S3SessionFactory implements SessionFactory, Shared } public S3SessionFactory(AmazonS3 amazonS3) { + this(amazonS3, null); + } + + public S3SessionFactory(AmazonS3 amazonS3, ResourceIdResolver resourceIdResolver) { Assert.notNull(amazonS3, "'amazonS3' must not be null."); - this.s3Session = new S3Session(amazonS3); + this.s3Session = new S3Session(amazonS3, resourceIdResolver); } @Override diff --git a/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd index 83866bf..d59d0d9 100644 --- a/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd +++ b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd @@ -232,6 +232,18 @@ + + + + The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean reference. + + + + + + + + @@ -830,8 +842,6 @@ This attribute isn't mandatory and the the topic can be specified on the 'com.amazonaws.services.sns.model.PublishRequest' payload of the request Message. - The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean can be used from - the expression to resolve logical topic name to the real ARN. diff --git a/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests-context.xml index 42c5da4..0d337d6 100644 --- a/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests-context.xml +++ b/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests-context.xml @@ -17,6 +17,10 @@ + + + + + upload-metadata-provider="uploadMetadataProvider" + resource-id-resolver="resourceIdResolver" /> diff --git a/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests.java index 505150b..4ae0cf6 100644 --- a/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests.java +++ b/src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests.java @@ -24,6 +24,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.integration.aws.outbound.S3MessageHandler; @@ -78,6 +79,9 @@ public class S3MessageHandlerParserTests { @Autowired private S3MessageHandler.UploadMetadataProvider uploadMetadataProvider; + @Autowired + private ResourceIdResolver resourceIdResolver; + @Autowired private BeanFactory beanFactory; @@ -110,6 +114,8 @@ public class S3MessageHandlerParserTests { .isSameAs(this.progressListener); assertThat(TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "uploadMetadataProvider")) .isSameAs(this.uploadMetadataProvider); + assertThat(TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "resourceIdResolver")) + .isSameAs(this.resourceIdResolver); assertThat(this.s3OutboundChannelAdapter.getPhase()).isEqualTo(100); assertThat(this.s3OutboundChannelAdapter.isAutoStartup()).isFalse(); diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml index 322cab3..ce69321 100644 --- a/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml @@ -11,7 +11,14 @@ - + + + + + @@ -27,6 +34,7 @@ topic-arn="foo" subject="bar" body-expression="payload.toUpperCase()" + resource-id-resolver="resourceIdResolver" auto-startup="false" phase="201"/> diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java index d93b3fa..f4e95bd 100644 --- a/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java @@ -26,6 +26,7 @@ import org.junit.runner.RunWith; import org.springframework.aop.support.AopUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.expression.Expression; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice; @@ -75,6 +76,9 @@ public class SnsOutboundChannelAdapterParserTests { @Qualifier("snsGateway.handler") private MessageHandler snsGatewayHandler; + @Autowired + private ResourceIdResolver resourceIdResolver; + @Test public void testSnsOutboundChannelAdapterDefaultParser() throws Exception { Object handler = TestUtils.getPropertyValue(this.defaultAdapter, "handler"); @@ -93,16 +97,20 @@ public class SnsOutboundChannelAdapterParserTests { assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "topicArnExpression")).isNull(); assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "subjectExpression")).isNull(); assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "bodyExpression")).isNull(); + assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "resourceIdResolver")) + .isSameAs(this.resourceIdResolver); } @Test - public void testSnsOutboundChannelAdapterParser() { + public void testSnsOutboundGatewayParser() { assertThat(TestUtils.getPropertyValue(this.snsGateway, "inputChannel")).isSameAs(this.notificationChannel); assertThat(TestUtils.getPropertyValue(this.snsGateway, "handler")).isSameAs(this.snsGatewayHandler); assertThat(TestUtils.getPropertyValue(this.snsGateway, "autoStartup", Boolean.class)).isFalse(); assertThat(TestUtils.getPropertyValue(this.snsGateway, "phase", Integer.class)).isEqualTo(201); assertThat(TestUtils.getPropertyValue(this.snsGatewayHandler, "produceReply", Boolean.class)).isTrue(); assertThat(TestUtils.getPropertyValue(this.snsGatewayHandler, "outputChannel")).isSameAs(this.errorChannel); + assertThat(TestUtils.getPropertyValue(this.snsGatewayHandler, "resourceIdResolver")) + .isSameAs(this.resourceIdResolver); assertThat(TestUtils.getPropertyValue(this.snsGatewayHandler, "amazonSns")).isSameAs(this.amazonSns); assertThat(TestUtils.getPropertyValue(this.snsGatewayHandler, "evaluationContext")).isNotNull(); diff --git a/src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java index 5c001c8..9299a3c 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java @@ -167,7 +167,7 @@ public class SnsInboundChannelAdapterTests { } @Bean - public HttpRequestHandler sqsMessageDrivenChannelAdapter() { + public HttpRequestHandler snsInboundChannelAdapter() { SnsInboundChannelAdapter adapter = new SnsInboundChannelAdapter(amazonSns(), "/mySampleTopic"); adapter.setRequestChannel(inputChannel()); adapter.setHandleNotificationStatus(true);