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

GH-344 - Polishing.

Browse Source 
Switch to DelegatingEventExternalizer APIs changed to return CompletableFuture. Imports, formatting. Drop dependencies not needed.
This commit is contained in:
Oliver Drotbohm
2023-10-31 23:32:06 +01:00
parent d95da874bd
commit e8898c1da1
12 changed files with 148 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pom.xml
View File

@@ -43,6 +43,7 @@
<spring.version>6.1.0-RC1</spring.version> <!-- For Javadoc links only -->
<spring-boot.version>3.2.0-RC1</spring-boot.version>
<spring-cloud-aws-bom.version>3.0.2</spring-cloud-aws-bom.version>
</properties>
<developers>

6
spring-modulith-events/spring-modulith-events-aws-sns/pom.xml
View File

@@ -34,12 +34,6 @@
<artifactId>spring-cloud-aws-sns</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<optional>true</optional>
</dependency>
<!-- Test dependencies -->
<dependency>

22
spring-modulith-events/spring-modulith-events-aws-sns/src/main/java/org/springframework/modulith/events/aws/sns/SnsEventExternalizerConfiguration.java
View File

@@ -18,10 +18,11 @@ package org.springframework.modulith.events.aws.sns;
import io.awspring.cloud.sns.core.SnsNotification;
import io.awspring.cloud.sns.core.SnsOperations;
import io.awspring.cloud.sns.core.SnsTemplate;
import java.util.concurrent.CompletableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.sns.model.InvalidParameterException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
@@ -30,7 +31,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.expression.BeanFactoryResolver;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.modulith.events.EventExternalizationConfiguration;
import org.springframework.modulith.events.config.EventExternalizationAutoConfiguration;
import org.springframework.modulith.events.support.BrokerRouting;
@@ -40,6 +40,7 @@ import org.springframework.modulith.events.support.DelegatingEventExternalizer;
* Auto-configuration to set up a {@link DelegatingEventExternalizer} to externalize events to SNS.
*
* @author Maciej Walkowiak
* @author Oliver Drotbohm
* @since 1.1
*/
@AutoConfiguration
@@ -64,22 +65,17 @@ class SnsEventExternalizerConfiguration {
return new DelegatingEventExternalizer(configuration, (target, payload) -> {
var routing = BrokerRouting.of(target, context);
var builder = SnsNotification.builder(payload);
var key = routing.getKey(payload);
// when routing key is set, SNS topic must be a FIFO topic
if (key != null) {
builder.groupId(key);
}
try {
operations.sendNotification(routing.getTarget(), builder.build());
} catch (MessageDeliveryException e) {
// message delivery may fail if groupId is set and topic is not a FIFO topic, or content based deduplication has not been set on topic attributes.
if (e.getCause() instanceof InvalidParameterException) {
logger.error("Failed to send notification to SNS topic {}:{}", routing.getTarget(), e.getCause().getMessage());
}
throw e;
}
operations.sendNotification(routing.getTarget(), builder.build());
return CompletableFuture.completedFuture(null);
});
}
}

2
spring-modulith-events/spring-modulith-events-aws-sns/src/main/java/org/springframework/modulith/events/aws/sns/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* SNS event externalization support.
* AWS SNS event externalization support.
*/
@org.springframework.lang.NonNullApi
package org.springframework.modulith.events.aws.sns;

14
spring-modulith-events/spring-modulith-events-aws-sns/src/test/java/org/springframework/modulith/events/aws/sns/SnsEventExternalizerConfigurationIntegrationTests.java
View File

