Gh 344 fix tests

pom.xml

@@ -48,6 +48,8 @@
 		<module>spring-cloud-aws-starters</module>
 		<module>spring-cloud-aws-samples</module>
 		<module>docs</module>
+        <module>spring-cloud-aws-messaging-support</module>
+		<module>spring-cloud-aws-sqs</module>
 
 		<!--
 		these modules need to be rewritten

spring-cloud-aws-messaging-support/pom.xml

@@ -0,0 +1,65 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<parent>
+		<artifactId>spring-cloud-aws</artifactId>
+		<groupId>io.awspring.cloud</groupId>
+		<version>3.0.0-SNAPSHOT</version>
+	</parent>
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-aws-messaging-support</artifactId>
+	<name>Spring Cloud AWS Messaging Support</name>
+	<description>Spring Cloud Support for AWS Messaging Services</description>
+
+	<properties>
+		<maven.compiler.source>8</maven.compiler.source>
+		<maven.compiler.target>8</maven.compiler.target>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>software.amazon.awssdk</groupId>
+			<artifactId>arns</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-messaging</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-context</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.awspring.cloud</groupId>
+			<artifactId>spring-cloud-aws-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>localstack</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<!-- AWS SDK v1 is required by testcontainers-localstack -->
+		<dependency>
+			<groupId>com.amazonaws</groupId>
+			<artifactId>aws-java-sdk-core</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+</project>

spring-cloud-aws-messaging-support/src/main/java/io/awspring/cloud/messaging/support/MessagingUtils.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.messaging.support;
+
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+/**
+ * @author Tomaz Fernandes
+ * @since 3.0
+ */
+public class MessagingUtils {
+
+	public static final MessagingUtils INSTANCE = new MessagingUtils();
+
+	private MessagingUtils() {
+	}
+
+	public <T> MessagingUtils acceptIfNotNull(T value, Consumer<T> consumer) {
+		if (value != null) {
+			consumer.accept(value);
+		}
+		return this;
+	}
+
+	public <T, V> MessagingUtils acceptBothIfNoneNull(T firstValue, V secondValue, BiConsumer<T, V> consumer) {
+		if (firstValue != null && secondValue != null) {
+			consumer.accept(firstValue, secondValue);
+		}
+		return this;
+	}
+
+	public <T> MessagingUtils acceptFirstNonNull(Consumer<T> consumer, T... values) {
+		Arrays.stream(values).filter(Objects::nonNull).findFirst().ifPresent(consumer);
+		return this;
+	}
+}

spring-cloud-aws-messaging-support/src/main/java/io/awspring/cloud/messaging/support/config/AbstractFactoryOptions.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.messaging.support.config;
+
+import io.awspring.cloud.messaging.support.listener.AsyncErrorHandler;
+import io.awspring.cloud.messaging.support.listener.AsyncMessageInterceptor;
+import io.awspring.cloud.messaging.support.listener.acknowledgement.AsyncAckHandler;
+
+/**
+ * @author Tomaz Fernandes
+ * @since 3.0
+ */
+public abstract class AbstractFactoryOptions<T, O extends AbstractFactoryOptions<?, ?>> implements FactoryOptions {
+
+	private AsyncErrorHandler<T> errorHandler;
+
+	private AsyncAckHandler<T> ackHandler;
+
+	private AsyncMessageInterceptor<T> messageInterceptor;
+
+	private Integer maxWorksPerContainer;
+
+	Integer getMaxWorkersPerContainer() {
+		return this.maxWorksPerContainer;
+	}
+
+	AsyncErrorHandler<T> getErrorHandler() {
+		return this.errorHandler;
+	}
+
+	AsyncAckHandler<T> getAckHandler() {
+		return this.ackHandler;
+	}
+
+	AsyncMessageInterceptor<T> getMessageInterceptor() {
+		return this.messageInterceptor;
+	}
+
+	@SuppressWarnings("unchecked")
+	private O self() {
+		return (O) this;
+	}
+
+	public O errorHandler(AsyncErrorHandler<T> errorHandler) {
+		this.errorHandler = errorHandler;
+		return self();
+	}
+
+	public O ackHandler(AsyncAckHandler<T> ackHandler) {
+		this.ackHandler = ackHandler;
+		return self();
+	}
+
+	public O interceptor(AsyncMessageInterceptor<T> messageInterceptor) {
+		this.messageInterceptor = messageInterceptor;
+		return self();
+	}
+
+	public O concurrentWorkersPerContainer(int maxWorksPerContainer) {
+		this.maxWorksPerContainer = maxWorksPerContainer;
+		return self();
+	}
+}

