Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Migrate Cassandra extension project (#3913)

Browse Source 
* Migrate Cassandra extension project

* Add `spring-integration-cassandra` module based on the extension project
* Add Java DSL for Cassandra module
* Add documentation
* Add `CassandraContainerTest` based on a Testcontainers

* Fix language in docs

Co-authored-by: Gary Russell <grussell@vmware.com>

* * Fix `reactive-streams.adoc` for a proper link to
the new `spring-integration-cassandra` module

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2022-10-17 16:03:37 -04:00
committed by GitHub
parent 83f2a9c246
commit a08713fe87
34 changed files with 2357 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
build.gradle
View File

@@ -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'
@@ -498,6 +498,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'

37
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraNamespaceHandler.java Normal file
View File

@@ -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());
}
}

45
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundChannelAdapterParser.java Normal file
View File

@@ -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();
}
}

52
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundGatewayParser.java Normal file
View File

@@ -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;
}
}

105
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraParserUtils.java Normal file
View File

@@ -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() {
}
}

20
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/package-info.java Normal file
View File

@@ -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;

80
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/dsl/Cassandra.java Normal file
View File

@@ -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() {
}
}

160
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/dsl/CassandraMessageHandlerSpec.java Normal file
View File

@@ -0,0 +1,160 @@
/*
* 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 java.util.HashMap;
import java.util.Map;
import java.util.function.Function;
import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
import org.springframework.data.cassandra.core.cql.WriteOptions;
import org.springframework.expression.Expression;
import org.springframework.integration.cassandra.outbound.CassandraMessageHandler;
import org.springframework.integration.dsl.MessageHandlerSpec;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.messaging.Message;
import com.datastax.oss.driver.api.core.cql.Statement;
/**
* The {@link MessageHandlerSpec} for {@link CassandraMessageHandler}.
*
* @author Artem Bilan
*
* @since 6.0
*/
public class CassandraMessageHandlerSpec extends
MessageHandlerSpec<CassandraMessageHandlerSpec, CassandraMessageHandler> {
private final Map<String, Expression> parameterExpressions = new HashMap<>();
protected CassandraMessageHandlerSpec(ReactiveCassandraOperations cassandraOperations) {
this.target = new CassandraMessageHandler(cassandraOperations);
}
protected CassandraMessageHandlerSpec(ReactiveCassandraOperations cassandraOperations,
CassandraMessageHandler.Type queryType) {
this.target = new CassandraMessageHandler(cassandraOperations, queryType);
}
protected CassandraMessageHandlerSpec producesReply(boolean producesReply) {
this.target.setProducesReply(producesReply);
return _this();
}
/**
* Set an ingest query.
* @param ingestQuery ingest query to use.
* @return this spec
*/
public CassandraMessageHandlerSpec ingestQuery(String ingestQuery) {
this.target.setIngestQuery(ingestQuery);
return _this();
}
/**
* Set a {@link WriteOptions} for {@code INSERT}, {@code UPDATE} or {@code DELETE} operations.
* @param writeOptions the {@link WriteOptions} to use.
* @return this spec
*/
public CassandraMessageHandlerSpec writeOptions(WriteOptions writeOptions) {
this.target.setWriteOptions(writeOptions);
return _this();
}
/**
* Set a SpEL expression to evaluate a {@link Statement} against request message.
* @param statementExpression the SpEL expression to use.
* @return this spec
*/
public CassandraMessageHandlerSpec statementExpression(String statementExpression) {
return statementExpression(PARSER.parseExpression(statementExpression));
}
/**
* Set a SpEL expression to evaluate a {@link Statement} against request message.
* @param statementExpression the SpEL expression to use.
* @return this spec
*/
public CassandraMessageHandlerSpec statementExpression(Expression statementExpression) {
this.target.setStatementExpression(statementExpression);
return _this();
}
/**
* Set a {@link Function} to evaluate a {@link Statement} against request message.
* @param statementFunction the function to use.
* @return this spec
*/
public CassandraMessageHandlerSpec statementFunction(Function<Message<?>, Statement<?>> statementFunction) {
this.target.setStatementProcessor(statementFunction::apply);
return _this();
}
/**
* Set a {@code SELECT} query.
* @param query the CQL query to execute
* @return this spec
*/
public CassandraMessageHandlerSpec query(String query) {
this.target.setQuery(query);
return _this();
}
/**
* Set a map for named parameters and expressions for their values against a request message.
* @param parameterExpressions the map to use.
* @return this spec
*/
public CassandraMessageHandlerSpec parameterExpressions(Map<String, Expression> parameterExpressions) {
this.target.setParameterExpressions(parameterExpressions);
return _this();
}
/**
* Add a named bindable parameter with a SpEL expression to evaluate its value against a request message.
* @param name the name of parameter.
* @param expression the SpEL expression for parameter value.
* @return this spec
*/
public CassandraMessageHandlerSpec parameter(String name, String expression) {
return parameter(name, PARSER.parseExpression(expression));
}
/**
* Add a named bindable parameter with a function to evaluate its value against a request message.
* @param name the name of parameter.
* @param function the function for parameter value.
* @return this spec
*/
public CassandraMessageHandlerSpec parameter(String name, Function<Message<?>, ?> function) {
return parameter(name, new FunctionExpression<>(function));
}
/**
* Add a named bindable parameter with a SpEL expression to evaluate its value against a request message.
* @param name the name of parameter.
* @param expression the SpEL expression for parameter value.
* @return this spec
*/
public CassandraMessageHandlerSpec parameter(String name, Expression expression) {
this.parameterExpressions.put(name, expression);
return parameterExpressions(this.parameterExpressions);
}
}

4
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/dsl/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides Apache Cassandra Components support for the Java DSL.
*/
package org.springframework.integration.cassandra.dsl;

359
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandler.java Normal file
View File

