Add spring-cloud-stream-tuple project

- This is a temporary location for the `tuple` project and would eventually be moved to something like `spring-data-commons`
    - Note: This project currently removes `batch` package from the tuple project in XD
 - The corresponding test classes are yet to be moved.

pom.xml

@@ -26,6 +26,7 @@
 		<module>spring-cloud-stream</module>
 		<module>spring-cloud-stream-binders</module>
 		<module>spring-cloud-stream-codec</module>
+		<module>spring-cloud-stream-tuple</module>
 		<module>spring-cloud-stream-starters</module>
 		<module>spring-cloud-stream-rxjava</module>
 		<module>spring-cloud-stream-samples</module>

spring-cloud-stream-tuple/pom.xml

@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-stream-tuple</artifactId>
+	<packaging>jar</packaging>
+	<name>spring-cloud-stream-tuple</name>
+	<description>Spring Cloud Stream Tuple</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-stream-parent</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>com.esotericsoftware</groupId>
+			<artifactId>kryo-shaded</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-context</artifactId>
+		</dependency>
+		<!-- TODO: Remove SI dependency by moving the corresponding classes in tuple to SI -->
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+		</dependency>
+		<!--<dependency>-->
+			<!--<groupId>org.springframework</groupId>-->
+			<!--<artifactId>spring-test</artifactId>-->
+		<!--</dependency>-->
+	</dependencies>
+</project>

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/DefaultTuple.java