spring-cloud-aws-messaging-support/src/main/java/io/awspring/cloud/messaging/support/config/AbstractMessageListenerContainerFactory.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.messaging.support.config;
+
+import io.awspring.cloud.messaging.support.MessagingUtils;
+import io.awspring.cloud.messaging.support.endpoint.Endpoint;
+import io.awspring.cloud.messaging.support.listener.AbstractMessageListenerContainer;
+import io.awspring.cloud.messaging.support.listener.AsyncMessageListener;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * @author Tomaz Fernandes
+ * @since 3.0
+ */
+public abstract class AbstractMessageListenerContainerFactory<T, C extends AbstractMessageListenerContainer<T>, E extends Endpoint>
+		implements MessageListenerContainerFactory<C, E>, SmartInitializingSingleton, BeanFactoryAware {
+
+	private static final Integer DEFAULT_THREADPOOL_SIZE = 11;
+	private final AbstractFactoryOptions<T, ?> factoryOptions;
+	private BeanFactory beanFactory;
+	private AsyncMessageListener<T> messageListener;
+
+	protected AbstractMessageListenerContainerFactory(AbstractFactoryOptions<T, ?> factoryOptions) {
+		this.factoryOptions = factoryOptions;
+	}
+
+	@Override
+	public C create(E endpoint) {
+		C container = createContainerInstance(endpoint);
+		MessagingUtils.INSTANCE.acceptIfNotNull(this.factoryOptions.getErrorHandler(), container::setErrorHandler)
+				.acceptIfNotNull(this.factoryOptions.getAckHandler(), container::setAckHandler)
+				.acceptIfNotNull(this.factoryOptions.getMessageInterceptor(), container::setMessageInterceptor);
+		initializeContainer(container);
+		return container;
+	}
+
+	protected abstract C createContainerInstance(E endpoint);
+
+	protected void initializeContainer(C container) {
+	}
+
+	protected ThreadPoolTaskExecutor createTaskExecutor() {
+		ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
+		int poolSize = this.factoryOptions.getMaxWorkersPerContainer() != null
+				? this.factoryOptions.getMaxWorkersPerContainer() + 1
+				: DEFAULT_THREADPOOL_SIZE;
+		taskExecutor.setMaxPoolSize(poolSize);
+		taskExecutor.setCorePoolSize(poolSize);
+		return taskExecutor;
+	}
+
+	public void setMessageListener(AsyncMessageListener<T> messageListener) {
+		Assert.notNull(messageListener, "messageListener cannot be null");
+		this.messageListener = messageListener;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	protected BeanFactory getBeanFactory() {
+		return this.beanFactory;
+	}
+
+	protected AsyncMessageListener<T> getMessageListener() {
+		return this.messageListener;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void afterSingletonsInstantiated() {
+		if (this.messageListener == null) {
+			Assert.isTrue(this.beanFactory.containsBean(MessagingConfigUtils.MESSAGE_LISTENER_BEAN_NAME),
+					"An AsyncMessageListener must be registered with name "
+							+ MessagingConfigUtils.MESSAGE_LISTENER_BEAN_NAME);
+			this.messageListener = this.beanFactory.getBean(MessagingConfigUtils.MESSAGE_LISTENER_BEAN_NAME,
+					AsyncMessageListener.class);
+		}
+		initializeFactory();
+	}
+
+	protected void initializeFactory() {
+	}
+}

spring-cloud-aws-messaging-support/src/main/java/io/awspring/cloud/messaging/support/config/FactoryOptions.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.messaging.support.config;
+
+/**
+ * @author Tomaz Fernandes
+ * @since 3.0
+ */
+public interface FactoryOptions {
+
+}

spring-cloud-aws-messaging-support/src/main/java/io/awspring/cloud/messaging/support/config/MessageListenerContainerFactory.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.messaging.support.config;
+
+import io.awspring.cloud.messaging.support.endpoint.Endpoint;
+import io.awspring.cloud.messaging.support.listener.MessageListenerContainer;
+
+/**
+ * @author Tomaz Fernandes
+ * @since 3.0
+ */
+public interface MessageListenerContainerFactory<C extends MessageListenerContainer, E extends Endpoint> {
+
+	C create(E endpoint);
+
+}