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

INTEXT-215: Cassandra Namespace Support

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-215
This commit is contained in:
Filippo Balicchia
2015-12-16 07:43:27 +01:00
committed by Artem Bilan
parent 2e9247711d
commit f7b31228c2
16 changed files with 976 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-cassandra/build.gradle
View File

@@ -40,7 +40,7 @@ ext {
jacocoVersion = '0.7.2.201409121644'
slf4jVersion = '1.7.12'
springDataCassandraVersion = '1.3.0.RELEASE'
springIntegrationVersion = '4.2.0.RELEASE'
springIntegrationVersion = '4.2.4.RELEASE'
idPrefix = 'cassandra'

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors
* Copyright 2015-2016 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.
@@ -20,12 +20,14 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
/**
* @author Artem Bilan
* @author Filippo Balicchia
*/
public class CassandraNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@Override
public void init() {
registerBeanDefinitionParser("outbound-channel-adapter", new CassandraOutboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-gateway", new CassandraOutboundGatewayParser());
}
}

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

@@ -0,0 +1,40 @@
/*
* Copyright 2016 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
*
* http://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;
/**
* @author Filippo Balicchia
*/
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();
}
}

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

@@ -0,0 +1,47 @@
/*
* Copyright 2016 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
*
* http://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;
/**
* @author Filippo Balicchia
*/
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;
}
}

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

@@ -0,0 +1,95 @@
/*
* Copyright 2016 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
*
* http://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;
/**
* @author Filippo Balicchia
*/
public 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.isEmpty(cassandraTemplate)) {
parserContext.getReaderContext().error("cassandra-template is required", element);
}
builder.addConstructorArgReference(cassandraTemplate);
if (!StringUtils.isEmpty(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");
List<Element> parameterExpressions = DomUtils.getChildElementsByTagName(element, "parameter-expression");
if (!CollectionUtils.isEmpty(parameterExpressions)) {
ManagedMap<String, Object> parameterExpressionsMap = new ManagedMap<String, Object>();
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.isEmpty(query) && statementExpressionDef == null && StringUtils.isEmpty(ingestQuery)
|| !(StringUtils.hasText(query) && statementExpressionDef != null && StringUtils.hasText(ingestQuery))
&& (StringUtils.hasText(query) ^ statementExpressionDef != null) ^ StringUtils.hasText(ingestQuery);
}
}

17
spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-2016 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.
@@ -44,6 +44,7 @@ import com.datastax.driver.core.Statement;
/**
* @author Soby Chacko
* @author Artem Bilan
* @author Filippo Balicchia
*/
@SuppressWarnings("unchecked")
public class CassandraMessageHandler<T> extends AbstractReplyProducingMessageHandler {
@@ -52,7 +53,7 @@ public class CassandraMessageHandler<T> extends AbstractReplyProducingMessageHan
private final CassandraOperations cassandraTemplate;
private Type queryType;
private Type mode;
private boolean producesReply;
@@ -78,13 +79,13 @@ public class CassandraMessageHandler<T> extends AbstractReplyProducingMessageHan
Assert.notNull(cassandraTemplate, "'cassandraTemplate' must not be null.");
Assert.notNull(queryType, "'queryType' must not be null.");
this.cassandraTemplate = cassandraTemplate;
this.queryType = queryType;
this.mode = queryType;
}
public void setIngestQuery(String ingestQuery) {
Assert.hasText(ingestQuery, "'ingestQuery' must not be empty");
this.ingestQuery = ingestQuery;
this.queryType = Type.INSERT;
this.mode = Type.INSERT;
}
public void setWriteOptions(WriteOptions writeOptions) {
@@ -147,7 +148,7 @@ public class CassandraMessageHandler<T> extends AbstractReplyProducingMessageHan
public void setStatementProcessor(MessageProcessor<Statement> statementProcessor) {
Assert.notNull(statementProcessor, "'statementProcessor' must not be null.");
this.statementProcessor = statementProcessor;
this.queryType = Type.STATEMENT;
this.mode = Type.STATEMENT;
}
@Override
@@ -175,16 +176,16 @@ public class CassandraMessageHandler<T> extends AbstractReplyProducingMessageHan
Object result = payload;
Type queryType = this.queryType;
Type mode = this.mode;
Statement statement = null;
if (payload instanceof Statement) {
statement = (Statement) payload;
queryType = Type.STATEMENT;
mode = Type.STATEMENT;
}
switch (queryType) {
switch (mode) {
case INSERT:
if (this.ingestQuery != null) {
Assert.isInstanceOf(List.class, payload,

213
spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra-1.0.xsd
View File

@@ -1,15 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/kafka"
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/kafka"
elementFormDefault="qualified" attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:schema
xmlns="http://www.springframework.org/schema/integration/cassandra"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
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" attributeFormDefault="unqualified">
<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="http://www.springframework.org/schema/integration/spring-integration.xsd"/>
schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd" />
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -17,5 +17,196 @@
]]></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" maxOccurs="1"/>
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
</xsd:sequence>
<xsd:attribute name="cassandra-template" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Reference to an instance of
org.springframework.data.cassandra.core.CassandraOperations]]>
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.data.cassandra.core.CassandraOperations" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="write-options" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Reference to an instance of
org.springframework.cassandra.core.WriteOptions]]>
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.cassandra.core.WriteOptions" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mode" default="INSERT" use="optional">
<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: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>

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

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