@@ -15,17 +15,17 @@
*/
package org.springframework.modulith.events.aws.sns;
import io.awspring.cloud.sns.core.SnsOperations;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import io.awspring.cloud.sns.core.SnsOperations;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.modulith.events.EventExternalizationConfiguration;
import org.springframework.modulith.events.support.DelegatingEventExternalizer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/**
* Integration tests for {@link SnsEventExternalizerConfiguration}.
*
@@ -34,7 +34,7 @@ import static org.mockito.Mockito.mock;
*/
class SnsEventExternalizerConfigurationIntegrationTests {
@Test // GH-342
@Test // GH-344
void registersExternalizerByDefault() {
basicSetup()
@@ -43,7 +43,7 @@ class SnsEventExternalizerConfigurationIntegrationTests {
});
}
@Test // GH-342
@Test // GH-344
void disablesExternalizationIfConfigured() {
basicSetup()

197
spring-modulith-events/spring-modulith-events-aws-sns/src/test/java/org/springframework/modulith/events/aws/sns/SnsEventPublicationIntegrationTests.java
View File

@@ -15,16 +15,18 @@
*/
package org.springframework.modulith.events.aws.sns;
import java.util.Map;
import static org.assertj.core.api.Assertions.*;
import static org.awaitility.Awaitility.*;
import lombok.RequiredArgsConstructor;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.utility.DockerImageName;
import lombok.Value;
import software.amazon.awssdk.services.sns.SnsClient;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
import java.util.Map;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
@@ -34,137 +36,132 @@ import org.springframework.modulith.events.ApplicationModuleListener;
import org.springframework.modulith.events.Externalized;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.transaction.annotation.Transactional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.utility.DockerImageName;
/**
* Integration tests for SQS-based event publication.
*
* @author Maciej Walkowiak
* @author Oliver Drotbohm
*/
@SpringBootTest
class SnsEventPublicationIntegrationTests {
@Autowired TestPublisher publisher;
@Autowired SnsClient snsClient;
@Autowired SqsAsyncClient sqsAsyncClient;
@Autowired TestPublisher publisher;
@Autowired SnsClient snsClient;
@Autowired SqsAsyncClient sqsAsyncClient;
@SpringBootApplication
static class TestConfiguration {
@SpringBootApplication
static class TestConfiguration {
@Bean
LocalStackContainer localStackContainer(DynamicPropertyRegistry registry) {
var localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:2.3.2"));
registry.add("spring.cloud.aws.endpoint", localstack::getEndpoint);
registry.add("spring.cloud.aws.credentials.access-key", localstack::getAccessKey);
registry.add("spring.cloud.aws.credentials.secret-key", localstack::getSecretKey);
registry.add("spring.cloud.aws.region.static", localstack::getRegion);
return localstack;
}
@Bean
LocalStackContainer localStackContainer(DynamicPropertyRegistry registry) {
@Bean
TestPublisher testPublisher(ApplicationEventPublisher publisher) {
return new TestPublisher(publisher);
}
var localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:2.3.2"));
@Bean
TestListener testListener() {
return new TestListener();
}
}
registry.add("spring.cloud.aws.endpoint", localstack::getEndpoint);
registry.add("spring.cloud.aws.credentials.access-key", localstack::getAccessKey);
registry.add("spring.cloud.aws.credentials.secret-key", localstack::getSecretKey);
registry.add("spring.cloud.aws.region.static", localstack::getRegion);
@Test
void publishesEventToSns() {
return localstack;
}
var topicArn = snsClient.createTopic(request -> request.name("target")).topicArn();
@Bean
TestPublisher testPublisher(ApplicationEventPublisher publisher) {
return new TestPublisher(publisher);
}
var queueUrl = sqsAsyncClient.createQueue(request -> request.queueName("queue"))
.join()
.queueUrl();
@Bean
TestListener testListener() {
return new TestListener();
}
}
var queueArn = sqsAsyncClient
.getQueueAttributes(r -> r.queueUrl(queueUrl).attributeNames(QueueAttributeName.QUEUE_ARN))
.join().attributes().get(QueueAttributeName.QUEUE_ARN);
snsClient.subscribe(r -> r.topicArn(topicArn).protocol("sqs").endpoint(queueArn));
@Test // GH-344
void publishesEventToSns() {
publisher.publishEvent();
var topicArn = snsClient.createTopic(request -> request.name("target")).topicArn();
await().untilAsserted(() -> {
var response = sqsAsyncClient.receiveMessage(r -> r.queueUrl(queueUrl)).join();
var queueUrl = sqsAsyncClient.createQueue(request -> request.queueName("queue"))
.join()
.queueUrl();
assertThat(response.hasMessages()).isTrue();
});
}
var queueArn = sqsAsyncClient
.getQueueAttributes(r -> r.queueUrl(queueUrl).attributeNames(QueueAttributeName.QUEUE_ARN))
.join().attributes().get(QueueAttributeName.QUEUE_ARN);
@Test
void publishesEventWithGroupIdToSns() {
snsClient.subscribe(r -> r.topicArn(topicArn).protocol("sqs").endpoint(queueArn));
var topicArn = snsClient.createTopic(request -> request.name("target.fifo")
.attributes(Map.of(
"FifoTopic", "true",
"ContentBasedDeduplication", "true"
)))
.topicArn();
publisher.publishEvent();
var queueUrl = sqsAsyncClient.createQueue(request -> request.queueName("queue.fifo")
.attributes(Map.of(QueueAttributeName.FIFO_QUEUE, "true")))
.join()
.queueUrl();
await().untilAsserted(() -> {
var queueArn = sqsAsyncClient
.getQueueAttributes(r -> r.queueUrl(queueUrl).attributeNames(QueueAttributeName.QUEUE_ARN))
.join().attributes().get(QueueAttributeName.QUEUE_ARN);
snsClient.subscribe(r -> r.topicArn(topicArn).protocol("sqs").endpoint(queueArn));
var response = sqsAsyncClient.receiveMessage(r -> r.queueUrl(queueUrl)).join();
publisher.publishEventWithKey();
assertThat(response.hasMessages()).isTrue();
});
}
await().untilAsserted(() -> {
var response = sqsAsyncClient.receiveMessage(r -> r.queueUrl(queueUrl)).join();
assertThat(response.hasMessages()).isTrue();
});
}
@Test // GH-344
void publishesEventWithGroupIdToSns() {
@Externalized("target")
static class TestEvent { }
var topicArn = snsClient.createTopic(request -> request.name("target.fifo")
.attributes(Map.of(
"FifoTopic", "true",
"ContentBasedDeduplication", "true")))
.topicArn();
@Externalized("target.fifo::#{getKey()}")
static class TestEventWithKey {
private final String key;
var queueUrl = sqsAsyncClient.createQueue(request -> request.queueName("queue.fifo")
.attributes(Map.of(QueueAttributeName.FIFO_QUEUE, "true")))
.join()
.queueUrl();
TestEventWithKey(String key) {
this.key = key;
}
var queueArn = sqsAsyncClient
.getQueueAttributes(r -> r.queueUrl(queueUrl).attributeNames(QueueAttributeName.QUEUE_ARN))
.join().attributes().get(QueueAttributeName.QUEUE_ARN);
snsClient.subscribe(r -> r.topicArn(topicArn).protocol("sqs").endpoint(queueArn));
public String getKey() {
return key;
}
}
publisher.publishEventWithKey();
@RequiredArgsConstructor
static class TestPublisher {
await().untilAsserted(() -> {
var response = sqsAsyncClient.receiveMessage(r -> r.queueUrl(queueUrl)).join();
assertThat(response.hasMessages()).isTrue();
});
}
private final ApplicationEventPublisher events;
@Externalized("target")
static class TestEvent {}
@Transactional
void publishEvent() {
events.publishEvent(new TestEvent());
}
@Value
@Externalized("target.fifo::#{getKey()}")
static class TestEventWithKey {
String key;
}
@Transactional
void publishEventWithKey() {
events.publishEvent(new TestEventWithKey("aKey"));
}
}
@RequiredArgsConstructor
static class TestPublisher {
static class TestListener {
private final ApplicationEventPublisher events;
@ApplicationModuleListener
void on(TestEvent event) {
}
@Transactional
void publishEvent() {
events.publishEvent(new TestEvent());
}
@ApplicationModuleListener
void on(TestEventWithKey event) {
}
}
@Transactional
void publishEventWithKey() {
events.publishEvent(new TestEventWithKey("aKey"));
}
}
static class TestListener {
@ApplicationModuleListener
void on(TestEvent event) {}
@ApplicationModuleListener
void on(TestEventWithKey event) {}
}
}

6
spring-modulith-events/spring-modulith-events-aws-sqs/pom.xml
View File

@@ -34,12 +34,6 @@
<artifactId>spring-cloud-aws-sqs</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<optional>true</optional>
</dependency>
<!-- Test dependencies -->
<dependency>

11
spring-modulith-events/spring-modulith-events-aws-sqs/src/main/java/org/springframework/modulith/events/aws/sqs/SqsEventExternalizerConfiguration.java
View File

@@ -17,9 +17,11 @@ package org.springframework.modulith.events.aws.sqs;
import io.awspring.cloud.sqs.operations.SqsOperations;
import io.awspring.cloud.sqs.operations.SqsTemplate;
import java.util.concurrent.CompletableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
@@ -37,6 +39,7 @@ import org.springframework.modulith.events.support.DelegatingEventExternalizer;
* Auto-configuration to set up a {@link DelegatingEventExternalizer} to externalize events to SQS.
*
* @author Maciej Walkowiak
* @author Oliver Drotbohm
* @since 1.1
*/
@AutoConfiguration
@@ -62,13 +65,15 @@ class SqsEventExternalizerConfiguration {
var routing = BrokerRouting.of(target, context);
operations.send(sqsSendOptions -> {
return CompletableFuture.completedFuture(operations.send(sqsSendOptions -> {
var options = sqsSendOptions.queue(routing.getTarget()).payload(payload);
var key = routing.getKey(payload);
if (key != null) {
options.messageGroupId(key);
}
});
}));
});
}
}

