Migrate Cassandra extension project

build.gradle

@@ -110,7 +110,7 @@ ext {
     springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '6.0.0-SNAPSHOT'
     springVersion = project.hasProperty('springVersion') ? project.springVersion : '6.0.0-SNAPSHOT'
     springWsVersion = '4.0.0-SNAPSHOT'
-    testcontainersVersion = '1.17.3'
+    testcontainersVersion = '1.17.5'
     tomcatVersion = '10.0.23'
     xmlUnitVersion = '2.9.0'
     xstreamVersion = '1.4.19'
@@ -497,6 +497,16 @@ project('spring-integration-camel') {
     }
 }
 
+project('spring-integration-cassandra') {
+    description = 'Spring Integration Support for Apache Cassandra'
+
+    dependencies {
+        api project(':spring-integration-core')
+        api 'org.springframework.data:spring-data-cassandra'
+
+        testImplementation 'org.testcontainers:cassandra'
+    }
+}
 
 project('spring-integration-core') {
     description = 'Spring Integration Core'

spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraNamespaceHandler.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 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 org.springframework.integration.cassandra.config.xml;
+
+import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
+
+/**
+ * The namespace handler for "int-cassandra" namespace.
+ *
+ * @author Artem Bilan
+ * @author Filippo Balicchia
+ *
+ * @since 6.0
+ */
+public class CassandraNamespaceHandler extends AbstractIntegrationNamespaceHandler {
+
+	@Override
+	public void init() {
+		registerBeanDefinitionParser("outbound-channel-adapter", new CassandraOutboundChannelAdapterParser());
+		registerBeanDefinitionParser("outbound-gateway", new CassandraOutboundGatewayParser());
+	}
+
+}

spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundChannelAdapterParser.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 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 org.springframework.integration.cassandra.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.cassandra.outbound.CassandraMessageHandler;
+import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
+
+/**
+ * The parser for the {@code <int-cassandra:outbound-channel-adapter>}.
+ *
+ * @author Filippo Balicchia
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class CassandraOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+
+	@Override
+	protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(CassandraMessageHandler.class);
+		builder.addPropertyValue("producesReply", false);
+		CassandraParserUtils.processOutboundTypeAttributes(element, parserContext, builder);
+		return builder.getBeanDefinition();
+	}
+
+}

spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundGatewayParser.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 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 org.springframework.integration.cassandra.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.cassandra.outbound.CassandraMessageHandler;
+import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+
+/**
+ * The parser for the {@code <int-cassandra:outbound-gateway>}.
+ *
+ * @author Filippo Balicchia
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class CassandraOutboundGatewayParser extends AbstractConsumerEndpointParser {
+
+
+	@Override
+	protected String getInputChannelAttributeName() {
+		return "request-channel";
+	}
+
+	@Override
+	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(CassandraMessageHandler.class);
+		builder.addPropertyValue("producesReply", true);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
+		CassandraParserUtils.processOutboundTypeAttributes(element, parserContext, builder);
+		return builder;
+	}
+
+}

spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraParserUtils.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright 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 org.springframework.integration.cassandra.config.xml;
+
+import java.util.List;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.ManagedMap;
+import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+import org.springframework.util.xml.DomUtils;
+
+/**
+ * The {@code int-cassandra} namespace XML parser helper.
+ *
+ * @author Filippo Balicchia
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public final class CassandraParserUtils {
+
+	public static void processOutboundTypeAttributes(Element element, ParserContext parserContext,
+			BeanDefinitionBuilder builder) {
+
+		String cassandraTemplate = element.getAttribute("cassandra-template");
+		String mode = element.getAttribute("mode");
+		String ingestQuery = element.getAttribute("ingest-query");
+		String query = element.getAttribute("query");
+
+		if (!StringUtils.hasText(cassandraTemplate)) {
+			parserContext.getReaderContext().error("cassandra-template is required", element);
+		}
+
+		builder.addConstructorArgReference(cassandraTemplate);
+		if (StringUtils.hasText(mode)) {
+			builder.addConstructorArgValue(mode);
+		}
+
+		BeanDefinition statementExpressionDef = IntegrationNamespaceUtils
+				.createExpressionDefIfAttributeDefined("statement-expression", element);
+
+		if (statementExpressionDef != null) {
+			builder.addPropertyValue("statementExpression", statementExpressionDef);
+		}
+
+		if (!areMutuallyExclusive(query, statementExpressionDef, ingestQuery)) {
+			parserContext.getReaderContext()
+					.error("'query', 'ingest-query', 'statement-expression' are mutually exclusive", element);
+		}
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "write-options");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ingest-query");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "query");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async");
+
+		List<Element> parameterExpressions = DomUtils.getChildElementsByTagName(element, "parameter-expression");
+		if (!CollectionUtils.isEmpty(parameterExpressions)) {
+			ManagedMap<String, Object> parameterExpressionsMap = new ManagedMap<>();
+			for (Element parameterExpressionElement : parameterExpressions) {
+				String name = parameterExpressionElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
+				BeanDefinition expression =
+						IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined(
+								IntegrationNamespaceUtils.EXPRESSION_ATTRIBUTE, parameterExpressionElement);
+				if (expression != null) {
+					parameterExpressionsMap.put(name, expression);
+				}
+			}
+			builder.addPropertyValue("parameterExpressions", parameterExpressionsMap);
+		}
+
+	}
+
+	public static boolean areMutuallyExclusive(String query, BeanDefinition statementExpressionDef,
+			String ingestQuery) {
+
+		return !StringUtils.hasText(query) && statementExpressionDef == null && !StringUtils.hasText(ingestQuery)
+				|| !(StringUtils.hasText(query) && statementExpressionDef != null && StringUtils.hasText(ingestQuery))
+				&& (StringUtils.hasText(query) ^ statementExpressionDef != null) ^ StringUtils.hasText(ingestQuery);
+	}
+
+	private CassandraParserUtils() {
+	}
+
+}

spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 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.
+ */
+
+/**
+ * Provides classes for Cassandra parsers and namespace handlers.
+ */
+package org.springframework.integration.cassandra.config.xml;

spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/dsl/Cassandra.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 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 org.springframework.integration.cassandra.dsl;
+
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+import org.springframework.integration.cassandra.outbound.CassandraMessageHandler;
+
+/**
+ * Factory class for Apache Cassandra components DSL.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public final class Cassandra {
+
+	/**
+	 * Create an instance of {@link CassandraMessageHandlerSpec} for the provided {@link ReactiveCassandraOperations}.
+	 * @param cassandraOperations the {@link ReactiveCassandraOperations} to use.
+	 * @return the spec.
+	 */
+	public static CassandraMessageHandlerSpec outboundChannelAdapter(ReactiveCassandraOperations cassandraOperations) {
+		return new CassandraMessageHandlerSpec(cassandraOperations);
+	}
+
+	/**
+	 * Create an instance of {@link CassandraMessageHandlerSpec} for the provided {@link ReactiveCassandraOperations}.
+	 * @param cassandraOperations the {@link ReactiveCassandraOperations} to use.
+	 * @param queryType the {@link CassandraMessageHandler.Type} to use.
+	 * @return the spec.
+	 */
+	public static CassandraMessageHandlerSpec outboundChannelAdapter(ReactiveCassandraOperations cassandraOperations,
+			CassandraMessageHandler.Type queryType) {
+
+		return new CassandraMessageHandlerSpec(cassandraOperations, queryType);
+	}
+
+	/**
+	 * Create an instance of {@link CassandraMessageHandlerSpec} for the provided {@link ReactiveCassandraOperations}
+	 * in an outbound gateway mode.
+	 * @param cassandraOperations the {@link ReactiveCassandraOperations} to use.
+	 * @return the spec.
+	 */
+	public static CassandraMessageHandlerSpec outboundGateway(ReactiveCassandraOperations cassandraOperations) {
+		return new CassandraMessageHandlerSpec(cassandraOperations)
+				.producesReply(true);
+	}
+
+	/**
+	 * Create an instance of {@link CassandraMessageHandlerSpec} for the provided {@link ReactiveCassandraOperations}
+	 * in an outbound gateway mode.
+	 * @param cassandraOperations the {@link ReactiveCassandraOperations} to use.
+	 * @param queryType the {@link CassandraMessageHandler.Type} to use.
+	 * @return the spec.
+	 */
+	public static CassandraMessageHandlerSpec outboundGateway(ReactiveCassandraOperations cassandraOperations,
+			CassandraMessageHandler.Type queryType) {
+
+		return new CassandraMessageHandlerSpec(cassandraOperations, queryType)
+				.producesReply(true);
+	}
+
+	private Cassandra() {
+	}
+
+}