@@ -0,0 +1,626 @@
+/*
+ * Copyright 2013-2015 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.cloud.stream.tuple;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.support.ConfigurableConversionService;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Default implementation of Tuple interface
+ * @author Mark Pollack
+ * @author David Turanski
+ * @author Michael Minella
+ */
+public class DefaultTuple implements Tuple {
+
+	private List<String> names;
+
+	private List<Object> values;
+
+	private transient ConfigurableConversionService configurableConversionService;
+
+	private transient Converter<Tuple, String> tupleToStringConverter = new TupleToJsonStringConverter();
+
+	public DefaultTuple(List<String> names, List<Object> values, ConfigurableConversionService 
+			configurableConversionService) {
+		Assert.notNull(names);
+		Assert.notNull(values);
+		Assert.notNull(configurableConversionService);
+		if (values.size() != names.size()) {
+			throw new IllegalArgumentException("Field names must be same length as values: names=" + names
+					+ ", values=" + values);
+		}
+		this.names = new ArrayList<>(names);
+		this.values = new ArrayList<>(values); // shallow copy
+		this.configurableConversionService = configurableConversionService;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.cloud.stream.tuple.Tuple#size()
+	 */
+	@Override
+	public int size() {
+		return values.size();
+	}
+
+	/**
+	 * Return the values for all the fields in this tuple
+	 * @return an unmodifiable List of names.
+	 */
+	@Override
+	public List<String> getFieldNames() {
+		return Collections.unmodifiableList(names);
+	}
+
+	/**
+	 * Return the values for all the fields in this tuple
+	 * @return an unmodifiable List list of values.
+	 */
+	@Override
+	public List<Object> getValues() {
+		return Collections.unmodifiableList(values);
+	}
+
+	@Override
+	public int getFieldCount() {
+		return this.names.size();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.cloud.stream.tuple.Tuple#hasName(java.lang.String)
+	 */
+	@Override
+	public boolean hasFieldName(String name) {
+		return names.contains(name);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.cloud.stream.tuple.Tuple#getValue(java.lang.String)
+	 */
+	@Override
+	public Object getValue(String name) {
+		int index = indexOf(name);
+		if (index == -1) {
+			throw new IllegalArgumentException("Field name [" + name + "] does not exist");
+		}
+		return getValue(index);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.cloud.stream.tuple.Tuple#getValue(int)
+	 */
+	@Override
+	public Object getValue(int index) {
+		return values.get(index);
+	}
+
+	@SuppressWarnings("rawtypes")
+	@Override
+	public List<Class> getFieldTypes() {
+		ArrayList<Class> types = new ArrayList<>(values.size());
+		for (Object val : values) {
+			types.add(val.getClass());
+		}
+		return Collections.unmodifiableList(types);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see java.lang.Object#hashCode()
+	 */
+	@Override
+	public int hashCode() {
+		final int prime = 31;
+		int result = 1;
+		result = prime * result + ((names == null) ? 0 : names.hashCode());
+		result = prime * result + ((values == null) ? 0 : values.hashCode());
+		return result;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see java.lang.Object#equals(java.lang.Object)
+	 */
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj) {
+			return true;
+		}
+		if (obj == null) {
+			return false;
+		}
+		if (!(obj instanceof DefaultTuple)) {
+			return false;
+		}
+		DefaultTuple other = (DefaultTuple) obj;
+		if (names == null) {
+			if (other.names != null) {
+				return false;
+			}
+		}
+		else if (!names.equals(other.names)) {
+			return false;
+		}
+		if (values == null) {
+			if (other.values != null) {
+				return false;
+			}
+		}
+		else if (!values.equals(other.values)) {
+			return false;
+		}
+		return true;
+	}
+
+	@Override
+	public String getString(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? null : getString(index);
+	}
+
+	@Override
+	public String getString(int index) {
+		return readAndTrim(index);
+	}
+
+	@Override
+	public Tuple getTuple(int index) {
+		return convert(values.get(index), Tuple.class);
+	}
+
+	@Override
+	public Tuple getTuple(String name) {
+		return getTuple(indexOf(name));
+	}
+
+	/**
+	 * @param index the index of the value
+	 * @return the converted raw value, trimmed
+	 */
+	private String readAndTrim(int index) {
+		Object rawValue = values.get(index);
+		if (rawValue != null) {
+			String value = convert(rawValue, String.class);
+			if (value != null) {
+				return value.trim();
+			}
+			else {
+				return null;
+			}
+		}
+		else {
+			return null;
+		}
+	}
+
+	@Override
+	public String getRawString(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? null : getRawString(index);
+	}
+
+	@Override
+	public String getRawString(int index) {
+		Object rawValue = values.get(index);
+		if (rawValue != null) {
+			String value = convert(rawValue, String.class);
+			if (value != null) {
+				return value;
+			}
+			else {
+				return null;
+			}
+		}
+		else {
+			return null;
+		}
+	}
+
+	@Override
+	public char getChar(int index) {
+		String value = readAndTrim(index);
+		if (value != null) {
+			Assert.isTrue(value.length() == 1, "Cannot convert field value '" + value + "' to char.");
+			return value.charAt(0);
+		}
+		return '\u0000';
+
+	}
+
+	@Override
+	public char getChar(String name) {
+		return getChar(indexOf(name));
+	}
+
+	@Override
+	public boolean getBoolean(int index) {
+		return getBoolean(index, "true");
+	}
+
+	@Override
+	public boolean getBoolean(String name) {
+		return getBoolean(indexOf(name));
+	}
+
+	@Override
+	public boolean getBoolean(int index, String trueValue) {
+		Assert.notNull(trueValue, "'trueValue' cannot be null.");
+		String value = readAndTrim(index);
+		return trueValue.equals(value);
+
+	}
+
+	@Override
+	public boolean getBoolean(String name, String trueValue) {
+		return getBoolean(indexOf(name), trueValue);
+	}
+
+	@Override
+	public byte getByte(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? 0 : getByte(index);
+	}
+
+	@Override
+	public byte getByte(int index) {
+		Byte b = convert(values.get(index), Byte.class);
+		return (b != null) ? b : 0;
+	}
+
+	@Override
+	public byte getByte(String name, byte defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getByte(index, defaultValue);
+	}
+
+	@Override
+	public byte getByte(int index, byte defaultValue) {
+		Byte b = convert(values.get(index), Byte.class);
+		return (b != null) ? b : defaultValue;
+	}
+
+	@Override
+	public short getShort(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? 0 : getShort(index);
+	}
+
+	@Override
+	public short getShort(int index) {
+		Short s = convert(values.get(index), Short.class);
+		return (s != null) ? s : 0;
+	}
+
+	@Override
+	public short getShort(String name, short defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getShort(index, defaultValue);
+	}
+
+	@Override
+	public short getShort(int index, short defaultValue) {
+		Short s = convert(values.get(index), Short.class);
+		return (s != null) ? s : defaultValue;
+	}
+
+	@Override
+	public int getInt(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? 0 : getInt(index);
+	}
+
+	@Override
+	public int getInt(int index) {
+		Integer i = convert(values.get(index), Integer.class);
+		return (i != null) ? i : 0;
+	}
+
+	@Override
+	public int getInt(String name, int defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getInt(index, defaultValue);
+	}
+
+	@Override
+	public int getInt(int index, int defaultValue) {
+		Integer i = convert(values.get(index), Integer.class);
+		return (i != null) ? i : defaultValue;
+	}
+
+	@Override
+	public long getLong(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? 0 : getLong(index);
+	}
+
+	@Override
+	public long getLong(int index) {
+		Long l = convert(values.get(index), Long.class);
+		return (l != null) ? l : 0;
+	}
+
+	@Override
+	public long getLong(String name, long defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getLong(index, defaultValue);
+	}
+
+	@Override
+	public long getLong(int index, long defaultValue) {
+		Long l = convert(values.get(index), Long.class);
+		return (l != null) ? l : defaultValue;
+	}
+
+	@Override
+	public float getFloat(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? 0 : getFloat(index);
+	}
+
+	@Override
+	public float getFloat(int index) {
+		Float f = convert(values.get(index), Float.class);
+		return (f != null) ? f : 0;
+	}
+
+	@Override
+	public float getFloat(String name, float defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getFloat(index, defaultValue);
+	}
+
+	@Override
+	public float getFloat(int index, float defaultValue) {
+		Float f = convert(values.get(index), Float.class);
+		return (f != null) ? f : defaultValue;
+	}
+
+	@Override
+	public double getDouble(String name) {
+		int index = indexOf(name);
+		return (index == -1) ? 0 : getDouble(index);
+	}
+
+	@Override
+	public double getDouble(int index) {
+		Double d = convert(values.get(index), Double.class);
+		return (d != null) ? d : 0;
+	}
+
+	@Override
+	public double getDouble(String name, double defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getDouble(index, defaultValue);
+	}
+
+	@Override
+	public double getDouble(int index, double defaultValue) {
+		Double d = convert(values.get(index), Double.class);
+		return (d != null) ? d : defaultValue;
+	}
+
+	@Override
+	public BigDecimal getBigDecimal(String name) {
+		return getBigDecimal(indexOf(name));
+	}
+
+	@Override
+	public BigDecimal getBigDecimal(int index) {
+		return convert(values.get(index), BigDecimal.class);
+	}
+
+	@Override
+	public BigDecimal getBigDecimal(String name, BigDecimal defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getBigDecimal(index, defaultValue);
+	}
+
+	@Override
+	public BigDecimal getBigDecimal(int index, BigDecimal defaultValue) {
+		BigDecimal bd = convert(values.get(index), BigDecimal.class);
+		return (bd != null) ? bd : defaultValue;
+	}
+
+	@Override
+	public Date getDate(int index) {
+		return convert(values.get(index), Date.class);
+	}
+
+	@Override
+	public Date getDate(String name) {
+		return getDate(indexOf(name));
+	}
+
+	@Override
+	public Date getDate(String name, Date defaultValue) {
+		int index = indexOf(name);
+		return (index == -1) ? defaultValue : getDate(index, defaultValue);
+	}
+
+	@Override
+	public Date getDate(int index, Date defaultValue) {
+		Date d = getDate(index);
+		return (d != null) ? d : defaultValue;
+	}
+
+	@Override
+	public Date getDateWithPattern(int index, String pattern) {
+		StringToDateConverter converter = new StringToDateConverter(pattern);
+		return converter.convert(this.readAndTrim(index));
+	}
+
+	@Override
+	public Date getDateWithPattern(String name, String pattern) {
+		try {
+			return getDateWithPattern(indexOf(name), pattern);
+		}
+		catch (IllegalArgumentException e) {
+			throw new IllegalArgumentException(e.getMessage() + ", name: [" + name + "]");
+		}
+	}
+
+	@Override
+	public Date getDateWithPattern(int index, String pattern, Date defaultValue) {
+		try {
+			Date d = getDateWithPattern(index, pattern);
+			return (d != null) ? d : defaultValue;
+		}
+		catch (IllegalArgumentException e) {
+			return defaultValue;
+		}
+
+	}
+
+	@Override
+	public Date getDateWithPattern(String name, String pattern, Date defaultValue) {
+		try {
+			return getDateWithPattern(indexOf(name), pattern, defaultValue);
+		}
+		catch (IllegalArgumentException e) {
+			throw new IllegalArgumentException(e.getMessage() + ", name: [" + name + "]");
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.cloud.stream.tuple.Tuple#getValue(java.lang.String, java.lang.Class)
+	 */
+	@Override
+	public <T> T getValue(String name, Class<T> valueClass) {
+		Object value = values.get(indexOf(name));
+		return convert(value, valueClass);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.cloud.stream.tuple.Tuple#getValue(int, java.lang.Class)
+	 */
+	@Override
+	public <T> T getValue(int index, Class<T> valueClass) {
+		return convert(values.get(index), valueClass);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Tuple select(String expression) {
+		EvaluationContext context = new StandardEvaluationContext(toMap());
+		ExpressionParser parser = new SpelExpressionParser();
+		Expression exp = parser.parseExpression(expression);
+
+		Object result = exp.getValue(context);
+		Map<String, Object> resultMap = null;
+		if (ClassUtils.isAssignableValue(Map.class, result)) {
+			resultMap = (Map<String, Object>) result;
+		}
+		if (resultMap != null) {
+			return toTuple(resultMap);
+		}
+		else {
+			return new DefaultTuple(new ArrayList<String>(0), new ArrayList<>(0),
+					this.configurableConversionService);
+		}
+	}
+
+	/**
+	 * @return names and values as a {@code Map<String, Object>}
+	 */
+	Map<String, Object> toMap() {
+		Map<String, Object> map = new LinkedHashMap<>(values.size());
+		for (int i = 0; i < values.size(); i++) {
+			map.put(names.get(i), values.get(i));
+		}
+		return map;
+	}
+
+	Tuple toTuple(Map<String, Object> resultMap) {
+
+		List<String> newNames = new ArrayList<>();
+		List<Object> newValues = new ArrayList<>();
+		for (String name : resultMap.keySet()) {
+			newNames.add(name);
+		}
+		for (Object value : resultMap.values()) {
+			newValues.add(value);
+		}
+		return new DefaultTuple(newNames, newValues, this.configurableConversionService);
+
+	}
+
+	@SuppressWarnings("unchecked")
+	<T> T convert(Object value, Class<T> targetType) {
+		return (T) configurableConversionService.convert(value, TypeDescriptor.forObject(value),
+				TypeDescriptor.valueOf(targetType));
+	}
+
+	/**
+	 * Find the index in the names collection for the given name.
+	 * Returns -1 if not found.
+	 */
+	protected int indexOf(String name) {
+		return names.indexOf(name);
+	}
+
+	/**
+	 * @param tupleToStringConverter used to convert a {@link Tuple} to a String
+	 */
+	protected void setTupleToStringConverter(Converter<Tuple, String> tupleToStringConverter) {
+		Assert.notNull(tupleToStringConverter, "tupleToStringConverter cannot be null");
+		this.tupleToStringConverter = tupleToStringConverter;
+	}
+
+	@Override
+	/**
+	 * The format is of the form DefaultTuple [names="n1'
+	 */
+	public String toString() {
+		return tupleToStringConverter.convert(this);
+	}
+
+	
+	public ConfigurableConversionService getConversionService() {
+		return this.configurableConversionService;
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/DefaultTupleConversionService.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2013 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.cloud.stream.tuple;
+
+import java.util.Collection;
+
+import org.springframework.core.convert.ConversionService;
+import org.springframework.format.support.DefaultFormattingConversionService;
+
+/**
+ * Base {@link ConversionService} implementation suitable for use with {@link Tuple}
+ * 
+ * @author David Turanski
+ * 
+ */
+public class DefaultTupleConversionService extends DefaultFormattingConversionService {
+
+	public DefaultTupleConversionService() {
+		/*
+		 * DefaultFormattingConversionService provides Collection -> Object conversion which will produce the first item
+		 * if the target type matches. Here, this results in an unfortunate side effect, getTuple(List<Tuple> list)
+		 * would return a Tuple. In this case it is preferable to treat it as an error if the argument is not a Tuple.
+		 */
+		removeConvertible(Collection.class, Object.class);
+	}
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/JsonBytesToTupleConverter.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple;
+
+import org.springframework.core.convert.converter.Converter;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * @author David Turanski
+ * 
+ */
+public class JsonBytesToTupleConverter implements Converter<byte[], Tuple> {
+
+	private final ObjectMapper mapper = new ObjectMapper();
+
+	private final JsonNodeToTupleConverter jsonNodeToTupleConverter = new JsonNodeToTupleConverter();
+
+	public JsonBytesToTupleConverter() {
+		mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
+	}
+
+	@Override
+	public Tuple convert(byte[] source) {
+		if (source == null) {
+			return null;
+		}
+		try {
+			return jsonNodeToTupleConverter.convert(mapper.readTree(source));
+		}
+		catch (Exception e) {
+			throw new IllegalArgumentException(e.getMessage(), e);
+		}
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/JsonNodeToTupleConverter.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+
+import org.springframework.core.convert.converter.Converter;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * @author David Turanski
+ *
+ */
+public class JsonNodeToTupleConverter implements Converter<JsonNode, Tuple> {
+
+	@Override
+	public Tuple convert(JsonNode root) {
+		TupleBuilder builder = TupleBuilder.tuple();
+		if (root.isValueNode()) {
+			return builder.of("value", root.asText());
+		}
+		try {
+			for (Iterator<Entry<String, JsonNode>> it = root.fields(); it.hasNext();) {
+				Entry<String, JsonNode> entry = it.next();
+				String name = entry.getKey();
+				JsonNode node = entry.getValue();
+				if (node.isObject()) {
+					// tuple
+					builder.addEntry(name, convert(node));
+				}
+				else if (node.isArray()) {
+					builder.addEntry(name, nodeToList(node));
+				}
+				else {
+					if (name.equals("id")) {//NOSONAR
+						// TODO how should this be handled?
+					}
+					else if (name.equals("timestamp")) {//NOSONAR
+						// TODO how should this be handled?
+					}
+					else {
+						builder.addEntry(name, node.asText());
+					}
+				}
+			}
+		}
+		catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+		return builder.build();
+	}
+
+	private List<Object> nodeToList(JsonNode node) {
+		List<Object> list = new ArrayList<Object>(node.size());
+		for (int i = 0; i < node.size(); i++) {
+			JsonNode item = node.get(i);
+			if (item.isObject()) {
+				list.add(convert(item));
+			}
+			else if (item.isArray()) {
+				list.add(nodeToList(item));
+			}
+			else {
+				list.add(item.asText());
+			}
+		}
+		return list;
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/JsonStringToTupleConverter.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2002-2013 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.cloud.stream.tuple;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+
+import org.springframework.core.convert.converter.Converter;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * @author David Turanski
+ *
+ */
+public class JsonStringToTupleConverter implements Converter<String, Tuple> {
+
+	private final ObjectMapper mapper = new ObjectMapper();
+
+	public JsonStringToTupleConverter() {
+		mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
+	}
+
+	@Override
+	public Tuple convert(String source) {
+		TupleBuilder builder = TupleBuilder.tuple();
+		try {
+
+			JsonNode root = mapper.readTree(source);
+			for (Iterator<Entry<String, JsonNode>> it = root.fields(); it.hasNext();) {
+				Entry<String, JsonNode> entry = it.next();
+				String name = entry.getKey();
+				JsonNode node = entry.getValue();
+				if (node.isObject()) {
+					// tuple
+					builder.addEntry(name, convert(node.toString()));
+				}
+				else if (node.isArray()) {
+					builder.addEntry(name, nodeToList(node));
+				}
+				else {
+					if (name.equals("id")) {//NOSONAR
+						// TODO how should this be handled?
+					}
+					else if (name.equals("timestamp")) {//NOSONAR
+						// TODO how should this be handled?
+					}
+					else {
+						builder.addEntry(name, node.asText());
+					}
+				}
+			}
+		}
+		catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+		return builder.build();
+	}
+
+	private List<Object> nodeToList(JsonNode node) {
+		List<Object> list = new ArrayList<Object>(node.size());
+		for (int i = 0; i < node.size(); i++) {
+			JsonNode item = node.get(i);
+			if (item.isObject()) {
+				list.add(convert(item.toString()));
+			}
+			else if (item.isArray()) {
+				list.add(nodeToList(item));
+			}
+			else {
+				list.add(item.asText());
+			}
+		}
+		return list;
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/LocaleAwareStringToNumberConverterFactory.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple;
+
+import java.text.NumberFormat;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.converter.ConverterFactory;
+import org.springframework.util.NumberUtils;
+
+/**
+ * Converts from a String any JDK-standard Number implementation.
+ * 
+ * <p>
+ * Support Number classes including Byte, Short, Integer, Float, Double, Long, BigInteger, BigDecimal. This class
+ * delegates to {@link NumberUtils#parseNumber(String, Class, NumberFormat)} to perform the conversion.
+ * 
+ * @author Mark Pollack
+ * @author Keith Donald
+ * 
+ * @see Byte
+ * @see Short
+ * @see Integer
+ * @see Long
+ * @see java.math.BigInteger
+ * @see Float
+ * @see Double
+ * @see java.math.BigDecimal
+ * @see NumberUtils
+ */
+public class LocaleAwareStringToNumberConverterFactory implements ConverterFactory<String, Number> {
+
+	private NumberFormat numberFormat;
+
+	public LocaleAwareStringToNumberConverterFactory(NumberFormat numberFormat) {
+		this.numberFormat = numberFormat;
+	}
+
+	@Override
+	public <T extends Number> Converter<String, T> getConverter(Class<T> targetType) {
+		return new StringToNumber<T>(targetType, numberFormat);
+	}
+
+	private static final class StringToNumber<T extends Number> implements Converter<String, T> {
+
+		private final Class<T> targetType;
+
+		private NumberFormat numberFormat;
+
+		public StringToNumber(Class<T> targetType, NumberFormat numberFormat) {
+			this.targetType = targetType;
+			this.numberFormat = numberFormat;
+		}
+
+		@Override
+		public T convert(String source) {
+			return NumberUtils.parseNumber(source, this.targetType, numberFormat);
+		}
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/StringToDateConverter.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple;
+
+import java.text.DateFormat;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import org.springframework.core.convert.converter.Converter;
+
+/**
+ * Converter for Strings to Date that can take into account date patterns
+ * 
+ * @author Mark Pollack
+ * 
+ */
+public class StringToDateConverter implements Converter<String, Date> {
+
+	private final static String DEFAULT_DATE_PATTERN = "yyyy-MM-dd";
+
+	private DateFormat dateFormat;
+
+	public StringToDateConverter() {
+		this.dateFormat = new SimpleDateFormat(DEFAULT_DATE_PATTERN);
+		this.dateFormat.setLenient(false);
+
+	}
+
+	public StringToDateConverter(String pattern) {
+		this.dateFormat = new SimpleDateFormat(pattern);
+		this.dateFormat.setLenient(false);
+	}
+
+	public StringToDateConverter(DateFormat dateFormat) {
+		this.dateFormat = dateFormat;
+	}
+
+	@Override
+	public Date convert(String source) {
+		try {
+			return dateFormat.parse(source);
+		}
+		catch (ParseException e) {
+			String pattern;
+			if (dateFormat instanceof SimpleDateFormat) {
+				pattern = ((SimpleDateFormat) dateFormat).toPattern();
+			}
+			else {
+				pattern = dateFormat.toString();
+			}
+			throw new IllegalArgumentException(e.getMessage() + ", format: [" + pattern + "]");
+		}
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/Tuple.java

@@ -0,0 +1,558 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple;
+
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.List;
+
+import org.springframework.core.convert.ConversionFailedException;
+
+/**
+ * Data structure that stores a fixed number of ordered key-value pairs and adds convenience methods to access values in
+ * a type-safe manner.
+ * 
+ * The structure is immutable once created and values to not need to be of the same type. When accessing values,
+ * Spring's type conversion system is used to convert the value type to the requested type. The type conversion system
+ * is extensible.
+ * 
+ * Tuples are created using the TupleBuilder class.
+ * 
+ * @author Mark Pollack
+ * 
+ */
+public interface Tuple {
+
+	/**
+	 * Return the number of elements in this tuple.
+	 * 
+	 * @return number of elements
+	 */
+	int size();
+
+	/**
+	 * Return the fields names that can reference elements by name in this tuple
+	 * 
+	 * @return list of field names
+	 */
+	List<String> getFieldNames();
+
+	/**
+	 * Return the values for all the fields in this tuple
+	 * 
+	 * @return list of values.
+	 */
+	List<Object> getValues();
+
+	/**
+	 * Return true if the tuple contains a field of the given name
+	 * 
+	 * @param name the name of the field
+	 * @return true if present, otherwise false
+	 */
+	boolean hasFieldName(String name);
+
+	/**
+	 * Return the Java types of the fields in this tuple.
+	 * 
+	 * @return the Java types of the fields in this tuple.
+	 */
+	@SuppressWarnings("rawtypes")
+	List<Class> getFieldTypes();
+
+	/**
+	 * Return the number of fields in this tuple.
+	 * 
+	 * @return the number of fields in this tuple.
+	 */
+	int getFieldCount();
+
+	/**
+	 * Return the value of the field given the name
+	 * 
+	 * @param name the name of the field
+	 * @return value of the field
+	 * @throws IllegalArgumentException if the name is not present
+	 */
+	Object getValue(String name);
+
+	/**
+	 * Return the value of the field given the index position
+	 * 
+	 * @param index position in the tuple
+	 * @return value of the field
+	 * @throws IndexOutOfBoundsException if the index position is out of bounds.
+	 */
+	Object getValue(int index);
+
+	/**
+	 * Return the value of the field given the name
+	 * 
+	 * @param name the field name
+	 * @param valueClass Class to coerce the value into.
+	 * @return value of the field
+	 */
+	<T> T getValue(String name, Class<T> valueClass);
+
+	/**
+	 * Return the value of the field given the index position
+	 * 
+	 * @param index position in the tuple
+	 * @param valueClass Class to coerce the value into
+	 * @return value of the field
+	 */
+	<T> T getValue(int index, Class<T> valueClass);
+
+	/**
+	 * Read the {@link String} value given the field '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 * @return value of the field
+	 */
+	String getString(String name);
+
+	/**
+	 * Read the String value given the index position
+	 * 
+	 * @param index position in the tuple
+	 * @return value of the field
+	 */
+	String getString(int index);
+
+	/**
+	 * Read the {@link Tuple} value given the field '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 * @return value of the field
+	 */
+	Tuple getTuple(String name);
+
+	/**
+	 * Read the Tuple value given the index position
+	 * 
+	 * @param index position in the tuple
+	 * @return value of the field
+	 */
+	Tuple getTuple(int index);
+
+	/**
+	 * Read the {@link String} value at index '<code>index</code>' including trailing whitespace (don't trim).
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	String getRawString(int index);
+
+	/**
+	 * Read the {@link String} value from column with given '<code>name</code>' including trailing whitespace (don't
+	 * trim).
+	 * 
+	 * @param name the field name.
+	 */
+	String getRawString(String name);
+
+	/**
+	 * Read the '<code>char</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	char getChar(int index);
+
+
+	/**
+	 * Read the '<code>char</code>' value from field with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 * @throws IllegalArgumentException if a field with given name is not defined.
+	 */
+	char getChar(String name);
+
+	/**
+	 * Read the '<code>boolean</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	boolean getBoolean(int index);
+
+	/**
+	 * Read the '<code>boolean</code>' value from field with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 * @throws IllegalArgumentException if a field with given name is not defined.
+	 */
+	boolean getBoolean(String name);
+
+	/**
+	 * Read the '<code>boolean</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @param trueValue the value that signifies {@link Boolean#TRUE true}; case-sensitive.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds, or if the supplied <code>trueValue</code> is
+	 *         <code>null</code>.
+	 */
+	boolean getBoolean(int index, String trueValue);
+
+	/**
+	 * Read the '<code>boolean</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 * @param trueValue the value that signifies {@link Boolean#TRUE true}; case-sensitive.
+	 * @throws IllegalArgumentException if a column with given name is not defined, or if the supplied
+	 *         <code>trueValue</code> is <code>null</code>.
+	 */
+	boolean getBoolean(String name, String trueValue);
+
+	/**
+	 * Read the '<code>byte</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	byte getByte(int index);
+
+	/**
+	 * Read the '<code>byte</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 */
+	byte getByte(String name);
+
+
+	/**
+	 * Read the '<code>byte</code>' value at index '<code>index</code>'. using the supplied <code>defaultValue</code> if
+	 * the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	byte getByte(int index, byte defaultValue);
+
+	/**
+	 * Read the '<code>byte</code>' value from column with given '<code>name</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 */
+	byte getByte(String name, byte defaultValue);
+
+	/**
+	 * Read the '<code>short</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	short getShort(int index);
+
+	/**
+	 * Read the '<code>short</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 */
+	short getShort(String name);
+
+	/**
+	 * Read the '<code>short</code>' value at index '<code>index</code>'. using the supplied <code>defaultValue</code>
+	 * if the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	short getShort(int index, short defaultValue);
+
+	/**
+	 * Read the '<code>short</code>' value from column with given '<code>name</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 */
+	short getShort(String name, short defaultValue);
+
+	/**
+	 * Read the '<code>int</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	int getInt(int index);
+
+	/**
+	 * Read the '<code>int</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 */
+	int getInt(String name);
+
+	/**
+	 * Read the '<code>int</code>' value at index '<code>index</code>'. using the supplied <code>defaultValue</code> if
+	 * the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	int getInt(int index, int defaultValue);
+
+	/**
+	 * Read the '<code>int</code>' value from column with given '<code>name</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 */
+	int getInt(String name, int defaultValue);
+
+	/**
+	 * Read the '<code>long</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	long getLong(int index);
+
+	/**
+	 * Read the '<code>int</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 */
+	long getLong(String name);
+
+	/**
+	 * Read the '<code>long</code>' value at index '<code>index</code>'. using the supplied <code>defaultValue</code> if
+	 * the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	long getLong(int index, long defaultValue);
+
+	/**
+	 * Read the '<code>long</code>' value from column with given '<code>name</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 */
+	long getLong(String name, long defaultValue);
+
+	/**
+	 * Read the '<code>float</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	float getFloat(int index);
+
+	/**
+	 * Read the '<code>float</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 */
+	float getFloat(String name);
+
+	/**
+	 * Read the '<code>float</code>' value at index '<code>index</code>'. using the supplied <code>defaultValue</code>
+	 * if the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	float getFloat(int index, float defaultValue);
+
+	/**
+	 * Read the '<code>float</code>' value from column with given '<code>name</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 */
+	float getFloat(String name, float defaultValue);
+
+
+	/**
+	 * Read the '<code>double</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	double getDouble(int index);
+
+	/**
+	 * Read the '<code>double</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 */
+	double getDouble(String name);
+
+
+	/**
+	 * Read the '<code>double</code>' value at index '<code>index</code>'. using the supplied <code>defaultValue</code>
+	 * if the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	double getDouble(int index, double defaultValue);
+
+	/**
+	 * Read the '<code>double</code>' value from column with given '<code>name</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 */
+	double getDouble(String name, double defaultValue);
+
+	/**
+	 * Read the '<code>BigDecimal</code>' value at index '<code>index</code>'.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	BigDecimal getBigDecimal(int index);
+
+	/**
+	 * Read the '<code>BigDecimal</code>' value from column with given '<code>name</code>'.
+	 * 
+	 * @param name the field name.
+	 */
+	BigDecimal getBigDecimal(String name);
+
+	/**
+	 * Read the '<code>BigDecimal</code>' value at index '<code>index</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 */
+	BigDecimal getBigDecimal(int index, BigDecimal defaultValue);
+
+	/**
+	 * Read the '<code>BigDecimal</code>' value from column with given '<code>name</code>'. using the supplied
+	 * <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 */
+	BigDecimal getBigDecimal(String name, BigDecimal defaultValue);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in default format at designated column <code>index</code>.
+	 * 
+	 * @param index the field index.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 * @throws ConversionFailedException if the value is not parseable
+	 */
+	Date getDate(int index);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in default format at designated column with given <code>name</code>.
+	 * 
+	 * @param name the field name.
+	 * @throws IllegalArgumentException if a column with given name is not defined
+	 * @throws ConversionFailedException if the value is not parseable
+	 */
+	Date getDate(String name);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in default format at designated column <code>index</code> using the
+	 * supplied <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 * @throws ConversionFailedException if the value is not parseable
+	 */
+	Date getDate(int index, Date defaultValue);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in default format at designated column with given <code>name</code>.
+	 * using the supplied <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IllegalArgumentException if a column with given name is not defined
+	 * @throws ConversionFailedException if the value is not parseable
+	 */
+	Date getDate(String name, Date defaultValue);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in default format at designated column <code>index</code>.
+	 * 
+	 * @param index the field index.
+	 * @param pattern the pattern describing the date and time format
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 * @throws IllegalArgumentException if the date cannot be parsed.
+	 * 
+	 */
+	Date getDateWithPattern(int index, String pattern);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in given format from column with given <code>name</code>.
+	 * 
+	 * @param name the field name.
+	 * @param pattern the pattern describing the date and time format
+	 * @throws IllegalArgumentException if a column with given name is not defined or if the specified field cannot be
+	 *         parsed
+	 * 
+	 */
+	Date getDateWithPattern(String name, String pattern);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in default format at designated column <code>index</code>. using the
+	 * supplied <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param index the field index.
+	 * @param pattern the pattern describing the date and time format
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IndexOutOfBoundsException if the index is out of bounds.
+	 * @throws IllegalArgumentException if the date cannot be parsed.
+	 * 
+	 */
+	Date getDateWithPattern(int index, String pattern, Date defaultValue);
+
+	/**
+	 * Read the <code>java.util.Date</code> value in given format from column with given <code>name</code>. using the
+	 * supplied <code>defaultValue</code> if the field value is a zero length string or null.
+	 * 
+	 * @param name the field name.
+	 * @param pattern the pattern describing the date and time format
+	 * @param defaultValue the default value to return if field value is not found.
+	 * @throws IllegalArgumentException if a column with given name is not defined or if the specified field cannot be
+	 *         parsed
+	 * 
+	 */
+	Date getDateWithPattern(String name, String pattern, Date defaultValue);
+
+	/**
+	 * Use SpEL expression to return a subset of the tuple that matches the expression
+	 * 
+	 * @param expression SpEL expression to select from a Map, e.g. ?[key.startsWith('b')]
+	 * @return a new Tuple with data selected from the current instance.
+	 */
+	Tuple select(String expression);
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleBuilder.java

@@ -0,0 +1,217 @@
+/*
+ * Copyright 2013-2015 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.cloud.stream.tuple;
+
+import java.text.DateFormat;
+import java.text.NumberFormat;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.core.convert.support.ConfigurableConversionService;
+import org.springframework.util.AlternativeJdkIdGenerator;
+import org.springframework.util.Assert;
+import org.springframework.util.IdGenerator;
+
+/**
+ * Builder class to create Tuple instances.
+ * 
+ * Default Locale is US for NumberFormat and default DatePattern is "yyyy-MM-dd"
+ *
+ * <em>Note:</em> Using a custom conversion service that is instance based (not configured
+ * as a singleton) will have <em>significant</em> performance impacts.
+ * 
+ * @author Mark Pollack
+ * @author David Turanski
+ * @author Michael Minella
+ * @author Gunnar Hillert
+ * 
+ */
+public class TupleBuilder {
+
+	private List<String> names = new ArrayList<>();
+
+	private List<Object> values = new ArrayList<>();
+
+	private static final ConfigurableConversionService defaultConversionService;
+
+	private ConfigurableConversionService customConversionService = null;
+
+	private final static String DEFAULT_DATE_PATTERN = "yyyy-MM-dd";
+
+	private final static Locale DEFAULT_LOCALE = Locale.US;
+
+	private static Converter<Tuple, String> tupleToStringConverter = new TupleToJsonStringConverter();
+
+	private static Converter<String, Tuple> stringToTupleConverter = new JsonStringToTupleConverter();
+
+	private static final IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator();
+
+	static {
+		defaultConversionService = new DefaultTupleConversionService();
+		defaultConversionService.addConverterFactory(new LocaleAwareStringToNumberConverterFactory(NumberFormat
+				.getInstance(DEFAULT_LOCALE)));
+		DateFormat dateFormat = new SimpleDateFormat(DEFAULT_DATE_PATTERN);
+		dateFormat.setLenient(false);
+		defaultConversionService.addConverter(new StringToDateConverter(dateFormat));
+	}
+
+	public static TupleBuilder tuple() {
+		return new TupleBuilder();
+	}
+
+	public Tuple of(String k1, Object v1) {
+		return newTuple(namesOf(k1), valuesOf(v1));
+	}
+
+	public Tuple of(String k1, Object v1, String k2, Object v2) {
+		addEntry(k1, v1);
+		addEntry(k2, v2);
+		return build();
+	}
+
+	public Tuple of(String k1, Object v1, String k2, Object v2, String k3, Object v3) {
+		addEntry(k1, v1);
+		addEntry(k2, v2);
+		addEntry(k3, v3);
+		return build();
+	}
+
+	public Tuple of(String k1, Object v1, String k2, Object v2, String k3, Object v3, String k4, Object v4) {
+		addEntry(k1, v1);
+		addEntry(k2, v2);
+		addEntry(k3, v3);
+		addEntry(k4, v4);
+		return build();
+	}
+
+	public Tuple ofNamesAndValues(List<String> names, List<Object> values) {
+		this.names = names;
+		this.values = values;
+		return build();
+	}
+
+	public TupleBuilder put(String k1, Object v1) {
+		addEntry(k1, v1);
+		return this;
+	}
+
+	/**
+	 * Add all names and values of the tuple to the built tuple.
+	 * @param tuple names and value to add to the tuple being built
+	 * @return builder to continue creating a new tuple instance
+	 */
+	public TupleBuilder putAll(Tuple tuple) {
+		for (int i = 0; i < tuple.size(); i++) {
+			Object value = tuple.getValues().get(i);
+			String name = tuple.getFieldNames().get(i);
+			addEntry(name, value);
+		}
+		return this;
+	}
+
+	public Tuple build() {
+		return newTuple(names, values);
+	}
+
+	public static Tuple fromString(String source) {
+		return stringToTupleConverter.convert(source);
+	}
+
+	public TupleBuilder setConfigurableConversionService(ConfigurableConversionService formattingConversionService) {
+		Assert.notNull(formattingConversionService);
+		this.customConversionService = formattingConversionService;
+		return this;
+	}
+
+	public ConversionServiceBuilder setFormats(Locale locale, DateFormat dateFormat) {
+		return new ConversionServiceBuilder(this, locale, dateFormat);
+	}
+
+	void addEntry(String k1, Object v1) {
+		names.add(k1);
+		values.add(v1);
+	}
+
+	static List<Object> valuesOf(Object v1) {
+		ArrayList<Object> values = new ArrayList<>();
+		values.add(v1);
+		return Collections.unmodifiableList(values);
+	}
+
+	static List<String> namesOf(String k1) {
+		List<String> fields = new ArrayList<>(1);
+		fields.add(k1);
+		return Collections.unmodifiableList(fields);
+
+	}
+
+	protected Tuple newTuple(List<String> names, List<Object> values) {
+		DefaultTuple tuple;
+
+		if(customConversionService != null) {
+			tuple = new DefaultTuple(names, values, customConversionService);
+		}
+		else {
+			tuple = new DefaultTuple(names, values, defaultConversionService);
+		}
+
+		tuple.setTupleToStringConverter(tupleToStringConverter);
+		return tuple;
+	}
+
+	/**
+	 * Provides the ability to inject a {@link ConfigurableConversionService} as a way to
+	 * customize conversion behavior in the built {@link Tuple}.
+	 *
+	 * @author Michael Minella
+	 */
+	public static class ConversionServiceBuilder {
+
+		private TupleBuilder builder;
+
+		private Locale locale;
+
+		private DateFormat dateFormat;
+
+		ConversionServiceBuilder(TupleBuilder builder, Locale locale, DateFormat dateFormat) {
+			this.builder = builder;
+			this.locale = locale;
+			this.dateFormat = dateFormat;
+		}
+
+		public TupleBuilder setConfigurableConversionService(ConfigurableConversionService formattingConversionService) {
+			Assert.notNull(formattingConversionService);
+
+			if(locale != null) {
+				formattingConversionService.addConverterFactory(new LocaleAwareStringToNumberConverterFactory(NumberFormat
+						.getInstance(locale)));
+			}
+
+			if(dateFormat != null) {
+				formattingConversionService.addConverter(new StringToDateConverter(dateFormat));
+			}
+
+			builder.setConfigurableConversionService(formattingConversionService);
+
+			return builder;
+		}
+	}
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleJsonMarshaller.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2015 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.cloud.stream.tuple;
+
+
+/**
+ * @author David Turanski
+ * 
+ */
+public class TupleJsonMarshaller extends TupleStringMarshaller {
+
+	public TupleJsonMarshaller() {
+		super(new TupleToJsonStringConverter(), new JsonStringToTupleConverter());
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleStringMarshaller.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2015 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.cloud.stream.tuple;
+
+import org.springframework.core.convert.converter.Converter;
+
+/**
+ * @author David Turanski
+ * 
+ */
+public class TupleStringMarshaller {
+
+	private final Converter<Tuple, String> tupleToStringConverter;
+
+	private final Converter<String, Tuple> stringToTupleConverter;
+
+	public TupleStringMarshaller(Converter<Tuple, String> tupleToStringConverter,
+			Converter<String, Tuple> stringToTupleConverter) {
+		this.tupleToStringConverter = tupleToStringConverter;
+		this.stringToTupleConverter = stringToTupleConverter;
+	}
+
+	public Tuple toTuple(String source) {
+		return stringToTupleConverter.convert(source);
+	}
+
+	public String fromTuple(Tuple source) {
+		return tupleToStringConverter.convert(source);
+	}
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleToJsonStringConverter.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2002-2015 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.cloud.stream.tuple;
+
+import org.springframework.core.convert.converter.Converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+/**
+ * Converts a Tuple to JSON representation
+ *
+ * @author David Turanski
+ * @author Gunnar Hillert
+ *
+ */
+public class TupleToJsonStringConverter implements Converter<Tuple, String> {
+
+	private final ObjectMapper mapper = new ObjectMapper();
+
+	@Override
+	public String convert(Tuple source) {
+		ObjectNode root = toObjectNode(source);
+		String json = null;
+		try {
+			json = mapper.writeValueAsString(root);
+		}
+		catch (Exception e) {
+			throw new IllegalArgumentException("Tuple to string conversion failed", e);
+		}
+		return json;
+	}
+
+	private ObjectNode toObjectNode(Tuple source) {
+		ObjectNode root = mapper.createObjectNode();
+//		root.put("id", source.getId().toString());
+//		root.put("timestamp", source.getTimestamp());
+		for (int i = 0; i < source.size(); i++) {
+			Object value = source.getValues().get(i);
+			String name = source.getFieldNames().get(i);
+			if (value != null) {
+				if (value instanceof Tuple) {
+					root.putPOJO(name, toObjectNode((Tuple) value));
+				}
+				else if (!value.getClass().isPrimitive()) {
+					root.putPOJO(name, root.pojoNode(value));
+				}
+				else {
+					root.put(name, value.toString());
+				}
+			}
+		}
+		return root;
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/integration/JsonToTupleTransformer.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple.integration;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.springframework.integration.transformer.AbstractPayloadTransformer;
+import org.springframework.cloud.stream.tuple.Tuple;
+import org.springframework.cloud.stream.tuple.TupleBuilder;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * Converts from a json string into a tuple data structure.
+ * 
+ * @author Mark Fisher
+ */
+public class JsonToTupleTransformer extends AbstractPayloadTransformer<String, Tuple> {
+
+	private final ObjectMapper mapper = new ObjectMapper();
+
+	public JsonToTupleTransformer() {
+		mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
+	}
+
+	@Override
+	public Tuple transformPayload(String json) throws Exception {
+		List<String> names = new ArrayList<String>();
+		List<Object> values = new ArrayList<Object>();
+		JsonNode node = this.mapper.readTree(json);
+		Iterator<String> fieldNames = node.fieldNames();
+		while (fieldNames.hasNext()) {
+			String name = fieldNames.next();
+			JsonNode valueNode = node.get(name);
+			Object value = mapper.treeToValue(valueNode, Object.class);
+			names.add(name);
+			values.add(value);
+		}
+		return TupleBuilder.tuple().ofNamesAndValues(names, values);
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/integration/MapToTupleTransformer.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple.integration;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.integration.transformer.AbstractPayloadTransformer;
+import org.springframework.cloud.stream.tuple.Tuple;
+import org.springframework.cloud.stream.tuple.TupleBuilder;
+
+/**
+ * Converts from a Map to the Tuple data structure.
+ * 
+ * @author Mark Pollack
+ */
+public class MapToTupleTransformer extends AbstractPayloadTransformer<Map<Object, Object>, Tuple> {
+
+	@Override
+	public Tuple transformPayload(Map<Object, Object> map) {
+
+		List<String> newNames = new ArrayList<String>();
+		List<Object> newValues = new ArrayList<Object>();
+		for (Object name : map.keySet()) {
+			newNames.add(name.toString());
+		}
+		for (Object value : map.values()) {
+			newValues.add(value);
+		}
+		return TupleBuilder.tuple().ofNamesAndValues(newNames, newValues);
+
+	}
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/integration/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Contains classes that supports tuple integration such as tuple transformers etc., 
+ */
+
+package org.springframework.cloud.stream.tuple.integration;

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/kryo/DefaultTupleSerializer.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2015 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.cloud.stream.tuple.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+import org.springframework.cloud.stream.tuple.Tuple;
+import org.springframework.cloud.stream.tuple.TupleBuilder;
+
+/**
+ * Deserializes Tuples by writing the field names and then the values as class/object pairs
+ * followed by the tuple Id and timestamp.
+ *
+ * @author David Turanski
+ */
+public class DefaultTupleSerializer extends Serializer<Tuple> {
+	@Override
+	public void write(Kryo kryo, Output output, Tuple tuple) {
+		kryo.writeObject(output, tuple.getFieldNames());
+		for (Object val: tuple.getValues()) {
+			kryo.writeClassAndObject(output, val);
+		}
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public Tuple read(Kryo kryo, Input input, Class<Tuple> type) {
+		List<String> names = kryo.readObject(input, ArrayList.class);
+		List<Object> values = new ArrayList<>(names.size());
+		for (int i = 0; i < names.size(); i++) {
+			Object val = kryo.readClassAndObject(input);
+			values.add(i, val);
+		}
+		return TupleBuilder.tuple().ofNamesAndValues(names, values);
+	}
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/kryo/TupleKryoRegistrar.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2015 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.cloud.stream.tuple.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.integration.codec.kryo.AbstractKryoRegistrar;
+import org.springframework.integration.codec.kryo.KryoRegistrar;
+import org.springframework.cloud.stream.tuple.DefaultTuple;
+
+import com.esotericsoftware.kryo.Registration;
+import com.esotericsoftware.kryo.serializers.CollectionSerializer;
+
+/**
+ * A {@link KryoRegistrar}
+ * used to register a Tuple serializer.
+ * @author David Turanski
+ * @since 1.2
+ */
+public class TupleKryoRegistrar extends AbstractKryoRegistrar {
+
+	private final static int TUPLE_REGISTRATION_ID = 41;
+
+	private final static int ARRAY_LIST_REGISTRATION_ID = 42;
+
+	private final DefaultTupleSerializer defaultTupleSerializer = new DefaultTupleSerializer();
+
+	private final CollectionSerializer collectionSerializer = new CollectionSerializer();
+
+
+	@Override
+	public List<Registration> getRegistrations() {
+		List<Registration> registrations = new ArrayList<>(2);
+		registrations.add(new Registration(DefaultTuple.class, defaultTupleSerializer, TUPLE_REGISTRATION_ID));
+		registrations.add(new Registration(ArrayList.class, collectionSerializer, ARRAY_LIST_REGISTRATION_ID));
+		return registrations;
+	}
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/kryo/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2014 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.
+ */
+
+/**
+ * Contains tuple processor classes.
+ */
+
+package org.springframework.cloud.stream.tuple.kryo;

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Base package for tuple classes.
+ */
+
+package org.springframework.cloud.stream.tuple;

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/processor/TupleProcessor.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple.processor;
+
+import org.springframework.cloud.stream.tuple.Tuple;
+
+/**
+ * Simple type-safe process callback method that returns another Tuple
+ * 
+ * @author Mark Pollack
+ * 
+ */
+public interface TupleProcessor {
+
+	Tuple process(Tuple tuple);
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/processor/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Contains tuple processor classes.
+ */
+
+package org.springframework.cloud.stream.tuple.processor;

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/spel/TuplePropertyAccessor.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2013 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.cloud.stream.tuple.spel;
+
+import org.springframework.expression.AccessException;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.PropertyAccessor;
+import org.springframework.expression.TypedValue;
+import org.springframework.cloud.stream.tuple.Tuple;
+
+/**
+ * A {@link PropertyAccessor} implementation that enables reading of {@link Tuple} values using dot notation within SpEL
+ * expressions. Writing is not supported since {@link Tuple}s are immutable.
+ * 
+ * @author Mark Fisher
+ */
+public class TuplePropertyAccessor implements PropertyAccessor {
+
+	@Override
+	public Class<?>[] getSpecificTargetClasses() {
+		return new Class<?>[] { Tuple.class };
+	}
+
+	@Override
+	public boolean canRead(EvaluationContext context, Object target, String name) throws AccessException {
+		Tuple tuple = (Tuple) target;
+		if (tuple.hasFieldName(name)) {
+			return true;
+		}
+		return maybeIndex(name, tuple) != null;
+	}
+
+	/**
+	 * Return an integer if the String property name can be parsed as an int, or null otherwise.
+	 */
+	private Integer maybeIndex(String name, Tuple tuple) {
+		Integer index = null;
+		try {
+			int i = Integer.parseInt(name);
+			if (i > -1 && tuple.size() > i) {
+				index = i;
+			}
+		}
+		catch (NumberFormatException e) {
+			// not an integer
+		}
+		return index;
+	}
+
+	@Override
+	public TypedValue read(EvaluationContext context, Object target, String name) throws AccessException {
+		Tuple tuple = (Tuple) target;
+		boolean hasKey = false;
+		Object value = null;
+		if (tuple.hasFieldName(name)) {
+			hasKey = true;
+			value = tuple.getValue(name);
+		}
+		else {
+			Integer index = maybeIndex(name, tuple);
+			if (index != null) {
+				hasKey = true;
+				value = tuple.getValue(index);
+			}
+		}
+		if (value == null && !hasKey) {
+			throw new TupleAccessException(name);
+		}
+		return new TypedValue(value);
+	}
+
+	@Override
+	public boolean canWrite(EvaluationContext context, Object target, String name) throws AccessException {
+		return false;
+	}
+
+	@Override
+	public void write(EvaluationContext context, Object target, String name, Object newValue) throws AccessException {
+		throw new UnsupportedOperationException("Tuple is immutable");
+	}
+
+	/**
+	 * Exception thrown from {@code read} in order to reset a cached PropertyAccessor, allowing other accessors to have
+	 * a try.
+	 */
+	@SuppressWarnings("serial")
+	private static class TupleAccessException extends AccessException {
+
+		private final String name;
+
+		public TupleAccessException(String name) {
+			super(null);
+			this.name = name;
+		}
+
+		@Override
+		public String getMessage() {
+			return "Tuple does not contain a value for field name '" + this.name + "'";
+		}
+	}
+
+}

spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/spel/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Contains tuple SpEL accessor classes.
+ */
+
+package org.springframework.cloud.stream.tuple.spel;