2
spring-modulith-events/spring-modulith-events-aws-sqs/src/main/java/org/springframework/modulith/events/aws/sqs/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* SQS event externalization support.
* AWS SQS event externalization support.
*/
@org.springframework.lang.NonNullApi
package org.springframework.modulith.events.aws.sqs;

14
spring-modulith-events/spring-modulith-events-aws-sqs/src/test/java/org/springframework/modulith/events/aws/sqs/SqsEventExternalizerConfigurationIntegrationTests.java
View File

@@ -15,17 +15,17 @@
*/
package org.springframework.modulith.events.aws.sqs;
import io.awspring.cloud.sqs.operations.SqsOperations;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import io.awspring.cloud.sqs.operations.SqsOperations;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.modulith.events.EventExternalizationConfiguration;
import org.springframework.modulith.events.support.DelegatingEventExternalizer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/**
* Integration tests for {@link SqsEventExternalizerConfiguration}.
*
@@ -34,7 +34,7 @@ import static org.mockito.Mockito.mock;
*/
class SqsEventExternalizerConfigurationIntegrationTests {
@Test // GH-342
@Test // GH-344
void registersExternalizerByDefault() {
basicSetup()
@@ -43,7 +43,7 @@ class SqsEventExternalizerConfigurationIntegrationTests {
});
}
@Test // GH-342
@Test // GH-344
void disablesExternalizationIfConfigured() {
basicSetup()

33
spring-modulith-events/spring-modulith-events-aws-sqs/src/test/java/org/springframework/modulith/events/aws/sqs/SqsEventPublicationIntegrationTests.java
View File

@@ -15,15 +15,17 @@
*/
package org.springframework.modulith.events.aws.sqs;
import java.util.Map;
import static org.assertj.core.api.Assertions.*;
import static org.awaitility.Awaitility.*;
import lombok.RequiredArgsConstructor;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.utility.DockerImageName;
import lombok.Value;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
import java.util.Map;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
@@ -33,14 +35,15 @@ import org.springframework.modulith.events.ApplicationModuleListener;
import org.springframework.modulith.events.Externalized;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.transaction.annotation.Transactional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.utility.DockerImageName;
/**
* Integration tests for SQS-based event publication.
*
* @author Maciej Walkowiak
* @author Oliver Drotbohm
* @since 1.1
*/
@SpringBootTest
class SqsEventPublicationIntegrationTests {
@@ -53,11 +56,14 @@ class SqsEventPublicationIntegrationTests {
@Bean
LocalStackContainer localStackContainer(DynamicPropertyRegistry registry) {
var localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:2.3.2"));
registry.add("spring.cloud.aws.endpoint", localstack::getEndpoint);
registry.add("spring.cloud.aws.credentials.access-key", localstack::getAccessKey);
registry.add("spring.cloud.aws.credentials.secret-key", localstack::getSecretKey);
registry.add("spring.cloud.aws.region.static", localstack::getRegion);
return localstack;
}
@@ -92,7 +98,7 @@ class SqsEventPublicationIntegrationTests {
void publishesEventWithGroupIdToSqs() throws Exception {
var queueUrl = sqsAsyncClient.createQueue(request -> request.queueName("target.fifo")
.attributes(Map.of(QueueAttributeName.FIFO_QUEUE, "true")))
.attributes(Map.of(QueueAttributeName.FIFO_QUEUE, "true")))
.join()
.queueUrl();
@@ -108,17 +114,10 @@ class SqsEventPublicationIntegrationTests {
@Externalized("target")
static class TestEvent {}
@Value
@Externalized("target.fifo::#{getKey()}")
static class TestEventWithKey {
private final String key;
TestEventWithKey(String key) {
this.key = key;
}
public String getKey() {
return key;
}
String key;
}
@RequiredArgsConstructor

2
spring-modulith-events/spring-modulith-events-core/pom.xml
View File

@@ -15,7 +15,7 @@
<properties>
<module.name>org.springframework.modulith.events.core</module.name>
</properties>
<dependencies>
<dependency>