feat: SqsMessageDrivenChannelAdapterParser

Author: boolivar

int-aws-support/src/main/java/org/springframework/integration/aws/config/xml/SqsContainerOptionsFactoryBean.java

@@ -0,0 +1,63 @@
+package org.springframework.integration.aws.config.xml;
+
+import io.awspring.cloud.sqs.listener.QueueNotFoundStrategy;
+import io.awspring.cloud.sqs.listener.SqsContainerOptions;
+import io.awspring.cloud.sqs.listener.SqsContainerOptionsBuilder;
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.AcknowledgementMode;
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.core.task.TaskExecutor;
+
+import java.time.Duration;
+
+class SqsContainerOptionsFactoryBean implements FactoryBean<SqsContainerOptions> {
+
+    private final SqsContainerOptionsBuilder builder = SqsContainerOptions.builder();
+
+    @Override
+    public boolean isSingleton() {
+        return false;
+    }
+
+    @Override
+    public Class<SqsContainerOptions> getObjectType() {
+        return SqsContainerOptions.class;
+    }
+
+    @Override
+    public SqsContainerOptions getObject() {
+        return builder.build();
+    }
+
+    public void setVisibilityTimeout(long seconds) {
+        builder.messageVisibility(Duration.ofSeconds(seconds));
+    }
+
+    public void setMaxNumberOfMessages(int maxMessages) {
+        builder.maxConcurrentMessages(maxMessages)
+            .maxMessagesPerPoll(maxMessages);
+    }
+
+    public void setTaskExecutor(TaskExecutor taskExecutor) {
+        builder.componentsTaskExecutor(taskExecutor);
+    }
+
+    public void setQueueStopTimeout(long millis) {
+        builder.listenerShutdownTimeout(Duration.ofMillis(millis));
+    }
+
+    public void setWaitTimeOut(long seconds) {
+        builder.pollTimeout(Duration.ofSeconds(seconds));
+    }
+
+    public void setMessageDeletionPolicy(String policy) {
+        builder.acknowledgementMode("NO_REDRIVE".equals(policy)
+            ? AcknowledgementMode.ON_SUCCESS
+            : AcknowledgementMode.valueOf(policy));
+    }
+
+    public void setFailOnMissingQueue(boolean fail) {
+        builder.queueNotFoundStrategy(fail
+            ? QueueNotFoundStrategy.FAIL
+            : QueueNotFoundStrategy.CREATE);
+    }
+}

int-aws-support/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java

@@ -0,0 +1,39 @@
+package org.springframework.integration.aws.config.xml;
+
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.aws.inbound.SqsMessageDrivenChannelAdapter;
+import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
+import org.w3c.dom.Element;
+
+public class SqsMessageDrivenChannelAdapterParser extends AbstractChannelAdapterParser {
+
+    @Override
+    protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
+        return XmlBeanDefinitionBuilder.newInstance(element, parserContext, SqsMessageDrivenChannelAdapter.class)
+            .unsupportedAttributeWarning("resource-id-resolver")
+            .addConstructorArgReference("sqs")
+            .addConstructorArgValue("queues")
+            .setPropertyValueIfAttributeDefined("send-timeout")
+            .setPropertyValueIfAttributeDefined("errorChannelName", "error-channel")
+            .setPropertyReferenceIfAttributeDefined("channelResolver", "destination-resolver")
+            .getBeanDefinitionBuilder()
+            .addPropertyValue("outputChannelName", channelName)
+            .addPropertyValue("sqsContainerOptions", sqsContainerOptions(element, parserContext))
+            .getBeanDefinition();
+    }
+
+    private AbstractBeanDefinition sqsContainerOptions(Element element, ParserContext parserContext) {
+        return XmlBeanDefinitionBuilder.newInstance(element, parserContext, SqsContainerOptionsFactoryBean.class)
+            .setPropertyValueIfAttributeDefined("max-number-of-messages")
+            .setPropertyValueIfAttributeDefined("visibility-timeout")
+            .setPropertyValueIfAttributeDefined("queue-stop-timeout")
+            .setPropertyValueIfAttributeDefined("wait-time-out")
+            .setPropertyValueIfAttributeDefined("message-deletion-policy")
+            .setPropertyValueIfAttributeDefined("fail-on-missing-queue")
+            .setPropertyReferenceIfAttributeDefined("task-executor")
+            .getBeanDefinitionBuilder()
+            .applyCustomizers(bean -> bean.setAutowireCandidate(false))
+            .getBeanDefinition();
+    }
+}

int-aws-support/src/main/resources/org/springframework/integration/aws/config/xml/element-parser.mapping

@@ -1 +1,2 @@
-sqs-outbound-channel-adapter: org.springframework.integration.aws.config.xml.SqsOutboundChannelAdapterParser
+sqs-outbound-channel-adapter: org.springframework.integration.aws.config.xml.SqsOutboundChannelAdapterParser
+sqs-message-driven-channel-adapter: org.springframework.integration.aws.config.xml.SqsMessageDrivenChannelAdapterParser

int-aws-support/src/test/java/org/springframework/integration/aws/config/xml/SpringIntegrationAwsNamespaceHandlerTest.java

@@ -15,6 +15,12 @@ void test() {
         handler.init();
         assertThat(handler)
             .extracting("parsers", InstanceOfAssertFactories.MAP)
-            .hasEntrySatisfying("sqs-outbound-channel-adapter", new Condition<>(SqsOutboundChannelAdapterParser.class::isInstance, "instance of SqsOutboundChannelAdapterParser"));
+            .hasEntrySatisfying("sqs-outbound-channel-adapter", isInstanceOf(SqsOutboundChannelAdapterParser.class))
+            .hasEntrySatisfying("sqs-message-driven-channel-adapter", isInstanceOf(SqsMessageDrivenChannelAdapterParser.class))
+            ;
+    }
+
+    private Condition<Object> isInstanceOf(Class<?> type) {
+        return new Condition<>(type::isInstance, "instance of " + type);
     }
 }