SqsAutoConfiguration.java

/*
 * Copyright 2013-2024 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.autoconfigure.sqs;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.awspring.cloud.autoconfigure.AwsAsyncClientCustomizer;
import io.awspring.cloud.autoconfigure.core.AwsClientBuilderConfigurer;
import io.awspring.cloud.autoconfigure.core.AwsConnectionDetails;
import io.awspring.cloud.autoconfigure.core.CredentialsProviderAutoConfiguration;
import io.awspring.cloud.autoconfigure.core.RegionProviderAutoConfiguration;
import io.awspring.cloud.sqs.config.SqsBootstrapConfiguration;
import io.awspring.cloud.sqs.config.SqsListenerConfigurer;
import io.awspring.cloud.sqs.config.SqsMessageListenerContainerFactory;
import io.awspring.cloud.sqs.listener.SqsContainerOptionsBuilder;
import io.awspring.cloud.sqs.listener.acknowledgement.AcknowledgementResultCallback;
import io.awspring.cloud.sqs.listener.acknowledgement.AsyncAcknowledgementResultCallback;
import io.awspring.cloud.sqs.listener.errorhandler.AsyncErrorHandler;
import io.awspring.cloud.sqs.listener.errorhandler.ErrorHandler;
import io.awspring.cloud.sqs.listener.interceptor.AsyncMessageInterceptor;
import io.awspring.cloud.sqs.listener.interceptor.MessageInterceptor;
import io.awspring.cloud.sqs.operations.SqsTemplate;
import io.awspring.cloud.sqs.operations.SqsTemplateBuilder;
import io.awspring.cloud.sqs.support.converter.MessagingMessageConverter;
import io.awspring.cloud.sqs.support.converter.SqsMessagingMessageConverter;
import io.awspring.cloud.sqs.support.converter.legacy.JacksonJsonMessageConverterMigration;
import io.awspring.cloud.sqs.support.converter.legacy.JacksonMessageConverterMigration;
import io.awspring.cloud.sqs.support.converter.legacy.LegacyJackson2MessageConverterMigration;
import io.awspring.cloud.sqs.support.converter.legacy.LegacyJackson2SqsMessagingMessageConverter;
import io.awspring.cloud.sqs.support.observation.SqsListenerObservation;
import io.awspring.cloud.sqs.support.observation.SqsTemplateObservation;
import io.micrometer.observation.ObservationRegistry;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.Message;
import tools.jackson.databind.json.JsonMapper;

/**
 * {@link EnableAutoConfiguration Auto-configuration} for SQS integration.
 *
 * @author Tomaz Fernandes
 * @author Maciej Walkowiak
 * @author Wei Jiang
 * @author Dongha Kim
 * @since 3.0
 */
@AutoConfiguration
@ConditionalOnClass({ SqsAsyncClient.class, SqsBootstrapConfiguration.class })
@EnableConfigurationProperties(SqsProperties.class)
@Import(SqsBootstrapConfiguration.class)
@AutoConfigureAfter({ CredentialsProviderAutoConfiguration.class, RegionProviderAutoConfiguration.class })
@ConditionalOnProperty(name = "spring.cloud.aws.sqs.enabled", havingValue = "true", matchIfMissing = true)
public class SqsAutoConfiguration {

	private final SqsProperties sqsProperties;

	public SqsAutoConfiguration(SqsProperties sqsProperties) {
		this.sqsProperties = sqsProperties;
	}

	@ConditionalOnMissingBean
	@Bean
	public SqsAsyncClient sqsAsyncClient(AwsClientBuilderConfigurer awsClientBuilderConfigurer,
			ObjectProvider<AwsConnectionDetails> connectionDetails,
			ObjectProvider<SqsAsyncClientCustomizer> sqsAsyncClientCustomizers,
			ObjectProvider<AwsAsyncClientCustomizer> awsAsyncClientCustomizers) {
		return awsClientBuilderConfigurer
				.configureAsyncClient(SqsAsyncClient.builder(), this.sqsProperties, connectionDetails.getIfAvailable(),
						sqsAsyncClientCustomizers.orderedStream(), awsAsyncClientCustomizers.orderedStream())
				.build();
	}

	@ConditionalOnMissingBean
	@Bean
	public SqsTemplate sqsTemplate(SqsAsyncClient sqsAsyncClient,
			ObjectProvider<JacksonMessageConverterMigration> messageConverterFactory,
			ObjectProvider<ObservationRegistry> observationRegistryProvider,
			ObjectProvider<SqsTemplateObservation.Convention> observationConventionProvider,
			MessagingMessageConverter<Message> messageConverter) {
		SqsTemplateBuilder builder = SqsTemplate.builder().sqsAsyncClient(sqsAsyncClient)
				.messageConverter(messageConverter);
		messageConverterFactory.ifAvailable(mcf -> mcf.configureLegacyObjectMapper(messageConverter));
		if (this.sqsProperties.isObservationEnabled()) {
			observationRegistryProvider
					.ifAvailable(registry -> builder.configure(options -> options.observationRegistry(registry)));
			observationConventionProvider
					.ifAvailable(convention -> builder.configure(options -> options.observationConvention(convention)));
		}
		if (sqsProperties.getQueueNotFoundStrategy() != null) {
			builder.configure((options) -> options.queueNotFoundStrategy(sqsProperties.getQueueNotFoundStrategy()));
		}
		return builder.build();
	}

