From f7b31228c2839e2a2f77caf27bcc33c04efa4649 Mon Sep 17 00:00:00 2001
From: Filippo Balicchia <fbalicchia@gmail.com>
Date: Wed, 16 Dec 2015 07:43:27 +0100
Subject: [PATCH] INTEXT-215: Cassandra Namespace Support

JIRA: https://jira.spring.io/browse/INTEXT-215
---
 spring-integration-cassandra/build.gradle     |   2 +-
 .../config/xml/CassandraNamespaceHandler.java |   6 +-
 ...CassandraOutboundChannelAdapterParser.java |  40 ++++
 .../xml/CassandraOutboundGatewayParser.java   |  47 ++++
 .../config/xml/CassandraParserUtils.java      |  95 ++++++++
 .../outbound/CassandraMessageHandler.java     |  17 +-
 .../xml/spring-integration-cassandra-1.0.xsd  | 213 +++++++++++++++++-
 .../xml/spring-integration-cassandra.gif      | Bin 0 -> 578 bytes
 ...utboundAdapterIntegrationTests-context.xml |  87 +++++++
 ...sandraOutboundAdapterIntegrationTests.java | 166 ++++++++++++++
 ...ndraOutboundAdapterParserTests-context.xml |  64 ++++++
 .../CassandraOutboundAdapterParserTests.java  |  86 +++++++
 .../config/CassandraParserUtilsTests.java     |  96 ++++++++
 .../CassandraMessageHandlerTests.java         |  63 ++----
 .../cassandra/test/domain/BookSampler.java    |  58 +++++
 .../src/test/resources/cassandra.properties   |   4 +-
 16 files changed, 976 insertions(+), 68 deletions(-)
 create mode 100644 spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundChannelAdapterParser.java
 create mode 100644 spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundGatewayParser.java
 create mode 100644 spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraParserUtils.java
 create mode 100644 spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra.gif
 create mode 100644 spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests-context.xml
 create mode 100644 spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests.java
 create mode 100644 spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests-context.xml
 create mode 100644 spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests.java
 create mode 100644 spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraParserUtilsTests.java
 create mode 100644 spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/test/domain/BookSampler.java

diff --git a/spring-integration-cassandra/build.gradle b/spring-integration-cassandra/build.gradle
index bc65249..8f3893f 100644
--- a/spring-integration-cassandra/build.gradle
+++ b/spring-integration-cassandra/build.gradle
@@ -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'
 
diff --git a/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraNamespaceHandler.java b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraNamespaceHandler.java
index 45e524f..4044f1a 100644
--- a/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraNamespaceHandler.java
+++ b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraNamespaceHandler.java
@@ -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());
 	}
 
 }