@@ -0,0 +1,87 @@
<?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:util="http://www.springframework.org/schema/util"
xmlns:cassandra="http://www.springframework.org/schema/data/cassandra"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:int-cassandra="http://www.springframework.org/schema/integration/cassandra"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/data/cassandra http://www.springframework.org/schema/data/cassandra/spring-cassandra.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration/cassandra http://www.springframework.org/schema/integration/cassandra/spring-integration-cassandra.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
<context:property-placeholder location="classpath:cassandra.properties" />
<int:poller default="true" fixed-delay="50" />
<cassandra:mapping entity-base-packages="org.springframework.integration.cassandra.test.domain">
<cassandra:entity
class="org.springframework.integration.cassandra.test.domain.Book">
<cassandra:table name="book" />
</cassandra:entity>
</cassandra:mapping>
<cassandra:cluster contact-points="${cassandra.contactpoints}" port="${cassandra.port}">
<cassandra:keyspace action="CREATE_DROP" name="${cassandra.keyspace}" />
</cassandra:cluster>
<cassandra:session keyspace-name="${cassandra.keyspace}" schema-action="RECREATE"/>
<cassandra:converter />
<cassandra:template id="cassandraTemplate" />
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler1"
cassandra-template="cassandraTemplate"
mode="INSERT"
auto-startup="true" />
<bean id="writeOptions" class="org.springframework.cassandra.core.WriteOptions">
<property name="ttl" value="60" />
<property name="consistencyLevel">
<util:constant
static-field="org.springframework.cassandra.core.ConsistencyLevel.ONE" />
</property>
<property name="retryPolicy">
<util:constant
static-field="org.springframework.cassandra.core.RetryPolicy.DOWNGRADING_CONSISTENCY" />
</property>
</bean>
<bean id="resultChannel" class="org.springframework.integration.channel.QueueChannel"/>
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler2"
cassandra-template="cassandraTemplate"
write-options="writeOptions"
auto-startup="true" />
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler3"
cassandra-template="cassandraTemplate"
ingest-query="insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)"
auto-startup="true" />
<int-cassandra:outbound-channel-adapter id="cassandraMessageHandler4"
cassandra-template="cassandraTemplate"
statement-expression="T(QueryBuilder).truncate('book')"
auto-startup="true" />
<int:channel id="inputChannel" />
<int-cassandra:outbound-gateway id="cassandraMessageHandler5"
request-channel="inputChannel"
cassandra-template="cassandraTemplate"
mode="STATEMENT"
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>
<cassandra:repositories base-package="org.springframework.integration.cassandra.test.domain" />
</beans>

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