	@ConditionalOnMissingBean
	@Bean
	public SqsMessageListenerContainerFactory<Object> defaultSqsListenerContainerFactory(
			ObjectProvider<SqsAsyncClient> sqsAsyncClient, ObjectProvider<AsyncErrorHandler<Object>> asyncErrorHandler,
			ObjectProvider<ErrorHandler<Object>> errorHandler,
			ObjectProvider<AsyncMessageInterceptor<Object>> asyncInterceptors,
			ObjectProvider<ObservationRegistry> observationRegistry,
			ObjectProvider<SqsListenerObservation.Convention> observationConventionProvider,
			ObjectProvider<MessageInterceptor<Object>> interceptors,
			ObjectProvider<AcknowledgementResultCallback<Object>> acknowledgementResultCallback,
			ObjectProvider<AsyncAcknowledgementResultCallback<Object>> asyncAcknowledgementResultCallback,
			ObjectProvider<JacksonMessageConverterMigration> messageConverterFactory,
			MessagingMessageConverter<?> messagingMessageConverter) {

		SqsMessageListenerContainerFactory<Object> factory = new SqsMessageListenerContainerFactory<>();
		factory.configure(this::configureProperties);
		sqsAsyncClient.ifAvailable(factory::setSqsAsyncClient);
		asyncErrorHandler.ifAvailable(factory::setErrorHandler);
		errorHandler.ifAvailable(factory::setErrorHandler);
		interceptors.forEach(factory::addMessageInterceptor);
		asyncInterceptors.forEach(factory::addMessageInterceptor);
		acknowledgementResultCallback.ifAvailable(factory::setAcknowledgementResultCallback);
		asyncAcknowledgementResultCallback.ifAvailable(factory::setAcknowledgementResultCallback);
		messageConverterFactory.ifAvailable(mcf -> mcf.configureLegacyObjectMapper(messagingMessageConverter));
		if (this.sqsProperties.isObservationEnabled()) {
			observationRegistry
					.ifAvailable(registry -> factory.configure(options -> options.observationRegistry(registry)));
			observationConventionProvider
					.ifAvailable(convention -> factory.configure(options -> options.observationConvention(convention)));
		}
		factory.configure(options -> options.messageConverter(messagingMessageConverter));
		return factory;
	}

	private void configureProperties(SqsContainerOptionsBuilder options) {
		PropertyMapper mapper = PropertyMapper.get();
		mapper.from(this.sqsProperties.getQueueNotFoundStrategy()).to(options::queueNotFoundStrategy);
		mapper.from(this.sqsProperties.getListener().getMaxConcurrentMessages()).to(options::maxConcurrentMessages);
		mapper.from(this.sqsProperties.getListener().getMaxMessagesPerPoll()).to(options::maxMessagesPerPoll);
		mapper.from(this.sqsProperties.getListener().getPollTimeout()).to(options::pollTimeout);
		mapper.from(this.sqsProperties.getListener().getMaxDelayBetweenPolls()).to(options::maxDelayBetweenPolls);
		mapper.from(this.sqsProperties.getListener().getAutoStartup()).to(options::autoStartup);
	}

	@ConditionalOnClass(name = "tools.jackson.databind.json.JsonMapper")
	@Configuration
	static class SqsJacksonConfiguration {
		@ConditionalOnMissingBean
		@Bean
		public MessagingMessageConverter<Message> messageConverter(ObjectProvider<JsonMapper> jsonMapperProvider) {
			JsonMapper jsonMapper = jsonMapperProvider.getIfAvailable();
			return jsonMapper != null ? new SqsMessagingMessageConverter(jsonMapper)
					: new SqsMessagingMessageConverter();
		}

		@Bean
		@ConditionalOnMissingBean
		public JacksonMessageConverterMigration jsonMapperWrapper(ObjectProvider<JsonMapper> jsonMapper) {
			JsonMapper mapper = jsonMapper.getIfAvailable(JsonMapper::new);
			return new JacksonJsonMessageConverterMigration(mapper);
		}
	}

	@ConditionalOnClass(name = "com.fasterxml.jackson.databind.ObjectMapper")
	@ConditionalOnMissingClass("tools.jackson.databind.json.JsonMapper")
	@Configuration
	static class LegacySqsJackson2Configuration {
		@ConditionalOnMissingBean
		@Bean
		public MessagingMessageConverter<Message> messageConverter() {
			return new LegacyJackson2SqsMessagingMessageConverter();
		}

		@Bean
		@ConditionalOnMissingBean
		public JacksonMessageConverterMigration legacyJackson2messageConverter(
				ObjectProvider<ObjectMapper> objectMapper) {
			ObjectMapper mapper = objectMapper.getIfAvailable(ObjectMapper::new);
			return new LegacyJackson2MessageConverterMigration(mapper);
		}
	}

	@Bean
	public SqsListenerConfigurer jacksonMessageConverterMigrationCustomizer(
			ObjectProvider<JacksonMessageConverterMigration> factoryProvider) {
		JacksonMessageConverterMigration factory = factoryProvider.getIfUnique();
		return registrar -> {
			if (factory != null && registrar.getJacksonMessageConverterMigration() == null) {
				registrar.setJacksonMessageConverterMigration(factory);
			}
		};
	}

}