@@ -0,0 +1,359 @@
/*
* 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.outbound;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.StreamSupport;
import org.springframework.dao.DataAccessException;
import org.springframework.data.cassandra.ReactiveResultSet;
import org.springframework.data.cassandra.ReactiveSession;
import org.springframework.data.cassandra.core.InsertOptions;
import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
import org.springframework.data.cassandra.core.UpdateOptions;
import org.springframework.data.cassandra.core.WriteResult;
import org.springframework.data.cassandra.core.cql.QueryOptionsUtil;
import org.springframework.data.cassandra.core.cql.ReactiveSessionCallback;
import org.springframework.data.cassandra.core.cql.WriteOptions;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.TypeLocator;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.expression.spel.support.StandardTypeLocator;
import org.springframework.integration.expression.ExpressionEvalMap;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import com.datastax.oss.driver.api.core.DriverException;
import com.datastax.oss.driver.api.core.cql.BatchStatementBuilder;
import com.datastax.oss.driver.api.core.cql.BatchType;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.cql.Statement;
import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* An {@link AbstractReplyProducingMessageHandler} implementation for Cassandra outbound operations.
*
* @author Soby Chacko
* @author Artem Bilan
* @author Filippo Balicchia
*
* @since 6.0
*/
public class CassandraMessageHandler extends AbstractReplyProducingMessageHandler {
private final Map<String, Expression> parameterExpressions = new HashMap<>();
private Type mode;
private final ReactiveCassandraOperations cassandraOperations;
private boolean producesReply;
/**
* Prepared statement to use in association with high throughput ingestion.
*/
private String ingestQuery;
/**
* Various options that can be used for Cassandra writes.
*/
private WriteOptions writeOptions = WriteOptions.empty();
private ReactiveSessionMessageCallback sessionMessageCallback;
private EvaluationContext evaluationContext;
public CassandraMessageHandler(ReactiveCassandraOperations cassandraOperations) {
this(cassandraOperations, Type.INSERT);
}
public CassandraMessageHandler(ReactiveCassandraOperations cassandraOperations,
CassandraMessageHandler.Type queryType) {
Assert.notNull(cassandraOperations, "'cassandraOperations' must not be null.");
Assert.notNull(queryType, "'queryType' must not be null.");
this.cassandraOperations = cassandraOperations;
this.mode = queryType;
setAsync(true);
switch (this.mode) {
case INSERT:
this.writeOptions = InsertOptions.empty();
break;
case UPDATE:
this.writeOptions = UpdateOptions.empty();
break;
}
}
public void setIngestQuery(String ingestQuery) {
Assert.hasText(ingestQuery, "'ingestQuery' must not be empty");
this.ingestQuery = ingestQuery;
this.mode = Type.INSERT;
}
public void setWriteOptions(WriteOptions writeOptions) {
Assert.notNull(writeOptions, "'writeOptions' must not be null");
this.writeOptions = writeOptions;
}
public void setProducesReply(boolean producesReply) {
this.producesReply = producesReply;
}
public void setStatementExpressionString(String statementExpression) {
setStatementExpression(EXPRESSION_PARSER.parseExpression(statementExpression));
}
@SuppressWarnings("unchecked")
public void setStatementExpression(Expression statementExpression) {
ExpressionEvaluatingMessageProcessor<?> expressionEvaluatingMessageProcessor =
new ExpressionEvaluatingMessageProcessor<>(statementExpression, Statement.class) {
@Override
protected StandardEvaluationContext getEvaluationContext() {
return (StandardEvaluationContext) CassandraMessageHandler.this.evaluationContext;
}
};
setStatementProcessor((ExpressionEvaluatingMessageProcessor<Statement<?>>) expressionEvaluatingMessageProcessor);
}
public void setQuery(String query) {
Assert.hasText(query, "'query' must not be empty");
this.sessionMessageCallback =
(session, requestMessage) ->
session.execute(query,
ExpressionEvalMap.from(this.parameterExpressions)
.usingEvaluationContext(this.evaluationContext)
.withRoot(requestMessage)
.build());
this.mode = Type.STATEMENT;
}
public void setParameterExpressions(Map<String, Expression> parameterExpressions) {
Assert.notEmpty(parameterExpressions, "'parameterExpressions' must not be empty.");
this.parameterExpressions.clear();
this.parameterExpressions.putAll(parameterExpressions);
}
public void setStatementProcessor(MessageProcessor<Statement<?>> statementProcessor) {
Assert.notNull(statementProcessor, "'statementProcessor' must not be null.");
this.sessionMessageCallback =
(session, requestMessage) ->
session.execute(
QueryOptionsUtil.addQueryOptions(statementProcessor.processMessage(requestMessage),
this.writeOptions));
this.mode = Type.STATEMENT;
}
@Override
public String getComponentType() {
return "cassandra:outbound-" + (this.producesReply ? "gateway" : "channel-adapter");
}
@Override
protected void doInit() {
super.doInit();
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
TypeLocator typeLocator = this.evaluationContext.getTypeLocator();
if (typeLocator instanceof StandardTypeLocator) {
/*
* Register the Cassandra Query DSL package so they don't need a FQCN for QueryBuilder, for example.
*/
((StandardTypeLocator) typeLocator).registerImport(QueryBuilder.class.getPackage().getName());
}
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
Object payload = requestMessage.getPayload();
Mono<? extends WriteResult> result = null;
Type mode = this.mode;
if (payload instanceof Statement) {
mode = Type.STATEMENT;
}
switch (mode) {
case INSERT:
result = handleInsert(payload);
break;
case UPDATE:
result = handleUpdate(payload);
break;
case DELETE:
result = handleDelete(payload);
break;
case STATEMENT:
result = handleStatement(requestMessage);
break;
}
if (this.producesReply) {
return isAsync() ? result : result.block();
}
else {
if (isAsync()) {
result.subscribe();
}
else {
result.block();
}
return null;
}
}
@SuppressWarnings("unchecked")
private Mono<? extends WriteResult> handleInsert(Object payload) {
if (this.ingestQuery != null) {
Assert.isInstanceOf(List.class, payload,
"to perform 'ingest' the 'payload' must be of 'List<List<?>>' type.");
List<?> list = (List<?>) payload;
for (Object o : list) {
Assert.isInstanceOf(List.class, o,
"to perform 'ingest' the 'payload' must be of 'List<List<?>>' type.");
}
List<List<?>> rows = (List<List<?>>) payload;
return this.cassandraOperations.getReactiveCqlOperations()
.execute((ReactiveSessionCallback<WriteResult>) session ->
session.prepare(
QueryOptionsUtil.addQueryOptions(SimpleStatement.newInstance(this.ingestQuery),
this.writeOptions))
.flatMapMany(s ->
Flux.fromIterable(rows)
.map(row -> s.bind(row.toArray())))
.collect(() -> new BatchStatementBuilder(BatchType.UNLOGGED),
BatchStatementBuilder::addStatement)
.map(BatchStatementBuilder::build)
.flatMap(session::execute)
.transform(this::transformToWriteResult))
.next();
}
else {
if (payload instanceof List) {
return this.cassandraOperations.batchOps()
.insert((List<?>) payload, this.writeOptions)
.execute();
}
else {
return this.cassandraOperations.insert(payload, (InsertOptions) this.writeOptions);
}
}
}
private Mono<? extends WriteResult> handleUpdate(Object payload) {
if (payload instanceof List) {
return this.cassandraOperations.batchOps()
.update((List<?>) payload, this.writeOptions)
.execute();
}
else {
return this.cassandraOperations.update(payload, (UpdateOptions) this.writeOptions);
}
}
private Mono<WriteResult> handleDelete(Object payload) {
if (payload instanceof List) {
return this.cassandraOperations.batchOps()
.delete((List<?>) payload)
.execute();
}
else {
return this.cassandraOperations.delete(payload, this.writeOptions);
}
}
private Mono<WriteResult> handleStatement(Message<?> requestMessage) {
Object payload = requestMessage.getPayload();
Mono<ReactiveResultSet> resultSetMono;
if (payload instanceof Statement) {
resultSetMono = this.cassandraOperations.getReactiveCqlOperations()
.queryForResultSet((Statement<?>) payload);
}
else {
resultSetMono = this.cassandraOperations.getReactiveCqlOperations()
.execute((ReactiveSessionCallback<ReactiveResultSet>) session ->
this.sessionMessageCallback.doInSession(session, requestMessage))
.next();
}
return resultSetMono.transform(this::transformToWriteResult);
}
private Mono<WriteResult> transformToWriteResult(Mono<ReactiveResultSet> resultSetMono) {
return resultSetMono.map(ReactiveWriteResult::new);
}
/**
* The mode for the {@link CassandraMessageHandler}.
*/
public enum Type {
/**
* Set a {@link CassandraMessageHandler} into an {@code insert} mode.
*/
INSERT,
/**
* Set a {@link CassandraMessageHandler} into an {@code update} mode.
*/
UPDATE,
/**
* Set a {@link CassandraMessageHandler} into a {@code delete} mode.
*/
DELETE,
/**
* Set a {@link CassandraMessageHandler} into a {@code statement} mode.
*/
STATEMENT;
}
@FunctionalInterface
private interface ReactiveSessionMessageCallback {
Mono<ReactiveResultSet> doInSession(ReactiveSession session, Message<?> requestMessage)
throws DriverException, DataAccessException;
}
private final class ReactiveWriteResult extends WriteResult {
ReactiveWriteResult(ReactiveResultSet reactiveResultSet) {
super(reactiveResultSet.getAllExecutionInfo(),
reactiveResultSet.wasApplied(),
StreamSupport.stream(reactiveResultSet.availableRows().toIterable().spliterator(), false).toList());
}
}
}

