Refactor

Author: MatejNedic

spring-cloud-aws-autoconfigure/src/main/java/io/awspring/cloud/autoconfigure/sns/SnsAutoConfiguration.java

@@ -35,7 +35,7 @@
 import io.awspring.cloud.sns.core.batch.converter.DefaultSnsMessageConverter;
 import io.awspring.cloud.sns.core.batch.converter.SnsMessageConverter;
 import io.awspring.cloud.sns.core.batch.executor.BatchExecutionStrategy;
-import io.awspring.cloud.sns.core.batch.executor.DefaultBatchExecutionStrategy;
+import io.awspring.cloud.sns.core.batch.executor.SequentialBatchExecutionStrategy;
 import io.awspring.cloud.sns.sms.SnsSmsOperations;
 import io.awspring.cloud.sns.sms.SnsSmsTemplate;
 import java.util.List;
@@ -146,8 +146,8 @@ public DefaultSnsMessageConverter defaultSnsMessageConverter(Optional<JsonMapper
 
 		@ConditionalOnMissingBean(BatchExecutionStrategy.class)
 		@Bean
-		public DefaultBatchExecutionStrategy defaultBatchExecutionStrategy(SnsClient snsClient) {
-			return new DefaultBatchExecutionStrategy(snsClient);
+		public SequentialBatchExecutionStrategy defaultBatchExecutionStrategy(SnsClient snsClient) {
+			return new SequentialBatchExecutionStrategy(snsClient);
 		}
 
 		@ConditionalOnMissingBean(SnsBatchOperations.class)

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/SnsHeaderConverterUtil.java

@@ -29,17 +29,14 @@
 import org.jspecify.annotations.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.support.AbstractMessageChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
 import org.springframework.util.NumberUtils;
 import software.amazon.awssdk.core.SdkBytes;
-import software.amazon.awssdk.services.sns.SnsClient;
 import software.amazon.awssdk.services.sns.model.MessageAttributeValue;
 
 /**
- * Implementation of {@link AbstractMessageChannel} which is used for converting and sending messages via
- * {@link SnsClient} to SNS.
+ * Util class used to convert {@link Message} headers to SNS messageAttributes.
  *
  * @author Agim Emruli
  * @author Alain Sahli
@@ -49,7 +46,7 @@
  */
 public class SnsHeaderConverterUtil {
 
-	public static JsonStringEncoder jsonStringEncoder = JsonStringEncoder.create();
+	private static final JsonStringEncoder jsonStringEncoder = JsonStringEncoder.create();
 	private static final Log logger = LogFactory.getLog(SnsHeaderConverterUtil.class);
 
 	public static Map<String, MessageAttributeValue> toSnsMessageAttributes(Message<?> message) {

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/SnsHeaders.java

@@ -48,12 +48,6 @@ public final class SnsHeaders {
 	 */
 	public static final String MESSAGE_GROUP_ID_HEADER = "message-group-id";
 
-	/**
-	 * Message id for SNS message (applies only to Batch request). The value of this header is set to
-	 * {@link PublishBatchRequestEntry#id()}}.
-	 */
-	public static final String MESSAGE_ID = "message-id";
-
 	/**
 	 * Message Deduplication id for SNS message. The value of this header is set to
 	 * {@link PublishRequest#messageDeduplicationId()}}}.

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/SnsBatchOperations.java

@@ -18,6 +18,7 @@
 import java.util.Collection;
 import java.util.Map;
 
+import io.awspring.cloud.sns.core.SnsNotification;
 import org.jspecify.annotations.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
@@ -45,17 +46,33 @@ public interface SnsBatchOperations {
 	<T> BatchResult sendBatch(String topicName, Collection<Message<T>> messages);
 
 	/**
-	 * Converts a collection of payloads to Spring messages and sends them as a batch to the specified SNS topic.
+	 * Converts a collection of POJOs to Spring messages and sends them as a batch to the specified SNS topic.
 	 * Batch API limit, messages will be split into groups of 10.
 	 * <p>
 	 * The result contains information about both successful and failed messages.
 	 *
 	 * @param topicName The logical name of the SNS topic
-	 * @param payload Collection of payloads to convert and send
+	 * @param payloads Collection of payloads to convert and send
 	 * @param <T> The type of the payload
 	 * @return BatchResult containing successful results and errors if there are any
 	 */
-	<T> BatchResult convertAndSend(String topicName, Collection<T> payload);
+	<T> BatchResult convertAndSend(String topicName, Collection<T> payloads);
+
+
+	/**
+	 * Converts a collection of notifications to Spring messages and sends them as a batch to the specified SNS topic.
+	 * Batch API limit, messages will be split into groups of 10.
+	 * <p>
+	 * The result contains information about both successful and failed messages.
+	 *
+	 * @param topicName The logical name of the SNS topic
+	 * @param notifications Collection of {@link SnsNotification} to convert and send
+	 * @param <T> The type of the payload
+	 * @return BatchResult containing successful results and errors if there are any
+	 */
+	<T> BatchResult sendBatchNotifications(String topicName, Collection<SnsNotification<T>> notifications);
+
+
 
 
 

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/SnsBatchTemplate.java

@@ -15,6 +15,7 @@
  */
 package io.awspring.cloud.sns.core.batch;
 
+import io.awspring.cloud.sns.core.SnsNotification;
 import io.awspring.cloud.sns.core.TopicArnResolver;
 import io.awspring.cloud.sns.core.batch.converter.SnsMessageConverter;
 import io.awspring.cloud.sns.core.batch.executor.BatchExecutionStrategy;
@@ -66,21 +67,43 @@ public SnsBatchTemplate(SnsMessageConverter snsMessageConverter, BatchExecutionS
 	 */
 	@Override
 	public <T> BatchResult sendBatch(String topicName, Collection<Message<T>> messages) {
+		Assert.notNull(topicName, "topicName is required");
+		Assert.notNull(messages, "messages are required");
+
 		var batchList = messages.stream().map(snsMessageConverter::covertMessage).toList();
 		return batchExecutionStrategy.send(topicArnResolver.resolveTopicArn(topicName), batchList);
 	}
 
 	/**
-	 * Converts a collection of payloads to Spring messages and sends them as a batch to the specified SNS topic.
+	 * Converts a collection of POJOs to Spring messages and sends them as a batch to the specified SNS topic.
+	 *
+	 * @param topicName The logical name of the SNS topic
+	 * @param payloads Collection of payloads to convert and send
+	 * @param <T> The type of the payload
+	 * @return BatchResult containing successful results and any errors
+	 */
+	@Override
+	public <T> BatchResult convertAndSend(String topicName, Collection<T> payloads) {
+		Assert.notNull(topicName, "topicName is required");
+		Assert.notNull(payloads, "payloads are required");
+		var batchList = payloads.stream().map(it -> MessageBuilder.withPayload(it).build()).map(snsMessageConverter::covertMessage).collect(Collectors.toList());
+		return batchExecutionStrategy.send(topicArnResolver.resolveTopicArn(topicName), batchList);
+	}
+
+
+	/**
+	 * Converts a collection of {@link SnsNotification} to Spring messages and sends them as a batch to the specified SNS topic.
 	 *
 	 * @param topicName The logical name of the SNS topic
-	 * @param payload Collection of payloads to convert and send
+	 * @param notifications Collection of payloads to convert and send
 	 * @param <T> The type of the payload
 	 * @return BatchResult containing successful results and any errors
 	 */
 	@Override
-	public <T> BatchResult convertAndSend(String topicName, Collection<T> payload) {
-		var batchList = payload.stream().map(it -> MessageBuilder.withPayload(it).build()).map(snsMessageConverter::covertMessage).collect(Collectors.toList());
+	public <T> BatchResult sendBatchNotifications(String topicName, Collection<SnsNotification<T>> notifications) {
+		Assert.notNull(topicName, "topicName is required");
+		Assert.notNull(notifications, "notifications are required");
+		var batchList = notifications.stream().map(it -> MessageBuilder.withPayload(it.getPayload()).copyHeaders(it.getHeaders()).build()).map(snsMessageConverter::covertMessage).collect(Collectors.toList());
 		return batchExecutionStrategy.send(topicArnResolver.resolveTopicArn(topicName), batchList);
 	}
 

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/converter/DefaultSnsMessageConverter.java

@@ -17,7 +17,7 @@
 
 import static io.awspring.cloud.sns.core.SnsHeaders.MESSAGE_DEDUPLICATION_ID_HEADER;
 import static io.awspring.cloud.sns.core.SnsHeaders.MESSAGE_GROUP_ID_HEADER;
-import static io.awspring.cloud.sns.core.SnsHeaders.MESSAGE_ID;
+import static io.awspring.cloud.sns.core.SnsHeaders.MESSAGE_ID_HEADER;
 
 import io.awspring.cloud.sns.core.SnsHeaderConverterUtil;
 import java.util.ArrayList;
@@ -92,7 +92,7 @@ private <T> void populateHeader(PublishBatchRequestEntry.Builder publishBatchReq
 			publishBatchRequestEntry.messageAttributes(messageAttributes);
 		}
 
-		String id = Optional.ofNullable(message.getHeaders().get(MESSAGE_ID, String.class)).filter(StringUtils::hasText)
+		String id = Optional.ofNullable(message.getHeaders().get(MESSAGE_ID_HEADER, String.class)).filter(StringUtils::hasText)
 				.orElseGet(() -> UUID.randomUUID().toString());
 		publishBatchRequestEntry.id(id);
 

…cutor/DefaultBatchExecutionStrategy.java …or/SequentialBatchExecutionStrategy.javaspring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/executor/DefaultBatchExecutionStrategy.java renamed to spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/executor/SequentialBatchExecutionStrategy.java spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/executor/DefaultBatchExecutionStrategy.java renamed to spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/executor/SequentialBatchExecutionStrategy.java

@@ -34,11 +34,13 @@
  * @author Matej Nedic
  * @since 4.0.1
  */
-public class DefaultBatchExecutionStrategy implements BatchExecutionStrategy {
+public class SequentialBatchExecutionStrategy implements BatchExecutionStrategy {
 
 	private final SnsClient snsClient;
 
-	public DefaultBatchExecutionStrategy(SnsClient snsClient) {
+	private static final int MAX_SNS_BATCH_SIZE = 10;
+
+	public SequentialBatchExecutionStrategy(SnsClient snsClient) {
 		Assert.notNull(snsClient, "SnsClient cannot be null!");
 		this.snsClient = snsClient;
 	}
@@ -52,14 +54,17 @@ public DefaultBatchExecutionStrategy(SnsClient snsClient) {
 	 */
 	@Override
 	public BatchResult send(Arn topicArn, Collection<PublishBatchRequestEntry> entries) {
+		Assert.notNull(topicArn, "topicArn is required");
+		Assert.notNull(topicArn, "entries are required");
+
 		List<BatchResult.SnsResult> allResults = new ArrayList<>();
 		List<BatchResult.SnsError> allErrors = new ArrayList<>();
-		List<PublishBatchRequestEntry> batch = new ArrayList<>(10);
+		List<PublishBatchRequestEntry> batch = new ArrayList<>(MAX_SNS_BATCH_SIZE);
 
 		for (PublishBatchRequestEntry entry : entries) {
 			batch.add(entry);
 
-			if (batch.size() == 10) {
+			if (batch.size() == MAX_SNS_BATCH_SIZE) {
 				processBatch(topicArn, batch, allResults, allErrors);
 				batch.clear();
 			}

spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/Person.java

@@ -15,6 +15,8 @@
  */
 package io.awspring.cloud.sns;
 
+import java.util.Objects;
+
 public class Person {
 	private String name;
 
@@ -32,4 +34,18 @@ public String getName() {
 	public void setName(String name) {
 		this.name = name;
 	}
+
+
+	@Override
+	public boolean equals(Object object) {
+		if (this == object) return true;
+		if (object == null || getClass() != object.getClass()) return false;
+		Person person = (Person) object;
+		return Objects.equals(name, person.name);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(name);
+	}
 }