diff --git a/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundChannelAdapterParser.java b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundChannelAdapterParser.java
new file mode 100644
index 0000000..e4c74f0
--- /dev/null
+++ b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundChannelAdapterParser.java
@@ -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();
+	}
+
+}
diff --git a/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundGatewayParser.java b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundGatewayParser.java
new file mode 100644
index 0000000..467b8ba
--- /dev/null
+++ b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraOutboundGatewayParser.java
@@ -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;
+	}
+
+}
diff --git a/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraParserUtils.java b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraParserUtils.java
new file mode 100644
index 0000000..8caf686
--- /dev/null
+++ b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/config/xml/CassandraParserUtils.java
@@ -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);
+	}
+
+}
diff --git a/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandler.java b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandler.java
index f0b134a..e1b8e19 100644
--- a/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandler.java
+++ b/spring-integration-cassandra/src/main/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandler.java
@@ -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,
diff --git a/spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra-1.0.xsd b/spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra-1.0.xsd
index eb7b9c3..5d686c2 100644
--- a/spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra-1.0.xsd
+++ b/spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra-1.0.xsd
@@ -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>
diff --git a/spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra.gif b/spring-integration-cassandra/src/main/resources/org/springframework/integration/config/xml/spring-integration-cassandra.gif
new file mode 100644
index 0000000000000000000000000000000000000000..750667e608fcaf4e40253ddf3505d59c2bdfcb44
GIT binary patch
literal 578
zcmV-I0=@l5Nk%w1VGsZi0OkMy00030|NnVQPjYW|Wnyk~b9G-}dtGRXTXMBraiCav
zyjgj<S#qX`m9A@YfMR!_ov6IK!_i%Pt6P1sOoY%@ipiXyrJJIuV||fVlCz7Mp-Y&-
zX_>5Kt;tik*L|a|uF=~>vd?g}#YDHzV2F`@y~K0D#e2uhjKRf}xwW*~)QiKor@*_l
z?$(5^oZ#x?((BIj{Oq&ReW=84gOF&XxOS$jmWhgdo19Cjt7@>Xn5U+aiHLc8dR?!q
zk?7yC;@!l|y^Gkyj?ur2>DHj<;LN_ER@uak;Lf7!-oNg`V(ihB?Z<nkhD)c3P^yem
z;nB0zz?|T-bC7E}jcGXU=-`E5GJ#z&jAl1^RV&J%aBWE?aZM+2O)2D~Vcw!;a7iYP
zVmQ~BVA-5xnQ=^NMJSPAJd|TT)skJkies*UU&n-0bx=2*g>=P$Rm6Q$pJqk2bx_8D
zSh;;%YCb2mbW^f)RI6=JvTj(ZY+9#fR+(H;mQ+&z|Nj6000000A^8LW004aeEC2ui
z01yBW000M{fPI34goTEOh<#2}R!xb5XJ158Z(WH`L}YndVsTrDQ(j9~YH4V2ZHHKR
zV@zOsb8=;eMto~-N_2X5Y=~KAb$2c)FDfV;hg@MAAW=nCEh8B=h7Bb#P&Gd*8xuFx
zA}~NRGC3p*27M6_4ebpbCnhvFJqiy6eMv}1LPJ3;Av|CZAR)m3AL`h-yYPWR1q2KT
Q?096s!2upR1O);BJ6#qKw*UYD

literal 0
HcmV?d00001

diff --git a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests-context.xml b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests-context.xml
new file mode 100644
index 0000000..536f115
--- /dev/null
+++ b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests-context.xml
@@ -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>
diff --git a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests.java b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests.java
new file mode 100644
index 0000000..7196bab
--- /dev/null
+++ b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterIntegrationTests.java
@@ -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());
+	}
+
+}
diff --git a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests-context.xml b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests-context.xml
new file mode 100644
index 0000000..8627f14
--- /dev/null
+++ b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests-context.xml
@@ -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>
diff --git a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests.java b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests.java
new file mode 100644
index 0000000..bfc74ad
--- /dev/null
+++ b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraOutboundAdapterParserTests.java
@@ -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"));
+
+	}
+
+}
diff --git a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraParserUtilsTests.java b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraParserUtilsTests.java
new file mode 100644
index 0000000..20bf91c
--- /dev/null
+++ b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/config/CassandraParserUtilsTests.java
@@ -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));
+	}
+
+}
diff --git a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandlerTests.java b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandlerTests.java
index b2e025b..0bd6539 100644
--- a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandlerTests.java
+++ b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/outbound/CassandraMessageHandlerTests.java
@@ -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;
-	}
-
 }
diff --git a/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/test/domain/BookSampler.java b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/test/domain/BookSampler.java
new file mode 100644
index 0000000..b7c031a
--- /dev/null
+++ b/spring-integration-cassandra/src/test/java/org/springframework/integration/cassandra/test/domain/BookSampler.java
@@ -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;
+	}
+
+}
diff --git a/spring-integration-cassandra/src/test/resources/cassandra.properties b/spring-integration-cassandra/src/test/resources/cassandra.properties
index 7cc737d..b1d7b86 100644
--- a/spring-integration-cassandra/src/test/resources/cassandra.properties
+++ b/spring-integration-cassandra/src/test/resources/cassandra.properties
@@ -1,3 +1,3 @@
 cassandra.contactpoints=127.0.0.1
-cassandra.port=9042
-cassandra.keyspace=demo
\ No newline at end of file
+cassandra.port=9043
+cassandra.keyspace=demo