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

Add ResourceIdResolver support

Browse Source 
I many cases we deal in application just with simple logical name for the target AWS entities, e.g. `myQueue`, `testBucket`.
Actually they must be resolved into the physical resources against the current environment.
E.g. the same  S3 `testBucket` ca be fully different in different regions.
 The SQS queue must be resolved into the resources with the current `Stack` context.
This commit is contained in:
Artem Bilan
2016-05-27 15:02:06 -04:00
parent 175cc06e0b
commit 6abc1210a1
13 changed files with 116 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
README.md
View File

@@ -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` (`<int-aws:sns-outbound-channel-adapter>`) 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.

1
src/main/java/org/springframework/integration/aws/config/xml/S3OutboundGatewayParser.java
View File

@@ -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;
}

1
src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java
View File

@@ -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;
}

19
src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java
View File

@@ -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;
}

18
src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java
View File

@@ -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);

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

@@ -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<S3ObjectSummary> {
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<S3ObjectSummary> {
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<S3ObjectSummary> {
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<S3ObjectSummary> {
@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<S3ObjectSummary> {
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;
}

7
src/main/java/org/springframework/integration/aws/support/S3SessionFactory.java
View File

@@ -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<S3ObjectSummary>, 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

14
src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd
View File

@@ -232,6 +232,18 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="resource-id-resolver">
<xsd:annotation>
<xsd:documentation>
The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean reference.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:simpleType name="s3CommandType">
@@ -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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

7
src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests-context.xml
View File

@@ -17,6 +17,10 @@
<constructor-arg value="org.springframework.integration.aws.outbound.S3MessageHandler$UploadMetadataProvider"/>
</bean>
<bean id="resourceIdResolver" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cloud.aws.core.env.ResourceIdResolver" />
</bean>
<int-aws:s3-outbound-channel-adapter s3="s3"
auto-startup="false"
channel="errorChannel"
@@ -29,7 +33,8 @@
destination-key-expression="'baz'"
object-acl-expression="'qux'"
progress-listener="s3ProgressListener"
upload-metadata-provider="uploadMetadataProvider"/>
upload-metadata-provider="uploadMetadataProvider"
resource-id-resolver="resourceIdResolver" />
<bean id="transferManager" class="com.amazonaws.services.s3.transfer.TransferManager"/>

6
src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests.java
View File

@@ -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();

10
src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml
View File

@@ -11,7 +11,14 @@
<constructor-arg value="com.amazonaws.services.sns.AmazonSNS"/>
</bean>
<int-aws:sns-outbound-channel-adapter id="defaultAdapter" sns="amazonSns">
<bean id="resourceIdResolver" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cloud.aws.core.env.ResourceIdResolver" />
</bean>
<int-aws:sns-outbound-channel-adapter
id="defaultAdapter"
sns="amazonSns"
resource-id-resolver="resourceIdResolver">
<int-aws:request-handler-advice-chain>
<bean class="org.springframework.integration.handler.advice.RequestHandlerRetryAdvice"/>
</int-aws:request-handler-advice-chain>
@@ -27,6 +34,7 @@
topic-arn="foo"
subject="bar"
body-expression="payload.toUpperCase()"
resource-id-resolver="resourceIdResolver"
auto-startup="false"
phase="201"/>

10
src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java
View File

@@ -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();

2
src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java
View File

@@ -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);