20
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/outbound/package-info.java Normal file
View File

@@ -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 supporting Cassandra outbound endpoints.
*/
package org.springframework.integration.cassandra.outbound;

1
spring-integration-cassandra/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/cassandra=org.springframework.integration.cassandra.config.xml.CassandraNamespaceHandler

3
spring-integration-cassandra/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,3 @@
http\://www.springframework.org/schema/integration/cassandra/spring-integration-cassandra.xsd=org/springframework/integration/config/xml/spring-integration-cassandra.xsd
https\://www.springframework.org/schema/integration/cassandra/spring-integration-cassandra.xsd=org/springframework/integration/config/xml/spring-integration-cassandra.xsd

4
spring-integration-cassandra/src/main/resources/META-INF/spring.tooling Normal file
View File

@@ -0,0 +1,4 @@
# Tooling related information for the integration cassandra namespace
http\://www.springframework.org/schema/integration/cassandra@name=integration cassandra Namespace
http\://www.springframework.org/schema/integration/cassandra@prefix=int-cassandra
http\://www.springframework.org/schema/integration/cassandra@icon=org/springframework/integration/config/xml/spring-integration-cassandra.gif

BIN
spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra.gif Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 578 B

228
spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra.xsd Normal file
View File

@@ -0,0 +1,228 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema
xmlns="http://www.springframework.org/schema/integration/cassandra"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/cassandra"
elementFormDefault="qualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="https://www.springframework.org/schema/integration/spring-integration.xsd"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for the Spring Integration Cassandra Adapters.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines cassandra outbound channel adapter that
writes the contents of the
Message into Cassandra cluster
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="outboundType">
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element name="parameter-expression" type="queryParameterType" minOccurs="0"
maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation>
Specify an expression for parameter variable placeholder in cql statement.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-gateway">
<xsd:annotation>
<xsd:documentation>
Defines cassandra outbound gateway that
writes the contents of the
Message into Cassandra cluster
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="outboundType">
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element name="parameter-expression" type="queryParameterType" minOccurs="0"
maxOccurs="unbounded">
<xsd:annotation>
<xsd:documentation>
Specify an expression for parameter variable placeholder in cql statement.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which replies should be sent after being received from Cassandra
cluster.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Unique ID for this gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which Messages should be sent to Cassandra.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="outboundType">
<xsd:annotation>
<xsd:documentation>
Common configuration for cassandra adapters.
</xsd:documentation>
</xsd:annotation>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0"/>
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType" minOccurs="0"/>
</xsd:sequence>
<xsd:attribute name="cassandra-template" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to an instance of
'org.springframework.data.cassandra.core.ReactiveCassandraOperations'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.data.cassandra.core.ReactiveCassandraOperations"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="write-options" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to an instance of
'org.springframework.data.cassandra.core.cql.WriteOptions'
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.data.cassandra.core.cql.WriteOptions"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mode" default="INSERT">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Indicates the `CassandraMessageHandler` behavior.
Ignored in case of explicit 'query', 'ingest-query' or `statement-expression`.]]>
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="cassandraHandlerType xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="ingest-query" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[Cql query to ingest data]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="query" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Statement to use in prepared statement
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="statement-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Statement expression that represent an executable query
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="async">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Process returned 'Mono<WriteResult>' async, reactive manner on the downstream
'FluxMessageChannel' subscription or via 'Mono.subscribe()' in the handler, if one-way.
Otherwise the 'Mono.block()' is called immediately before returning from the handler.
]]>
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="queryParameterType">
<xsd:annotation>
<xsd:documentation>
Expression to be evaluated against the Message to replace a query Parameter
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="name" use="required">
<xsd:annotation>
<xsd:documentation>
Name of the placeholder to be replaced
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="expression" use="required">
<xsd:annotation>
<xsd:documentation><![CDATA[
Expression to be evaluated to determine the replacement value.
The Message is the root object of the expression, therefore
the 'payload' and 'headers' are available directly. Any bean
may be resolved if the bean name is preceded with '@'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:simpleType name="cassandraHandlerType">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="INSERT"/>
<xsd:enumeration value="UPDATE"/>
<xsd:enumeration value="DELETE"/>
<xsd:enumeration value="STATEMENT"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:schema>

44
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/CassandraContainerTest.java Normal file
View File

@@ -0,0 +1,44 @@
/*
* 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;
import org.junit.jupiter.api.BeforeAll;
import org.testcontainers.containers.CassandraContainer;
import org.testcontainers.junit.jupiter.Testcontainers;
/**
* The base contract for JUnit tests based on the container for Apache Cassandra.
* The Testcontainers 'reuse' option must be disabled,so, Ryuk container is started
* and will clean all the containers up from this test suite after JVM exit.
* Since the MqSQL container instance is shared via static property, it is going to be
* started only once per JVM, therefore the target Docker container is reused automatically.
*
* @author Artem Bilan
*
* @since 6.0
*/
@Testcontainers(disabledWithoutDocker = true)
public interface CassandraContainerTest {
CassandraContainer<?> CASSANDRA_CONTAINER = new CassandraContainer<>("cassandra:4.1");
@BeforeAll
static void startContainer() {
CASSANDRA_CONTAINER.start();
}
}

86
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/IntegrationTestConfig.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* 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;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.cassandra.config.AbstractReactiveCassandraConfiguration;
import org.springframework.data.cassandra.config.SchemaAction;
import org.springframework.data.cassandra.core.cql.keyspace.CreateKeyspaceSpecification;
import org.springframework.integration.cassandra.test.domain.Book;
/**
* Setup any spring configuration for unit tests.
* Must be used in combination with {@link CassandraContainerTest}.
*
* @author David Webb
* @author Matthew T. Adams
* @author Artem Bilan
*
* @since 6.0
*/
@Configuration
public class IntegrationTestConfig extends AbstractReactiveCassandraConfiguration {
public String keyspaceName = randomKeyspaceName();
public static String randomKeyspaceName() {
return "ks" + UUID.randomUUID().toString().replace("-", "");
}
@Override
protected String getContactPoints() {
return CassandraContainerTest.CASSANDRA_CONTAINER.getContactPoint().getHostName();
}
@Override
protected int getPort() {
return CassandraContainerTest.CASSANDRA_CONTAINER.getContactPoint().getPort();
}
@Override
public SchemaAction getSchemaAction() {
return SchemaAction.RECREATE;
}
@Override
protected String getKeyspaceName() {
return this.keyspaceName;
}
@Override
protected List<CreateKeyspaceSpecification> getKeyspaceCreations() {
return Collections.singletonList(
CreateKeyspaceSpecification.createKeyspace(getKeyspaceName())
.withSimpleReplication());
}
@Override
protected String getLocalDataCenter() {
return CassandraContainerTest.CASSANDRA_CONTAINER.getLocalDatacenter();
}
@Override
public String[] getEntityBasePackages() {
return new String[]{ Book.class.getPackage().getName() };
}
}

42
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests-context.xml Normal file
View File

@@ -0,0 +1,42 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-cassandra="http://www.springframework.org/schema/integration/cassandra"
xsi:schemaLocation="http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/cassandra https://www.springframework.org/schema/integration/cassandra/spring-integration-cassandra.xsd">
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler1"
cassandra-template="reactiveCassandraTemplate"
async="false"/>
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler2"
cassandra-template="reactiveCassandraTemplate"
async="false"/>
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler3"
cassandra-template="reactiveCassandraTemplate"
ingest-query="insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)"
async="false"/>
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler4"
cassandra-template="reactiveCassandraTemplate"
statement-expression="T(QueryBuilder).truncate('book').build()"
async="false"/>
<int:channel id="inputChannel"/>
<bean id="resultChannel" class="org.springframework.integration.channel.FluxMessageChannel"/>
<int-cassandra:outbound-gateway id="cassandraMessageHandler5"
request-channel="inputChannel"
cassandra-template="reactiveCassandraTemplate"
mode="STATEMENT"
query="SELECT * FROM book limit :size"
reply-channel="resultChannel">
<int-cassandra:parameter-expression name="author" expression="payload"/>
<int-cassandra:parameter-expression name="size" expression="headers.limit"/>
</int-cassandra:outbound-gateway>
</beans>

156
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests.java Normal file
View File

@@ -0,0 +1,156 @@
/*
* 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;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.ArrayList;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
import org.springframework.data.cassandra.core.WriteResult;
import org.springframework.integration.cassandra.CassandraContainerTest;
import org.springframework.integration.cassandra.IntegrationTestConfig;
import org.springframework.integration.cassandra.test.domain.Book;
import org.springframework.integration.cassandra.test.domain.BookSampler;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
import com.datastax.oss.driver.api.querybuilder.select.Select;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
/**
* @author Filippo Balicchia
* @author Artem Bilan
*
* @since 6.0
*/
@SpringJUnitConfig(CassandraOutboundAdapterIntegrationTests.Config.class)
@DirtiesContext
class CassandraOutboundAdapterIntegrationTests implements CassandraContainerTest {
@Autowired
private ReactiveCassandraTemplate cassandraTemplate;
@Autowired
private DirectChannel cassandraMessageHandler1;
@Autowired
private DirectChannel cassandraMessageHandler2;
@Autowired
private DirectChannel cassandraMessageHandler3;
@Autowired
private DirectChannel cassandraMessageHandler4;
@Autowired
private DirectChannel inputChannel;
@Autowired
private FluxMessageChannel resultChannel;
@Test
void testBasicCassandraInsert() {
Book b1 = BookSampler.getBook();
Message<Book> message = MessageBuilder.withPayload(b1).build();
this.cassandraMessageHandler1.send(message);
Select select = QueryBuilder.selectFrom("book").all();
List<Book> books = this.cassandraTemplate.select(select.build(), Book.class).collectList().block();
assertThat(books).hasSize(1);
this.cassandraTemplate.delete(b1);
}
@Test
void testCassandraBatchInsertAndSelectStatement() {
List<Book> books = BookSampler.getBookList(5);
this.cassandraMessageHandler2.send(new GenericMessage<>(books));
Message<?> message = MessageBuilder.withPayload("Cassandra Puppy Guru").setHeader("limit", 2).build();
this.inputChannel.send(message);
Mono<Integer> testMono =
Mono.from(this.resultChannel)
.map(Message::getPayload)
.cast(WriteResult.class)
.map(r -> r.getRows().size());
StepVerifier.create(testMono)
.expectNext(2)
.expectComplete()
.verify();
this.cassandraMessageHandler1.send(new GenericMessage<>(QueryBuilder.truncate("book").build()));
}
@Test
void testCassandraBatchIngest() {
List<Book> books = BookSampler.getBookList(5);
List<List<?>> ingestBooks = new ArrayList<>();
for (Book b : books) {
List<Object> l = new ArrayList<>();
l.add(b.isbn());
l.add(b.title());
l.add(b.author());
l.add(b.pages());
l.add(b.saleDate());
l.add(b.isInStock());
ingestBooks.add(l);
}
Message<List<List<?>>> message = MessageBuilder.withPayload(ingestBooks).build();
this.cassandraMessageHandler3.send(message);
Select select = QueryBuilder.selectFrom("book").all();
books = this.cassandraTemplate.select(select.build(), Book.class).collectList().block();
assertThat(books).hasSize(5);
this.cassandraTemplate.batchOps().delete(books);
}
@Test
void testExpressionTruncate() {
Message<Book> message = MessageBuilder.withPayload(BookSampler.getBook()).build();
this.cassandraMessageHandler1.send(message);
Select select = QueryBuilder.selectFrom("book").all();
List<Book> books = this.cassandraTemplate.select(select.build(), Book.class).collectList().block();
assertThat(books).hasSize(1);
this.cassandraMessageHandler4.send(MessageBuilder.withPayload("Empty").build());
books = this.cassandraTemplate.select(select.build(), Book.class).collectList().block();
assertThat(books).hasSize(0);
}
@Configuration
@EnableIntegration
@ImportResource("org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests-context.xml")
public static class Config extends IntegrationTestConfig {
}
}

59
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests-context.xml Normal file
View File

@@ -0,0 +1,59 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-cassandra="http://www.springframework.org/schema/integration/cassandra"
xsi:schemaLocation="http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/cassandra https://www.springframework.org/schema/integration/cassandra/spring-integration-cassandra.xsd">
<int:poller default="true" fixed-delay="50"/>
<int:channel id="input">
<int:queue/>
</int:channel>
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
<bean id="cassandraTemplate" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.data.cassandra.core.ReactiveCassandraOperations"/>
</bean>
<bean id="writeOptions" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.data.cassandra.core.InsertOptions"/>
</bean>
<int-cassandra:outbound-channel-adapter id="outbound1"
cassandra-template="cassandraTemplate"
write-options="writeOptions"
auto-startup="false"
async="false"/>
<int-cassandra:outbound-channel-adapter id="outbound2"
channel="input"
cassandra-template="cassandraTemplate"
ingest-query="insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)"/>
<int-cassandra:outbound-gateway id="outgateway"
request-channel="input"
cassandra-template="cassandraTemplate"
mode="STATEMENT"
write-options="writeOptions"
query="SELECT * FROM book limit :size"
reply-channel="resultChannel"
auto-startup="true">
<int-cassandra:parameter-expression name="author" expression="payload"/>
<int-cassandra:parameter-expression name="size" expression="headers.limit"/>
</int-cassandra:outbound-gateway>
<int-cassandra:outbound-channel-adapter id="outbound4"
cassandra-template="cassandraTemplate"
write-options="writeOptions"
statement-expression="T(QueryBuilder).truncate('book')"
auto-startup="false"/>
</beans>

91
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests.java Normal file
View File

@@ -0,0 +1,91 @@
/*
* 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;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.cassandra.outbound.CassandraMessageHandler;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Filippo Balicchia
* @author Artem Bilan
*
* @since 6.0
*/
@SpringJUnitConfig
class CassandraOutboundAdapterParserTests {
@Autowired
private ApplicationContext context;
@Test
void minimalConfig() {
CassandraMessageHandler handler =
TestUtils.getPropertyValue(this.context.getBean("outbound1.adapter"), "handler",
CassandraMessageHandler.class);
assertThat(TestUtils.getPropertyValue(handler, "componentName")).isEqualTo("outbound1.adapter");
assertThat(TestUtils.getPropertyValue(handler, "mode")).isEqualTo(CassandraMessageHandler.Type.INSERT);
assertThat(TestUtils.getPropertyValue(handler, "cassandraOperations"))
.isSameAs(this.context.getBean("cassandraTemplate"));
assertThat(TestUtils.getPropertyValue(handler, "writeOptions")).isSameAs(this.context.getBean("writeOptions"));
assertThat(TestUtils.getPropertyValue(handler, "async", Boolean.class)).isFalse();
}
@Test
void ingestConfig() {
CassandraMessageHandler handler =
TestUtils.getPropertyValue(this.context.getBean("outbound2"), "handler",
CassandraMessageHandler.class);
assertThat(TestUtils.getPropertyValue(handler, "ingestQuery"))
.isEqualTo("insert into book (isbn, title, author, pages, saleDate, isInStock) " +
"values (?, ?, ?, ?, ?, ?)");
assertThat(TestUtils.getPropertyValue(handler, "producesReply", Boolean.class)).isFalse();
}
@Test
void fullConfig() {
CassandraMessageHandler handler =
TestUtils.getPropertyValue(this.context.getBean("outgateway"), "handler",
CassandraMessageHandler.class);
assertThat(TestUtils.getPropertyValue(handler, "producesReply", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(handler, "mode")).isEqualTo(CassandraMessageHandler.Type.STATEMENT);
assertThat(TestUtils.getPropertyValue(handler, "writeOptions")).isSameAs(this.context.getBean("writeOptions"));
}
@Test
void statementConfig() {
CassandraMessageHandler handler =
TestUtils.getPropertyValue(this.context.getBean("outbound4.adapter"), "handler",
CassandraMessageHandler.class);
assertThat(TestUtils.getPropertyValue(handler, "componentName")).isEqualTo("outbound4.adapter");
assertThat(TestUtils.getPropertyValue(handler, "mode")).isEqualTo(CassandraMessageHandler.Type.STATEMENT);
assertThat(TestUtils.getPropertyValue(handler, "cassandraOperations"))
.isSameAs(this.context.getBean("cassandraTemplate"));
assertThat(TestUtils.getPropertyValue(handler, "writeOptions")).isSameAs(this.context.getBean("writeOptions"));
}
}

103
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraParserUtilsTests.java Normal file
View File

@@ -0,0 +1,103 @@
/*
* 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;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.cassandra.config.xml.CassandraParserUtils;
/**
* @author Filippo Balicchia
* @author Artem Bilan
*
* @since 6.0
*/
class CassandraParserUtilsTests {
@Test
void mutuallyExclusiveCase1() {
String query = "";
BeanDefinition statementExpressionDef = null;
String ingestQuery = "";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isTrue();
}
@Test
void mutuallyExclusiveCase2() {
String query = "";
BeanDefinition statementExpressionDef = null;
String ingestQuery =
"insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isTrue();
}
@Test
void mutuallyExclusiveCase3() {
String query = "";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery = "";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isTrue();
}
@Test
void mutuallyExclusiveCase4() {
String query = "";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery =
"insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isFalse();
}
@Test
void mutuallyExclusiveCase5() {
String query = "SELECT * FROM book limit :size";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery = "";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isFalse();
}
@Test
void mutuallyExclusiveCase6() {
String query = "SELECT * FROM book limit :size";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery =
"insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isFalse();
}
@Test
void mutuallyExclusiveCase7() {
String query = "SELECT * FROM book limit :size";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery = "";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isFalse();
}
@Test
void mutuallyExclusiveCase8() {
String query = "SELECT * FROM book limit :size";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery =
"insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
assertThat(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery)).isFalse();
}
}

128
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/dsl/CassandraDslTests.java Normal file
View File

@@ -0,0 +1,128 @@
/*
* 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 java.time.Duration;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.cassandra.core.InsertOptions;
import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
import org.springframework.data.cassandra.core.WriteResult;
import org.springframework.integration.cassandra.CassandraContainerTest;
import org.springframework.integration.cassandra.IntegrationTestConfig;
import org.springframework.integration.cassandra.test.domain.BookSampler;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.datastax.oss.driver.api.core.ConsistencyLevel;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
/**
* @author Artem Bilan
*
* @since 6.0
*/
@SpringJUnitConfig
@DirtiesContext
public class CassandraDslTests implements CassandraContainerTest {
@Autowired
@Qualifier("cassandraTruncateFlow.input")
MessageChannel cassandraTruncateFlowInput;
@Autowired
@Qualifier("cassandraInsertFlow.input")
MessageChannel cassandraInsertFlowInput;
@Autowired
@Qualifier("cassandraSelectFlow.input")
MessageChannel cassandraSelectFlowInput;
@Autowired
FluxMessageChannel resultChannel;
@Test
void testCassandraDslConfiguration() {
this.cassandraInsertFlowInput.send(new GenericMessage<>(BookSampler.getBookList(5)));
Mono<Integer> testMono =
Mono.from(this.resultChannel)
.map(Message::getPayload)
.cast(WriteResult.class)
.map(r -> r.getRows().size());
StepVerifier stepVerifier = StepVerifier.create(testMono)
.expectNext(1)
.expectComplete()
.verifyLater();
this.cassandraSelectFlowInput.send(MessageBuilder.withPayload("Cassandra Guru").setHeader("limit", 2).build());
stepVerifier.verify(Duration.ofSeconds(10));
this.cassandraTruncateFlowInput.send(new GenericMessage<>(""));
}
@Configuration
@EnableIntegration
public static class Config extends IntegrationTestConfig {
@Bean
IntegrationFlow cassandraTruncateFlow(ReactiveCassandraOperations cassandraOperations) {
return flow -> flow
.handle(Cassandra.outboundChannelAdapter(cassandraOperations)
.statementExpression("T(QueryBuilder).truncate('book').build()"),
e -> e.async(false));
}
@Bean
IntegrationFlow cassandraInsertFlow(ReactiveCassandraOperations cassandraOperations) {
return flow -> flow
.handle(Cassandra.outboundChannelAdapter(cassandraOperations)
.writeOptions(InsertOptions.builder()
.ttl(60)
.consistencyLevel(ConsistencyLevel.ONE)
.build()),
e -> e.async(false));
}
@Bean
IntegrationFlow cassandraSelectFlow(ReactiveCassandraOperations cassandraOperations) {
return flow -> flow
.handle(Cassandra.outboundGateway(cassandraOperations)
.query("SELECT * FROM book WHERE author = :author limit :size")
.parameter("author", "payload")
.parameter("size", m -> m.getHeaders().get("limit")))
.channel(c -> c.flux("resultChannel"));
}
}
}

217
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandlerTests.java Normal file
View File

@@ -0,0 +1,217 @@
/*
* 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.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.cassandra.core.CassandraOperations;
import org.springframework.data.cassandra.core.InsertOptions;
import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
import org.springframework.data.cassandra.core.WriteResult;
import org.springframework.data.cassandra.core.cql.WriteOptions;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.cassandra.CassandraContainerTest;
import org.springframework.integration.cassandra.IntegrationTestConfig;
import org.springframework.integration.cassandra.test.domain.Book;
import org.springframework.integration.cassandra.test.domain.BookSampler;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.datastax.oss.driver.api.core.ConsistencyLevel;
import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
import com.datastax.oss.driver.api.querybuilder.select.Select;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
/**
* @author Soby Chacko
* @author Artem Bilan
*
* @since 6.0
*/
@SpringJUnitConfig
@DirtiesContext
public class CassandraMessageHandlerTests implements CassandraContainerTest {
private static final SpelExpressionParser PARSER = new SpelExpressionParser();
@Autowired
public MessageHandler cassandraMessageHandler1;
@Autowired
public MessageHandler cassandraMessageHandler2;
@Autowired
public MessageHandler cassandraMessageHandler3;
@Autowired
public MessageHandler cassandraMessageHandler4;
@Autowired
public CassandraOperations template;
@Autowired
public FluxMessageChannel resultChannel;
@Test
void testBasicCassandraInsert() {
Book b1 = BookSampler.getBook();
Message<Book> message = MessageBuilder.withPayload(b1).build();
this.cassandraMessageHandler1.handleMessage(message);
Select select = QueryBuilder.selectFrom("book").all();
List<Book> books = this.template.select(select.build(), Book.class);
assertThat(books).hasSize(1);
this.template.delete(b1);
}
@Test
void testCassandraBatchInsertAndSelectStatement() {
List<Book> books = BookSampler.getBookList(5);
this.cassandraMessageHandler2.handleMessage(new GenericMessage<>(books));
Message<?> message = MessageBuilder.withPayload("Cassandra Guru").setHeader("limit", 2).build();
this.cassandraMessageHandler4.handleMessage(message);
Mono<Integer> testMono =
Mono.from(this.resultChannel)
.map(Message::getPayload)
.cast(WriteResult.class)
.map(r -> r.getRows().size());
StepVerifier.create(testMono)
.expectNext(1)
.expectComplete()
.verify();
this.cassandraMessageHandler1.handleMessage(new GenericMessage<>(QueryBuilder.truncate("book").build()));
}
@Test
void testCassandraBatchIngest() {
List<Book> books = BookSampler.getBookList(5);
List<List<Object>> ingestBooks =
books.stream()
.map(book ->
List.<Object>of(
book.isbn(),
book.title(),
book.author(),
book.pages(),
book.saleDate(),
book.isInStock()))
.toList();
this.cassandraMessageHandler3.handleMessage(MessageBuilder.withPayload(ingestBooks).build());
Select select = QueryBuilder.selectFrom("book").all();
books = this.template.select(select.build(), Book.class);
assertThat(books).hasSize(5);
this.template.batchOps().delete(books);
}
@Configuration
@EnableIntegration
public static class Config extends IntegrationTestConfig {
@Autowired
public ReactiveCassandraOperations template;
@Bean
public MessageHandler cassandraMessageHandler1() {
CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
cassandraMessageHandler.setAsync(false);
return cassandraMessageHandler;
}
@Bean
public PollableChannel messageChannel() {
return new NullChannel();
}
@Bean
public MessageHandler cassandraMessageHandler2() {
CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
WriteOptions options =
InsertOptions.builder()
.ttl(60)
.consistencyLevel(ConsistencyLevel.ONE)
.build();
cassandraMessageHandler.setWriteOptions(options);
cassandraMessageHandler.setOutputChannel(messageChannel());
cassandraMessageHandler.setAsync(false);
return cassandraMessageHandler;
}
@Bean
public MessageHandler cassandraMessageHandler3() {
CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
String cqlIngest =
"insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
cassandraMessageHandler.setIngestQuery(cqlIngest);
cassandraMessageHandler.setAsync(false);
return cassandraMessageHandler;
}
@Bean
public FluxMessageChannel resultChannel() {
return new FluxMessageChannel();
}
@Bean
public MessageHandler cassandraMessageHandler4() {
CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
cassandraMessageHandler.setQuery("SELECT * FROM book WHERE author = :author limit :size");
Map<String, Expression> params = new HashMap<>();
params.put("author", PARSER.parseExpression("payload"));
params.put("size", PARSER.parseExpression("headers.limit"));
cassandraMessageHandler.setParameterExpressions(params);
cassandraMessageHandler.setOutputChannel(resultChannel());
cassandraMessageHandler.setProducesReply(true);
return cassandraMessageHandler;
}
}
}

43
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/test/domain/Book.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* 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.test.domain;
import java.time.LocalDate;
import org.springframework.data.cassandra.core.mapping.Indexed;
import org.springframework.data.cassandra.core.mapping.PrimaryKey;
import org.springframework.data.cassandra.core.mapping.Table;
/**
* Test POJO
*
* @author David Webb
* @author Artem Bilan
*
* @since 6.0
*/
@Table("book")
public record Book(
@PrimaryKey String isbn,
String title,
@Indexed String author,
Integer pages,
LocalDate saleDate,
Boolean isInStock) {
}

49
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/test/domain/BookSampler.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* 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.test.domain;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
/**
* @author Filippo Balicchia
* @author Artem Bilan
*
* @since 6.0
*/
public final class BookSampler {
public static List<Book> getBookList(int numBooks) {
List<Book> books = new ArrayList<>();
for (int i = 0; i < numBooks - 1; i++) {
books.add(new Book(UUID.randomUUID().toString(), "Spring Data Cassandra Guide", "Cassandra Guru puppy",
i * 10 + 5, LocalDate.now(), true));
}
books.add(getBook());
return books;
}
public static Book getBook() {
return new Book("123456-1", "Spring Integration Cassandra", "Cassandra Guru", 521, LocalDate.now(), true);
}
private BookSampler() {
}
}

15
spring-integration-cassandra/src/test/resources/log4j2-test.xml Normal file
View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
<Appenders>
<Console name="STDOUT" target="SYSTEM_OUT">
<PatternLayout pattern="%d %p [%t] [%c] - %m%n" />
</Console>
</Appenders>
<Loggers>
<Logger name="org.springframework.integration" level="warn"/>
<Logger name="org.springframework.integration.cassandra" level="info"/>
<Root level="warn">
<AppenderRef ref="STDOUT" />
</Root>
</Loggers>
</Configuration>

179
src/reference/asciidoc/cassandra.adoc Normal file
View File

@@ -0,0 +1,179 @@
[[cassandra]]
== Apache Cassandra Support
Spring Integration provides channel adapters (starting with version 6.0) for performing database operations against an Apache Cassandra cluster.
It is fully based on the https://spring.io/projects/spring-data-cassandra[Spring Data for Apache Cassandra] project.
You need to include this dependency into your project:
====
[source, xml, subs="normal", role="primary"]
.Maven
----
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-cassandra</artifactId>
<version>{project-version}</version>
</dependency>
----
[source, groovy, subs="normal", role="secondary"]
.Gradle
----
compile "org.springframework.integration:spring-integration-cassandra:{project-version}"
----
====
[[cassandra-outbound]]
=== Cassandra Outbound Components
The `CassandraMessageHandler` is an `AbstractReplyProducingMessageHandler` implementation and can work in both one-way (default) and request-reply modes (a `producesReply` option).
It is asynchronous by default (`setAsync(false)` to reset) and performs reactive `INSERT`, `UPDATE`, `DELETE` or `STATEMENT` operations against the provided `ReactiveCassandraOperations`.
The type of operation can be configured via the `CassandraMessageHandler.Type` option.
The `ingestQuery` sets the mode into an `INSERT`; the `query` or `statementExpression`, or `statementProcessor` sets the mode into a `STATEMENT`.
The following code snippets demonstrate various configuration for this channel adapter or gateway:
====
[source, java, role="primary"]
.Java DSL
----
@Bean
IntegrationFlow cassandraSelectFlow(ReactiveCassandraOperations cassandraOperations) {
return flow -> flow
.handle(Cassandra.outboundGateway(cassandraOperations)
.query("SELECT * FROM book WHERE author = :author limit :size")
.parameter("author", "payload")
.parameter("size", m -> m.getHeaders().get("limit")))
.channel(c -> c.flux("resultChannel"));
}
----
[source, kotlin, role="secondary"]
.Kotlin DSL
----
@Bean
fun outboundReactive(cassandraOperations: ReactiveCassandraOperations) =
integrationFlow {
handle(
Cassandra.outboundChannelAdapter(cassandraOperations)
.statementExpression("T(QueryBuilder).truncate('book').build()")
) { async(false) }
}
----
[source, java, role="secondary"]
.Java
----
@ServiceActivator(inputChannel = "cassandraSelectChannel")
@Bean
public MessageHandler cassandraMessageHandler() {
CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
cassandraMessageHandler.setQuery("SELECT * FROM book WHERE author = :author limit :size");
Map<String, Expression> params = new HashMap<>();
params.put("author", PARSER.parseExpression("payload"));
params.put("size", PARSER.parseExpression("headers.limit"));
cassandraMessageHandler.setParameterExpressions(params);
cassandraMessageHandler.setOutputChannel(resultChannel());
cassandraMessageHandler.setProducesReply(true);
return cassandraMessageHandler;
}
----
[source, xml, role="secondary"]
.XML
----
<int-cassandra:outbound-channel-adapter id="outboundAdapter"
cassandra-template="cassandraTemplate"
write-options="writeOptions"
auto-startup="false"
async="false"/>
<int-cassandra:outbound-gateway id="outgateway"
request-channel="input"
cassandra-template="cassandraTemplate"
mode="STATEMENT"
write-options="writeOptions"
query="SELECT * FROM book limit :size"
reply-channel="resultChannel"
auto-startup="true">
<int-cassandra:parameter-expression name="author" expression="payload"/>
<int-cassandra:parameter-expression name="size" expression="headers.limit"/>
</int-cassandra:outbound-gateway>
----
====
If a `CassandraMessageHandler` is used as a gateway in the default async mode, a `Mono<WriteResult>` is produced, which is handled according to the provided `MessageChannel` implementation.
For true reactive processing a `FluxMessageChannel` is recommended for the output channel configuration.
In sync mode `Mono.block()` is called to obtain the reply value.
If `INSERT`, `UPDATE` or `DELETE` operations are performed, an entity (marked `org.springframework.data.cassandra.core.mapping.Table`) is expected in the request message payload.
If the payload is a list of entities, then the respective batch operation is performed.
The `ingestQuery` mode expects the payload to be present as a matrix of values to insert - `List<List<?>>`.
For example, if the entity is like this:
====
[source,java]
----
@Table("book")
public record Book(@PrimaryKey String isbn,
String title,
@Indexed String author,
int pages,
LocalDate saleDate,
boolean isInStock) {
}
----
====
And channel adapter has this configuration:
====
[source,java]
----
@Bean
public MessageHandler cassandraMessageHandler3() {
CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
String cqlIngest = "insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
cassandraMessageHandler.setIngestQuery(cqlIngest);
cassandraMessageHandler.setAsync(false);
return cassandraMessageHandler;
}
----
====
The request message payload must be converted like this:
====
[source,java]
----
List<List<Object>> ingestBooks =
payload.stream()
.map(book ->
List.<Object>of(
book.isbn(),
book.title(),
book.author(),
book.pages(),
book.saleDate(),
book.isInStock()))
.toList();
----
====
For more sophisticated use-cases, the payload can be as an instance of `com.datastax.oss.driver.api.core.cql.Statement`.
The `com.datastax.oss.driver.api.querybuilder.QueryBuilder` API is recommended to build various statements to execute against Apache Cassandra.
For example, to remove all the data from the `Book` table, a message with a payload like this can be sent to the `CassandraMessageHandler`: `QueryBuilder.truncate("book").build()`.
Alternatively, for logic based on a request message, a `statementExpression` or `statementProcessor` can be provided for the `CassandraMessageHandler` to build a `Statement` based on that message.
For convenience, a `com.datastax.oss.driver.api.querybuilder` is registered as an `import` into a SpEL evaluation context, so a target expression can be as simple as this:
====
[source,xml]
----
statement-expression="T(QueryBuilder).selectFrom("book").all()"
----
====
The `setParameterExpressions(Map<String, Expression> parameterExpressions)` represents bindable named query parameters and is used only with a `setQuery(String query)` option.
See Java and XML samples mentioned above.

6
src/reference/asciidoc/endpoint-summary.adoc
View File

@@ -54,6 +54,12 @@ The following table summarizes the various endpoints with quick links to the app
| <<./amqp.adoc#amqp-inbound-gateway,Inbound Gateway>>
| <<./amqp.adoc#amqp-outbound-gateway,Outbound Gateway>>
| *Apache Cassandra*
| N
| <<./cassandra.adoc#cassandra-outbound,Outbound Channel Adapter>>
| N
| <<./cassandra.adoc#cassandra-outbound,Outbound Gateway>>
| *Events*
| <<./event.adoc#appevent-inbound,Receiving Spring Application Events>>
| <<./event.adoc#appevent-outbound,Sending Spring Application Events>>

2
src/reference/asciidoc/index-single.adoc
View File

@@ -41,6 +41,8 @@ include::./endpoint-summary.adoc[]
include::./amqp.adoc[]
include::./cassandra.adoc[]
include::./event.adoc[]
include::./feed.adoc[]

1
src/reference/asciidoc/index.adoc
View File

@@ -34,6 +34,7 @@ Welcome to the Spring Integration reference documentation!
[horizontal]
<<./endpoint-summary.adoc#spring-integration-endpoints,Integration Endpoint Summary>> :: Protocol-specific channel adapters and gateways summary
<<./amqp.adoc#amqp,AMQP Support>> :: AMQP channels, adapters and gateways
<<./cassandra.adoc#cassandra,Apache Cassandra Support>> :: Apache Cassandra channel adapters
<<./event.adoc#applicationevent,Spring `ApplicationEvent` Support>> :: Handling and consuming Spring application events with channel adapters
<<./feed.adoc#feed,Feed Adapter>> :: RSS and Atom channel adapters
<<./file.adoc#files,File Support>> :: Channel adapters and gateways for file system support

3
src/reference/asciidoc/reactive-streams.adoc
View File

@@ -326,8 +326,7 @@ public class MainFlow {
----
====
Currently, Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>>, <<./rsocket.adoc#rsocket,RSocket>>, <<./mongodb.adoc#mongodb,MongoDb>>, <<./r2dbc.adoc#r2dbc,R2DBC>>, <<./zeromq.adoc#zeromq,ZeroMQ>>, <<./graphql.adoc#graphql,GraphQL>>.
Currently, Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>>, <<./rsocket.adoc#rsocket,RSocket>>, <<./mongodb.adoc#mongodb,MongoDb>>, <<./r2dbc.adoc#r2dbc,R2DBC>>, <<./zeromq.adoc#zeromq,ZeroMQ>>, <<./graphql.adoc#graphql,GraphQL>>, <<./cassandra.adoc#cassandra,Apache Cassandra>>.
The <<./redis.adoc#redis-stream-outbound,Redis Stream Channel Adapters>> are also reactive and uses `ReactiveStreamOperations` from Spring Data.
Also, an https://github.com/spring-projects/spring-integration-extensions/tree/main/spring-integration-cassandra[Apache Cassandra Extension] provides a `MessageHandler` implementation for the Cassandra reactive driver.
More reactive channel adapters are coming, for example for Apache Kafka in <<./kafka.adoc#kafka,Kafka>> based on the `ReactiveKafkaProducerTemplate` and `ReactiveKafkaConsumerTemplate` from https://spring.io/projects/spring-kafka[Spring for Apache Kafka] etc.
For many other non-reactive channel adapters thread pools are recommended to avoid blocking during reactive stream processing.

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -73,6 +73,12 @@ The Scripting module now provides a `PolyglotScriptExecutor` implementation base
JavaScript support is now based on this executor since its JSR223 implementation has been removed from Java by itself.
See <<./scripting.adoc#scripting,Scripting Support>> for more information.
[[x6.0-cassandra]]
==== Apache Cassandra Support
The Apache Cassandra Spring Integration Extensions project has been migrated as the `spring-integration-cassandra` module.
See <<./cassandra.adoc#cassandra,Apache Cassandra Support>> for more information.
[[x6.0-general]]
=== General Changes