@@ -0,0 +1,166 @@
/*
* Copyright 2016 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
*
* http://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.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.thrift.transport.TTransportException;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.cassandra.core.CassandraTemplate;
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.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
/**
* @author Filippo Balicchia
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@DirtiesContext
public class CassandraOutboundAdapterIntegrationTests {
protected static final String CASSANDRA_CONFIG = "spring-cassandra.yaml";
@Autowired
private CassandraTemplate cassandraTemplate;
@Autowired
private DirectChannel cassandraMessageHandler1;
@Autowired
private DirectChannel cassandraMessageHandler2;
@Autowired
private DirectChannel cassandraMessageHandler3;
@Autowired
private DirectChannel cassandraMessageHandler4;
@Autowired
private DirectChannel inputChannel;
@Autowired
private PollableChannel resultChannel;
@BeforeClass
public static void init() throws TTransportException, IOException, InterruptedException, ConfigurationException {
startCassandra();
}
private static void startCassandra()
throws TTransportException, IOException, InterruptedException, ConfigurationException {
EmbeddedCassandraServerHelper.startEmbeddedCassandra(CASSANDRA_CONFIG, "build/embeddedCassandra");
}
@AfterClass
public static void cleanup() {
EmbeddedCassandraServerHelper.cleanEmbeddedCassandra();
}
@Test
public void testBasicCassandraInsert() throws Exception {
Book b1 = BookSampler.getBook();
Message<Book> message = MessageBuilder.withPayload(b1).build();
cassandraMessageHandler1.send(message);
Select select = QueryBuilder.select().all().from("book");
List<Book> books = cassandraTemplate.select(select, Book.class);
assertEquals(1, books.size());
cassandraTemplate.delete(b1);
}
@Test
public void testCassandraBatchInsertAndSelectStatement() throws Exception {
List<Book> books = BookSampler.getBookList(5);
cassandraMessageHandler2.send(new GenericMessage<>(books));
Message<?> message = MessageBuilder.withPayload("Cassandra Puppy Guru").setHeader("limit", 2).build();
inputChannel.send(message);
Message<?> receive = resultChannel.receive(10000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(ResultSet.class));
ResultSet resultSet = (ResultSet) receive.getPayload();
assertNotNull(resultSet);
List<Row> rows = resultSet.all();
assertEquals(2, rows.size());
cassandraMessageHandler1.send(new GenericMessage<>(QueryBuilder.truncate("book")));
}
@Test
public void testCassandraBatchIngest() throws Exception {
List<Book> books = BookSampler.getBookList(5);
List<List<?>> ingestBooks = new ArrayList<>();
for (Book b : books) {
List<Object> l = new ArrayList<>();
l.add(b.getIsbn());
l.add(b.getTitle());
l.add(b.getAuthor());
l.add(b.getPages());
l.add(b.getSaleDate());
l.add(b.isInStock());
ingestBooks.add(l);
}
Message<List<List<?>>> message = MessageBuilder.withPayload(ingestBooks).build();
cassandraMessageHandler3.send(message);
Select select = QueryBuilder.select().all().from("book");
books = cassandraTemplate.select(select, Book.class);
assertEquals(5, books.size());
cassandraTemplate.delete(books);
}
@Test
public void testExpressionTrucante() throws Exception {
Message<Book> message = MessageBuilder.withPayload(BookSampler.getBook()).build();
cassandraMessageHandler1.send(message);
Select select = QueryBuilder.select().all().from("book");
List<Book> books = cassandraTemplate.select(select, Book.class);
assertEquals(1, books.size());
cassandraMessageHandler4.send(MessageBuilder.withPayload("Empty").build());
books = cassandraTemplate.select(select, Book.class);
assertEquals(0, books.size());
}
}

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

@@ -0,0 +1,64 @@
<?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:util="http://www.springframework.org/schema/util"
xmlns:cassandra="http://www.springframework.org/schema/data/cassandra"
xmlns:int-cassandra="http://www.springframework.org/schema/integration/cassandra"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/data/cassandra http://www.springframework.org/schema/data/cassandra/spring-cassandra.xsd
http://www.springframework.org/schema/integration/cassandra http://www.springframework.org/schema/integration/cassandra/spring-integration-cassandra.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.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.CassandraOperations" />
</bean>
<bean id="writeOptions" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cassandra.core.WriteOptions" />
</bean>
<int-cassandra:outbound-channel-adapter id="outbound1"
cassandra-template="cassandraTemplate"
write-options="writeOptions"
auto-startup="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>

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

@@ -0,0 +1,86 @@
/*
* Copyright 2016 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
*
* http://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.junit.Assert.assertEquals;
import org.junit.Test;
import org.junit.runner.RunWith;
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.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Filippo Balicchia
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class CassandraOutboundAdapterParserTests {
@Autowired
private ApplicationContext context;
@Test
public void minimalConfig() {
CassandraMessageHandler<?> handler = TestUtils.getPropertyValue(context.getBean("outbound1.adapter"), "handler",
CassandraMessageHandler.class);
assertEquals("outbound1.adapter", TestUtils.getPropertyValue(handler, "componentName"));
assertEquals(CassandraMessageHandler.Type.INSERT, TestUtils.getPropertyValue(handler, "mode"));
assertEquals(context.getBean("cassandraTemplate"), TestUtils.getPropertyValue(handler, "cassandraTemplate"));
assertEquals(context.getBean("writeOptions"), TestUtils.getPropertyValue(handler, "writeOptions"));
}
@Test
public void ingestConfig() {
CassandraMessageHandler<?> handler = TestUtils.getPropertyValue(context.getBean("outbound2"), "handler",
CassandraMessageHandler.class);
assertEquals("insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)",
TestUtils.getPropertyValue(handler, "ingestQuery"));
assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(handler, "producesReply"));
}
@Test
public void fullConfig() {
CassandraMessageHandler<?> handler = TestUtils.getPropertyValue(context.getBean("outgateway"), "handler",
CassandraMessageHandler.class);
assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(handler, "producesReply"));
assertEquals(CassandraMessageHandler.Type.STATEMENT, TestUtils.getPropertyValue(handler, "mode"));
assertEquals(context.getBean("writeOptions"), TestUtils.getPropertyValue(handler, "writeOptions"));
}
@Test
public void statementConfig() {
CassandraMessageHandler<?> handler = TestUtils.getPropertyValue(context.getBean("outbound4.adapter"), "handler",
CassandraMessageHandler.class);
assertEquals("outbound4.adapter", TestUtils.getPropertyValue(handler, "componentName"));
assertEquals(CassandraMessageHandler.Type.STATEMENT, TestUtils.getPropertyValue(handler, "mode"));
assertEquals(context.getBean("cassandraTemplate"), TestUtils.getPropertyValue(handler, "cassandraTemplate"));
assertEquals(context.getBean("writeOptions"), TestUtils.getPropertyValue(handler, "writeOptions"));
}
}

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

@@ -0,0 +1,96 @@
/*
* Copyright 2016 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
*
* http://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 org.junit.Assert;
import org.junit.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
*/
public class CassandraParserUtilsTests {
@Test
public void mutuallyExclusiveCase1() {
String query = "";
BeanDefinition statementExpressionDef = null;
String ingestQuery = "";
Assert.assertTrue(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
@Test
public void mutuallyExclusiveCase2() {
String query = "";
BeanDefinition statementExpressionDef = null;
String ingestQuery = "insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
Assert.assertTrue(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
@Test
public void mutuallyExclusiveCase3() {
String query = "";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery = "";
Assert.assertTrue(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
@Test
public void mutuallyExclusiveCase4() {
String query = "";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery = "insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
Assert.assertFalse(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
@Test
public void mutuallyExclusiveCase5() {
String query = "SELECT * FROM book limit :size";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery = "";
Assert.assertFalse(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
@Test
public 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 (?, ?, ?, ?, ?, ?)";
Assert.assertFalse(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
@Test
public void mutuallyExclusiveCase7() {
String query = "SELECT * FROM book limit :size";
BeanDefinition statementExpressionDef = new RootBeanDefinition();
String ingestQuery = "";
Assert.assertFalse(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
@Test
public 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 (?, ?, ?, ?, ?, ?)";
Assert.assertFalse(CassandraParserUtils.areMutuallyExclusive(query, statementExpressionDef, ingestQuery));
}
}

63
spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandlerTests.java
View File

@@ -27,8 +27,14 @@ import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.thrift.transport.TTransportException;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cassandra.core.ConsistencyLevel;
import org.springframework.cassandra.core.RetryPolicy;
@@ -40,6 +46,7 @@ import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.cassandra.config.IntegrationTestConfig;
import org.springframework.integration.cassandra.test.domain.Book;
import org.springframework.integration.cassandra.test.domain.BookSampler;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
@@ -52,14 +59,6 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.thrift.transport.TTransportException;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
@@ -87,7 +86,7 @@ public class CassandraMessageHandlerTests {
@Override
public String[] getEntityBasePackages() {
return new String[]{Book.class.getPackage().getName()};
return new String[] { Book.class.getPackage().getName() };
}
@Bean
@@ -120,7 +119,7 @@ public class CassandraMessageHandlerTests {
@Bean
public MessageHandler cassandraMessageHandler3() {
CassandraMessageHandler<Book> cassandraMessageHandler = new CassandraMessageHandler<>(this.template);
CassandraMessageHandler<Book> cassandraMessageHandler = new CassandraMessageHandler<>(this.template);
String cqlIngest = "insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
cassandraMessageHandler.setIngestQuery(cqlIngest);
return cassandraMessageHandler;
@@ -131,12 +130,12 @@ public class CassandraMessageHandlerTests {
return new QueueChannel();
}
@Bean
public MessageHandler cassandraMessageHandler4() {
CassandraMessageHandler<Book> cassandraMessageHandler = new CassandraMessageHandler<>(this.template);
//TODO https://jira.spring.io/browse/DATACASS-213
//cassandraMessageHandler.setQuery("SELECT * FROM book WHERE author = :author limit :size");
// TODO https://jira.spring.io/browse/DATACASS-213
// cassandraMessageHandler.setQuery("SELECT * FROM book WHERE author
// = :author limit :size");
cassandraMessageHandler.setQuery("SELECT * FROM book limit :size");
Map<String, Expression> params = new HashMap<>();
@@ -183,12 +182,10 @@ public class CassandraMessageHandlerTests {
protected static Session system;
@BeforeClass
public static void startCassandra() throws TTransportException, IOException, InterruptedException,
ConfigurationException {
public static void startCassandra()
throws TTransportException, IOException, InterruptedException, ConfigurationException {
EmbeddedCassandraServerHelper.startEmbeddedCassandra(CASSANDRA_CONFIG, "build/embeddedCassandra");
cluster = Cluster.builder()
.addContactPoint(IntegrationTestConfig.HOST)
.withPort(IntegrationTestConfig.PORT)
cluster = Cluster.builder().addContactPoint(IntegrationTestConfig.HOST).withPort(IntegrationTestConfig.PORT)
.build();
system = cluster.connect();
}
@@ -221,13 +218,11 @@ public class CassandraMessageHandlerTests {
@Test
public void testCassandraBatchInsertAndSelectStatement() throws Exception {
List<Book> books = getBookList(5);
List<Book> books = BookSampler.getBookList(5);
this.cassandraMessageHandler2.handleMessage(new GenericMessage<>(books));
Message<?> message = MessageBuilder.withPayload("Cassandra Guru")
.setHeader("limit", 2)
.build();
Message<?> message = MessageBuilder.withPayload("Cassandra Guru").setHeader("limit", 2).build();
this.cassandraMessageHandler4.handleMessage(message);
Message<?> receive = this.resultChannel.receive(10000);
@@ -243,7 +238,7 @@ public class CassandraMessageHandlerTests {
@Test
public void testCassandraBatchIngest() throws Exception {
List<Book> books = getBookList(5);
List<Book> books = BookSampler.getBookList(5);
List<List<?>> ingestBooks = new ArrayList<>();
for (Book b : books) {
@@ -266,24 +261,4 @@ public class CassandraMessageHandlerTests {
this.template.delete(books);
}
private List<Book> getBookList(int numBooks) {
List<Book> books = new ArrayList<>();
Book b;
for (int i = 0; i < numBooks; i++) {
b = new Book();
b.setIsbn(UUID.randomUUID().toString());
b.setTitle("Spring Data Cassandra Guide");
b.setAuthor("Cassandra Guru");
b.setPages(i * 10 + 5);
b.setInStock(true);
b.setSaleDate(new Date());
books.add(b);
}
return books;
}
}

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

@@ -0,0 +1,58 @@
/*
* Copyright 2016 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
*
* http://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.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;
/**
* @author Filippo Balicchia
*/
public class BookSampler {
public static List<Book> getBookList(int numBooks) {
List<Book> books = new ArrayList<>();
Book b;
for (int i = 0; i < numBooks; i++) {
b = new Book();
b.setIsbn(UUID.randomUUID().toString());
b.setTitle("Spring Data Cassandra Guide");
b.setAuthor("Cassandra Guru puppy");
b.setPages(i * 10 + 5);
b.setInStock(true);
b.setSaleDate(new Date());
books.add(b);
}
return books;
}
public static Book getBook() {
Book b1 = new Book();
b1.setIsbn("123456-1");
b1.setTitle("Spring Integration Cassandra");
b1.setAuthor("Cassandra Guru");
b1.setPages(521);
b1.setSaleDate(new Date());
b1.setInStock(true);
return b1;
}
}

4
spring-integration-cassandra/src/test/resources/cassandra.properties
View File

@@ -1,3 +1,3 @@
cassandra.contactpoints=127.0.0.1
cassandra.port=9042
cassandra.keyspace=demo
cassandra.port=9043
cassandra.keyspace=demo