diff --git a/spring-cloud-stream-tuple/pom.xml b/spring-cloud-stream-tuple/pom.xml
index 4058763b4..700af84ad 100644
--- a/spring-cloud-stream-tuple/pom.xml
+++ b/spring-cloud-stream-tuple/pom.xml
@@ -34,14 +34,18 @@
 			<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>-->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-logging</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 </project>
diff --git a/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/JsonStringToTupleConverter.java b/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/JsonStringToTupleConverter.java
index dc54f1dfa..5e6710ab9 100644
--- a/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/JsonStringToTupleConverter.java
+++ b/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/JsonStringToTupleConverter.java
@@ -18,6 +18,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map.Entry;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import org.springframework.core.convert.converter.Converter;
 
 import com.fasterxml.jackson.core.JsonParser;
@@ -62,7 +63,7 @@ public class JsonStringToTupleConverter implements Converter<String, Tuple> {
 					builder.addEntry(name, node.numberValue());
 				}
 				else {
-					builder.addEntry(name, node.asText());
+					builder.addEntry(name, mapper.treeToValue(node, Object.class));
 				}
 			}
 		}
@@ -72,7 +73,7 @@ public class JsonStringToTupleConverter implements Converter<String, Tuple> {
 		return builder.build();
 	}
 
-	private List<Object> nodeToList(JsonNode node) {
+	private List<Object> nodeToList(JsonNode node) throws JsonProcessingException {
 		List<Object> list = new ArrayList<Object>(node.size());
 		for (int i = 0; i < node.size(); i++) {
 			JsonNode item = node.get(i);
@@ -92,7 +93,7 @@ public class JsonStringToTupleConverter implements Converter<String, Tuple> {
 				list.add(item.numberValue());
 			}
 			else {
-				list.add(item.asText());
+				list.add(mapper.treeToValue(item, Object.class));
 			}
 		}
 		return list;
diff --git a/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleToJsonStringConverter.java b/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleToJsonStringConverter.java
index 30d62b1f9..641bb7621 100644
--- a/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleToJsonStringConverter.java
+++ b/spring-cloud-stream-tuple/src/main/java/org/springframework/cloud/stream/tuple/TupleToJsonStringConverter.java
@@ -16,11 +16,15 @@
 
 package org.springframework.cloud.stream.tuple;
 
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.BaseJsonNode;
 import org.springframework.core.convert.converter.Converter;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
+import java.util.List;
+
 /**
  * Converts a Tuple to JSON representation
  *
@@ -51,20 +55,39 @@ public class TupleToJsonStringConverter implements Converter<Tuple, String> {
 			String name = source.getFieldNames().get(i);
 			if (value == null) {
 				root.putNull(name);
-			}
-			else {
-				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());
-				}
+			} else {
+				root.putPOJO(name, toNode(value));
 			}
 		}
 		return root;
 	}
 
+	private ArrayNode toArrayNode(List<?> source) {
+		ArrayNode array = mapper.createArrayNode();
+		for (Object value : source) {
+			if (value != null) {
+				array.add(toNode(value));
+			}
+		}
+		return array;
+	}
+
+	private BaseJsonNode toNode(Object value) {
+		if (value != null) {
+			if (value instanceof Tuple) {
+				return toObjectNode((Tuple) value);
+			}
+			else if (value instanceof List<?>) {
+				return toArrayNode((List<?>) value);
+			}
+			else if (!value.getClass().isPrimitive()) {
+				return mapper.getNodeFactory().pojoNode(value);
+			}
+			else {
+				return mapper.valueToTree(value);
+			}
+		}
+		return null;
+	}
+
 }
diff --git a/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/AbstractTupleMarshallerTests.java b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/AbstractTupleMarshallerTests.java
new file mode 100644
index 000000000..79fc85ec6
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/AbstractTupleMarshallerTests.java
@@ -0,0 +1,119 @@
+/*
+ * 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 com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.core.io.Resource;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.springframework.cloud.stream.tuple.TupleBuilder.tuple;
+
+/**
+ * @author David Turanski
+ * 
+ */
+public abstract class AbstractTupleMarshallerTests {
+
+	private TupleStringMarshaller marshaller;
+
+	private ObjectMapper mapper = new ObjectMapper();
+
+	@Before
+	public void setUp() {
+		marshaller = getMarshaller();
+		mapper.enable(SerializationFeature.INDENT_OUTPUT);
+	}
+
+	@Test
+	public void testMarshallSimpleTuple() {
+
+		DefaultTuple tuple = (DefaultTuple) tuple().of("foo", "bar", "int", 1);
+		String result = marshaller.fromTuple(tuple);
+		Tuple newTuple = marshaller.toTuple(result);
+		assertEquals(2, newTuple.getFieldCount());
+		assertEquals("bar", newTuple.getString(0));
+		assertEquals(1, newTuple.getInt(1));
+	}
+
+	@Test
+	public void testMarshallNestedTuple() {
+		Tuple t1 = tuple().of("foo", "bar", "int", 1);
+		Tuple tuple = tuple().of("t1", t1, "hello", "world");
+
+		String result = marshaller.fromTuple(tuple);
+		Tuple newTuple = marshaller.toTuple(result);
+
+		assertEquals(2, newTuple.getFieldCount());
+		assertEquals("world", newTuple.getString(1));
+
+		Tuple nested = newTuple.getTuple(0);
+		assertEquals(2, nested.getFieldCount());
+		assertEquals("bar", nested.getString(0));
+		assertEquals(1, nested.getInt(1));
+	}
+
+	@Test
+	public void testMarshallTupleWithCollection() {
+		List<?> values = Arrays.asList("a", "b", "c");
+		Tuple tuple = tuple().of("list", values);
+
+		String result = marshaller.fromTuple(tuple);
+		Tuple newTuple = marshaller.toTuple(result);
+		assertTrue(newTuple.getValue(0) instanceof List);
+		List<?> list = (List<?>) newTuple.getValue(0);
+		assertEquals(values.size(), list.size());
+	}
+
+	@Test
+	public void testMarshallTupleWithMap() {
+		Map<String, String> map = new HashMap<String, String>();
+		map.put("k1", "v1");
+		map.put("k2", "v2");
+		Tuple tuple = tuple().of("map", map);
+
+		String result = marshaller.fromTuple(tuple);
+
+		Tuple newTuple = marshaller.toTuple(result);
+		assertNotNull(newTuple.getValue("map"));
+
+		// Map is converted to a tuple
+		assertEquals("v1", newTuple.getTuple("map").getString("k1"));
+		assertEquals("v2", newTuple.getTuple("map").getString("k2"));
+	}
+
+	public String prettyPrintJson(String json) throws IOException {
+		Object jsonObject = mapper.readValue(json, Object.class);
+		return mapper.writeValueAsString(jsonObject);
+	}
+
+	public String readJson(Resource resource) throws IOException {
+		Object jsonObject = mapper.readValue(resource.getInputStream(), Object.class);
+		return mapper.writeValueAsString(jsonObject);
+	}
+
+
+
+	protected abstract TupleStringMarshaller getMarshaller();
+}
diff --git a/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/DefaultTupleTestForBatch.java b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/DefaultTupleTestForBatch.java
new file mode 100644
index 000000000..e71432bc2
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/DefaultTupleTestForBatch.java
@@ -0,0 +1,469 @@
+/*
+ * 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.junit.Before;
+import org.junit.Test;
+import org.springframework.core.convert.ConversionFailedException;
+
+import java.math.BigDecimal;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+import java.util.Locale;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.springframework.cloud.stream.tuple.TupleBuilder.tuple;
+
+/**
+ * This is a port of the FieldSet tests from Spring Batch
+ * 
+ */
+public class DefaultTupleTestForBatch {
+
+	Tuple tuple;
+
+	List<Object> values;
+
+	List<String> names;
+
+
+	@Before
+	public void setUp() throws Exception {
+
+		Object[] tokens = new String[] { "TestString", "true", "C", "10", "-472", "354224", "543", "124.3", "424.3",
+			"1,3245",
+			null, "2007-10-12", "12-10-2007", "" };
+		String[] nameArray = new String[] { "String", "Boolean", "Char", "Byte", "Short", "Integer", "Long", "Float",
+			"Double",
+			"BigDecimal", "Null", "Date", "DatePattern", "BlankInput" };
+
+		names = Arrays.asList(nameArray);
+		values = Arrays.asList(tokens);
+		/*
+		 * values = new ArrayList<Object>(); for (String token : tokens) { values.add(token); }
+		 */
+		tuple = tuple().ofNamesAndValues(names, values);
+		assertEquals(14, tuple.size());
+
+	}
+
+	@Test
+	public void testNames() throws Exception {
+		// MLP - tuples always have names, FieldSet in Spring Batch doesn't require a name.
+		assertThat(tuple.getFieldCount(), is(tuple.getFieldNames().size()));
+	}
+
+	@Test
+	public void testReadString() {
+		assertThat(tuple.getString(0), is("TestString"));
+		assertThat(tuple.getString("String"), is("TestString"));
+	}
+
+	@Test
+	public void testReadChar() {
+		assertThat(tuple.getChar(2), is('C'));
+		assertThat(tuple.getChar("Char"), is('C'));
+	}
+
+	@Test
+	public void testReadBooleanTrue() {
+		assertThat(tuple.getBoolean(1), is(true));
+		assertThat(tuple.getBoolean("Boolean"), is(true));
+	}
+
+	@Test
+	public void testReadByte() {
+		assertTrue(tuple.getByte(3) == 10);
+		assertTrue(tuple.getByte("Byte") == 10);
+	}
+
+	@Test
+	public void testReadShort() {
+		assertTrue(tuple.getShort(4) == -472);
+		assertTrue(tuple.getShort("Short") == -472);
+	}
+
+	@Test
+	public void testReadIntegerAsFloat() {
+		assertEquals(354224, tuple.getFloat(5), .001);
+		assertEquals(354224, tuple.getFloat("Integer"), .001);
+	}
+
+	@Test
+	public void testReadFloat() throws Exception {
+		assertTrue(tuple.getFloat(7) == 124.3F);
+		assertTrue(tuple.getFloat("Float") == 124.3F);
+	}
+
+	@Test
+	public void testReadIntegerAsDouble() throws Exception {
+		assertEquals(354224, tuple.getDouble(5), .001);
+		assertEquals(354224, tuple.getDouble("Integer"), .001);
+	}
+
+	@Test
+	public void testReadDouble() throws Exception {
+		assertTrue(tuple.getDouble(8) == 424.3);
+		assertTrue(tuple.getDouble("Double") == 424.3);
+	}
+
+	@Test
+	public void testReadBigDecimal() throws Exception {
+		BigDecimal bd = new BigDecimal("424.3");
+		assertEquals(bd, tuple.getBigDecimal(8));
+		assertEquals(bd, tuple.getBigDecimal("Double"));
+	}
+
+	@Test
+	public void testReadBigBigDecimal() throws Exception {
+		BigDecimal bd = new BigDecimal("12345678901234567890");
+		Tuple tuple = TupleBuilder.tuple().of("bigd", "12345678901234567890");
+		assertEquals(bd, tuple.getBigDecimal(0));
+	}
+
+	@Test
+	public void testReadBigDecimalWithFormat() throws Exception {
+		Tuple numberFormatTuple = TupleBuilder.tuple()
+				.setFormats(Locale.US, null)
+				.setConfigurableConversionService(new DefaultTupleConversionService())
+				.ofNamesAndValues(tuple.getFieldNames(), tuple.getValues());
+		BigDecimal bd = new BigDecimal("424.3");
+		assertEquals(bd, numberFormatTuple.getBigDecimal(8));
+	}
+
+	@Test
+	public void testReadBigDecimalWithEuroFormat() throws Exception {
+		Tuple numberFormatTuple = TupleBuilder.tuple()
+				.setFormats(Locale.GERMANY, null)
+				.setConfigurableConversionService(new DefaultTupleConversionService())
+				.ofNamesAndValues(tuple.getFieldNames(), tuple.getValues());
+		BigDecimal bd = new BigDecimal("1.3245");
+		assertEquals(bd, numberFormatTuple.getBigDecimal(9));
+	}
+
+	@Test
+	public void testReadNonExistentField() {
+		String s = tuple.getString("something");
+		assertThat(s, nullValue());
+	}
+
+	@Test
+	public void testReadIndexOutOfRange() {
+		try {
+			tuple.getShort(-1);
+			fail("field set returns value even index is out of range!");
+		}
+		catch (IndexOutOfBoundsException e) {
+			assertTrue(true);
+		}
+
+		try {
+			tuple.getShort(99);
+			fail("field set returns value even index is out of range!");
+		}
+		catch (Exception e) {
+			assertTrue(true);
+		}
+	}
+
+	@Test
+	public void testReadBooleanWithTrueValue() {
+		assertTrue(tuple.getBoolean(1, "true"));
+		assertFalse(tuple.getBoolean(1, "incorrect trueValue"));
+
+		assertTrue(tuple.getBoolean("Boolean", "true"));
+		assertFalse(tuple.getBoolean("Boolean", "incorrect trueValue"));
+	}
+
+	@Test
+	public void testReadBooleanFalse() {
+		Tuple t = TupleBuilder.tuple().of("foo", false);
+		assertFalse(t.getBoolean(0));
+	}
+
+	@Test
+	public void testReadCharException() {
+		try {
+			tuple.getChar(1);
+			fail("the value read was not a character, exception expected");
+		}
+		catch (IllegalArgumentException expected) {
+			assertTrue(true);
+		}
+
+		try {
+			tuple.getChar("Boolean");
+			fail("the value read was not a character, exception expected");
+		}
+		catch (IllegalArgumentException expected) {
+			assertTrue(true);
+		}
+	}
+
+
+	@Test
+	public void testReadInt() throws Exception {
+		assertThat(354224, equalTo(tuple.getInt(5)));
+		assertThat(354224, equalTo(tuple.getInt("Integer")));
+	}
+
+	@Test
+	public void testReadIntWithSeparator() {
+		Tuple t = TupleBuilder.tuple().of("foo", "354,224");
+		assertThat(354224, equalTo(t.getInt(0)));
+	}
+
+	@Test
+	public void testReadIntWithSeparatorAndFormat() throws Exception {
+		Tuple t = TupleBuilder.tuple()
+				.setFormats(Locale.GERMAN, null)
+				.setConfigurableConversionService(new DefaultTupleConversionService())
+				.of("foo", "354.224");
+		assertThat(354224, equalTo(t.getInt(0)));
+	}
+
+	@Test
+	public void testReadBlankInt() {
+
+		// Trying to parse a blank field as an integer, but without a default
+		// value should throw a NumberFormatException
+		try {
+			tuple.getInt(13);
+			fail();
+		}
+		catch (ConversionFailedException ex) {
+			// expected
+		}
+
+		try {
+			tuple.getInt("BlankInput");
+			fail();
+		}
+		catch (ConversionFailedException ex) {
+			// expected
+		}
+
+	}
+
+	@Test
+	public void testReadLong() throws Exception {
+		assertThat(543L, equalTo(tuple.getLong(6)));
+		assertThat(543L, equalTo(tuple.getLong("Long")));
+	}
+
+	@Test
+	public void testReadLongWithPadding() throws Exception {
+		Tuple t = TupleBuilder.tuple().of("foo", "000009");
+		assertThat(9L, equalTo(t.getLong(0)));
+	}
+
+	@Test
+	public void testReadIntWithNullValue() {
+		assertThat(5, equalTo(tuple.getInt(10, 5)));
+		assertThat(5, equalTo(tuple.getInt("Null", 5)));
+	}
+
+	@Test
+	public void testReadIntWithDefaultAndNotNull() {
+		assertThat(354224, equalTo(tuple.getInt(5, 5)));
+		assertThat(354224, equalTo(tuple.getInt("Integer", 5)));
+	}
+
+	@Test
+	public void testReadLongWithNullValue() {
+		long defaultValue = 5;
+		int indexOfNull = 10;
+		int indexNotNull = 6;
+		String nameNull = "Null";
+		String nameNotNull = "Long";
+		long longValueAtIndex = 543;
+
+		assertThat(defaultValue, equalTo(tuple.getLong(indexOfNull, defaultValue)));
+		assertThat(longValueAtIndex, equalTo(tuple.getLong(indexNotNull, defaultValue)));
+
+		assertThat(defaultValue, equalTo(tuple.getLong(nameNull, defaultValue)));
+		assertThat(longValueAtIndex, equalTo(tuple.getLong(nameNotNull, defaultValue)));
+	}
+
+	@Test
+	public void testReadBigDecimalInvalid() {
+		int index = 0;
+
+		try {
+			tuple.getBigDecimal(index);
+			fail("field value is not a number, exception expected");
+		}
+		// TODO - in batch this used to be IllegalArgumentException (which is the nested exception type now)
+		catch (ConversionFailedException e) {
+			assertTrue(e.getMessage().indexOf("TestString") > 0);
+		}
+
+	}
+
+	@Test
+	public void testReadBigDecimalByNameInvalid() throws Exception {
+		try {
+			tuple.getBigDecimal("String");
+			fail("field value is not a number, exception expected");
+		}
+		catch (ConversionFailedException e) {
+			assertTrue(e.getMessage().indexOf("TestString") > 0);
+			// TODO - in batch this is part of the message, indicating what the name of the field is...
+			// assertTrue(e.getMessage().indexOf("name: [String]") > 0);
+		}
+	}
+
+
+	@Test
+	public void testReadDate() throws Exception {
+		assertNotNull(tuple.getDate(11));
+		assertNotNull(tuple.getDate("Date"));
+	}
+
+	@Test
+	public void testReadDateWithDefault() {
+		Date date = null;
+		assertEquals(date, tuple.getDateWithPattern(13, "dd-MM-yyyy", date));
+		assertEquals(date, tuple.getDateWithPattern("BlankInput", "dd-MM-yyyy", date));
+	}
+
+	@Test
+	public void testReadDateWithFormat() throws Exception {
+		SimpleDateFormat dateFormat = new SimpleDateFormat("dd/MM/yyyy");
+		Tuple t = TupleBuilder.tuple()
+				.setFormats(null, dateFormat)
+				.setConfigurableConversionService(new DefaultTupleConversionService())
+				.of("foo", "13/01/1999");
+		assertEquals(dateFormat.parse("13/01/1999"), t.getDate(0));
+	}
+
+	@Test
+	public void testReadDateInvalid() throws Exception {
+		try {
+			tuple.getDate(0);
+			fail("field value is not a date, exception expected");
+		}
+		catch (ConversionFailedException e) {
+			assertTrue(e.getMessage().indexOf("TestString") > 0);
+		}
+
+	}
+
+	@Test
+	public void testReadDateInvalidByName() throws Exception {
+
+		try {
+			tuple.getDate("String");
+			fail("field value is not a date, exception expected");
+		}
+		catch (ConversionFailedException e) {
+			assertTrue(e.getMessage().indexOf("TestString") > 0);
+			// TODO - in batch this is part of the message, indicating what the name of the field is...
+			// assertTrue(e.getMessage().indexOf("name: [String]") > 0);
+		}
+
+	}
+
+	@Test
+	public void testReadDateInvalidWithPattern() throws Exception {
+
+		try {
+			tuple.getDateWithPattern(0, "dd-MM-yyyy");
+			fail("field value is not a date, exception expected");
+		}
+		catch (IllegalArgumentException e) {
+			assertTrue(e.getMessage().indexOf("dd-MM-yyyy") > 0);
+		}
+	}
+
+	@Test
+	public void testReadDateWithPatternAndDefault() {
+		Date date = null;
+		assertEquals(date, tuple.getDateWithPattern(13, "dd-MM-yyyy", date));
+		assertEquals(date, tuple.getDateWithPattern("BlankInput", "dd-MM-yyyy", date));
+	}
+
+	@Test
+	public void testStrictReadDateWithPattern() throws Exception {
+
+		Tuple t = tuple().of("foo", "50-2-13");
+		try {
+			t.getDateWithPattern(0, "dd-MM-yyyy");
+			fail("field value is not a valid date for strict parser, exception expected");
+		}
+		catch (IllegalArgumentException e) {
+			String message = e.getMessage();
+			assertTrue("Message did not contain: " + message, message.indexOf("dd-MM-yyyy") > 0);
+		}
+	}
+
+	@Test
+	public void testStrictReadDateWithPatternAndStrangeDate() throws Exception {
+
+		Tuple t = tuple().of("foo", "5550212");
+		try {
+			System.err.println(t.getDateWithPattern(0, "yyyyMMdd"));
+			fail("field value is not a valid date for strict parser, exception expected");
+		}
+		catch (IllegalArgumentException e) {
+			String message = e.getMessage();
+			assertTrue("Message did not contain: " + message, message.indexOf("yyyyMMdd") > 0);
+		}
+	}
+
+	@Test
+	public void testReadDateByNameInvalidWithPattern() throws Exception {
+
+		try {
+			tuple.getDateWithPattern("String", "dd-MM-yyyy");
+			fail("field value is not a date, exception expected");
+		}
+		catch (IllegalArgumentException e) {
+			assertTrue(e.getMessage().indexOf("dd-MM-yyyy") > 0);
+			assertTrue(e.getMessage().indexOf("String") > 0);
+		}
+	}
+
+	@Test
+	public void testPaddedLong() {
+		Tuple t = tuple().of("foo", "00000009");
+		// FieldSet fs = new DefaultFieldSet(new String[] { "00000009" });
+
+		long value = t.getLong(0);
+		assertEquals(value, 9);
+	}
+
+	@Test
+	public void testReadRawString() {
+		String name = "fieldName";
+		String value = " string with trailing whitespace   ";
+		Tuple t = tuple().of(name, value);
+
+		assertEquals(value, t.getRawString(0));
+		assertEquals(value, t.getRawString(name));
+	}
+}
diff --git a/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/DefaultTupleTests.java b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/DefaultTupleTests.java
new file mode 100644
index 000000000..9d9b05dd0
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/DefaultTupleTests.java
@@ -0,0 +1,492 @@
+/*
+ * 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 com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.springframework.core.convert.ConverterNotFoundException;
+
+import java.awt.*;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.springframework.cloud.stream.tuple.TupleBuilder.tuple;
+
+public class DefaultTupleTests {
+
+	@Test(expected = IllegalArgumentException.class)
+	public void nullForNameArray() {
+		List<Object> values = new ArrayList<Object>();
+		values.add("bar");
+		tuple().ofNamesAndValues(null, values);
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void nullForValueArray() {
+		List<String> names = new ArrayList<String>();
+		names.add("foo");
+		tuple().ofNamesAndValues(names, null);
+	}
+
+	@Rule
+	public ExpectedException thrown = ExpectedException.none();
+
+	@Test
+	public void notEqualNumberOfNamesAndValues() {
+		thrown.expect(IllegalArgumentException.class);
+		thrown.expectMessage("Field names must be same length as values: names=[foo, oof], values=[bar]");
+		List<String> names = new ArrayList<String>();
+		names.add("foo");
+		names.add("oof");
+		List<Object> values = new ArrayList<Object>();
+		values.add("bar");
+		tuple().ofNamesAndValues(names, values);
+	}
+
+	@Test
+	public void accessNonExistentEntry() {
+		thrown.expect(IllegalArgumentException.class);
+		thrown.expectMessage("Field name [does-not-exist] does not exist");
+		Tuple tuple = TupleBuilder.tuple().of("foo", "bar");
+		Object v = tuple.getValue("does-not-exist");
+		assertThat(v, nullValue());
+	}
+
+	@Test
+	public void singleEntry() {
+		Tuple tuple = TupleBuilder.tuple().of("foo", "bar");
+		assertThat(tuple.size(), equalTo(1));
+		List<String> names = tuple.getFieldNames();
+		assertThat(names.get(0), equalTo("foo"));
+		assertThat((String) tuple.getValue("foo"), equalTo("bar"));
+		assertThat(tuple.hasFieldName("foo"), equalTo(true));
+		assertThat(tuple.getValue("foo").toString(), equalTo("bar"));
+	}
+
+	@Test
+	public void twoEntries() {
+		Tuple tuple = TupleBuilder.tuple().of("up", "down", "charm", "strange");
+		assertTwoEntries(tuple);
+	}
+
+	@Test
+	public void testTupleAsValue() {
+		Tuple t1 = tuple().of("up", 1, "down", 2);
+		Tuple t2 = tuple().of("1stgen", t1, "charm", 3, "strange", 4);
+		assertThat(t2.size(), equalTo(3));
+		Tuple t3 = t2.getTuple(0);
+		assertThat(t3.size(), equalTo(2));
+		assertThat(t3.getFieldNames().get(0), equalTo("up"));
+		assertThat(t3.getFieldNames().get(1), equalTo("down"));
+		assertThat(t3.getInt("up"), equalTo(1));
+		assertThat(t3.getInt("down"), equalTo(2));
+		Tuple t4 = t2.getTuple("1stgen");
+		assertThat(t3, equalTo(t4));
+	}
+
+	/**
+	 * @param tuple
+	 */
+	private void assertTwoEntries(Tuple tuple) {
+		assertThat(tuple.size(), equalTo(2));
+		assertThat(tuple.getFieldNames().get(0), equalTo("up"));
+		assertThat(tuple.getFieldNames().get(1), equalTo("charm"));
+		assertThat((String) tuple.getValue("up"), equalTo("down"));
+		assertThat((String) tuple.getValue("charm"), equalTo("strange"));
+	}
+
+	@Test
+	public void threeEntries() {
+		Tuple tuple = TupleBuilder.tuple().of("up", 1, "charm", 2, "top", 3);
+		assertThat(tuple.size(), equalTo(3));
+		assertThat(tuple.getFieldNames().get(0), equalTo("up"));
+		assertThat(tuple.getFieldNames().get(1), equalTo("charm"));
+		assertThat(tuple.getFieldNames().get(2), equalTo("top"));
+		// access by name
+		assertThat((Integer) tuple.getValue("up"), equalTo(1));
+		assertThat((Integer) tuple.getValue("charm"), equalTo(2));
+		assertThat((Integer) tuple.getValue("top"), equalTo(3));
+		// access by position
+		assertThat((Integer) tuple.getValue(0), equalTo(1));
+		assertThat((Integer) tuple.getValue(1), equalTo(2));
+		assertThat((Integer) tuple.getValue(2), equalTo(3));
+		// access from separate collection
+		List<Object> values = tuple.getValues();
+		assertThat((Integer) values.get(0), equalTo(1));
+		assertThat((Integer) values.get(1), equalTo(2));
+		assertThat((Integer) values.get(2), equalTo(3));
+	}
+
+	@Test
+	public void fourEntries() {
+		Tuple tuple = TupleBuilder.tuple().of("up", 1, "charm", 2, "top", 3, "e", 4);
+		assertThat(tuple.size(), equalTo(4));
+		assertThat(tuple.getFieldNames().get(0), equalTo("up"));
+		assertThat(tuple.getFieldNames().get(1), equalTo("charm"));
+		assertThat(tuple.getFieldNames().get(2), equalTo("top"));
+		assertThat(tuple.getFieldNames().get(3), equalTo("e"));
+		// access by name
+		assertThat((Integer) tuple.getValue("up"), equalTo(1));
+		assertThat((Integer) tuple.getValue("charm"), equalTo(2));
+		assertThat((Integer) tuple.getValue("top"), equalTo(3));
+		assertThat((Integer) tuple.getValue("e"), equalTo(4));
+		// access by position
+		assertThat((Integer) tuple.getValue(0), equalTo(1));
+		assertThat((Integer) tuple.getValue(1), equalTo(2));
+		assertThat((Integer) tuple.getValue(2), equalTo(3));
+		assertThat((Integer) tuple.getValue(3), equalTo(4));
+	}
+
+	@Test
+	public void getValue() {
+		Tuple tuple = TupleBuilder.tuple().of("up", 1, "charm", 2.0, "top", true);
+		assertThat(tuple.getValue(0, Integer.class), equalTo(1));
+		assertThat(tuple.getValue(0, String.class), equalTo("1"));
+		assertThat(tuple.getValue(1, Double.class), equalTo(2.0D));
+		assertThat(tuple.getValue(2, Boolean.class), equalTo(true));
+		assertThat(tuple.getValue(2, String.class), equalTo("true"));
+
+		assertThat(tuple.getValue("up", Integer.class), equalTo(1));
+		assertThat(tuple.getValue("up", String.class), equalTo("1"));
+		assertThat(tuple.getValue("charm", Double.class), equalTo(2.0D));
+		assertThat(tuple.getValue("top", Boolean.class), equalTo(true));
+		assertThat(tuple.getValue("top", String.class), equalTo("true"));
+	}
+
+	@Test
+	public void testPrimitiveGetters() {
+		Tuple tuple = TupleBuilder.tuple().of("up", "down", "charm", 2.0, "top", true);
+		assertThat(tuple.getBoolean("top"), equalTo(true));
+		assertThat(tuple.getBoolean(2), equalTo(true));
+		assertThat(tuple.getBoolean("up", "down"), equalTo(true));
+	}
+
+	@Test
+	public void testToString() throws JsonProcessingException, IOException {
+		Tuple tuple = TupleBuilder.tuple().put("up", "down").put("charm", "strange").build();
+
+		String tupleString = tuple.toString();
+
+		// valid JSON
+		new ObjectMapper().readTree(tupleString);
+
+		Tuple tupleFromString = TupleBuilder.fromString(tupleString);
+		assertEquals(2, tupleFromString.getFieldCount());
+		assertEquals("down", tupleFromString.getString("up"));
+		assertEquals("strange", tupleFromString.getString("charm"));
+
+	}
+
+	@Test
+	public void testPutApi() {
+		TupleBuilder builder = TupleBuilder.tuple();
+		Tuple tuple = builder.put("up", "down").put("charm", "strange").build();
+		assertTwoEntries(tuple);
+	}
+
+	@Test
+	public void testPutAllApi() {
+		Tuple tuple = TupleBuilder.tuple().put("red", "rot").put("brown", "braun").put("blue", "blau").put("yellow", "gelb")
+				.put("beige", "beige").build();
+		assertThat(tuple.size(), equalTo(5));
+		Tuple tuplePlusOne = TupleBuilder.tuple().putAll(tuple).put("up", 1).build();
+		assertThat(tuplePlusOne.size(), equalTo(6));
+		assertThat(tuplePlusOne.getFieldNames().get(0), equalTo("red"));
+		assertThat(tuplePlusOne.getFieldNames().get(5), equalTo("up"));
+	}
+
+	@Test
+	public void testEqualsAndHashCodeSunnyDay() {
+		Tuple tuple1 = TupleBuilder.tuple().of("up", 1, "charm", 2, "top", 3);
+		Tuple tuple2 = TupleBuilder.tuple().of("up", 1, "charm", 2, "top", 3);
+		assertThat(tuple1, equalTo(tuple2));
+		assertThat(tuple1.hashCode(), equalTo(tuple2.hashCode()));
+		assertThat(tuple1, not(sameInstance(tuple2)));
+	}
+
+	@Test
+	public void testEqualsAndHashFailureCases() {
+		Tuple tuple1 = TupleBuilder.tuple().of("up", 1, "charm", 2, "top", 3);
+		Tuple tuple2 = TupleBuilder.tuple().of("up", 2, "charm", 3, "top", 4);
+		assertThat(tuple1, not(equalTo((tuple2))));
+		assertThat(tuple1.hashCode(), not(equalTo(tuple2.hashCode())));
+
+		tuple1 = TupleBuilder.tuple().of("up", 1, "charm", 2, "top", 3);
+		tuple2 = TupleBuilder.tuple().of("top", 1, "charm", 2, "up", 3);
+		assertThat(tuple1, not(equalTo((tuple2))));
+		assertThat(tuple1.hashCode(), not(equalTo(tuple2.hashCode())));
+	}
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testGetFieldTypes() {
+		Tuple tuple = TupleBuilder.tuple().of("up", 1, "charm", 2, "top", 3);
+		Class[] expectedTypes = new Class[] { Integer.class, Integer.class, Integer.class };
+		assertThat(tuple.getFieldTypes(), equalTo(Arrays.asList(expectedTypes)));
+
+		tuple = TupleBuilder.tuple().of("up", 1, "charm", 2.0f, "top", "bottom");
+		expectedTypes = new Class[] { Integer.class, Float.class, String.class };
+		assertThat(tuple.getFieldTypes(), equalTo(Arrays.asList(expectedTypes)));
+
+		tuple = TupleBuilder.tuple().of("up", 1, "charm", 2.0, "top", true);
+		expectedTypes = new Class[] { Integer.class, Double.class, Boolean.class };
+		assertThat(tuple.getFieldTypes(), equalTo(Arrays.asList(expectedTypes)));
+
+	}
+
+	@Test
+	public void testGetString() {
+		// test conversions of string, int, and float.
+		Tuple tuple = TupleBuilder.tuple().of("up", "down", "charm", 2, "top", 2.0f);
+		assertThat(tuple.getString("up"), equalTo("down"));
+		assertThat(tuple.getString("charm"), equalTo("2"));
+		assertThat(tuple.getString("top"), equalTo("2.0"));
+	}
+
+	@Test
+	public void testGetNullValue() {
+		Tuple tuple = tuple().of("foo", null);
+		// non primitive types will return null
+		assertThat(tuple.getString("foo"), nullValue());
+		assertThat(tuple.getBigDecimal("foo"), nullValue());
+		assertThat(tuple.getDate("foo"), nullValue());
+
+		// primitive types will return default values
+		assertThat(tuple.getChar("foo"), equalTo('\u0000'));
+		assertThat(tuple.getBoolean("foo"), equalTo(false));
+		byte b = 0;
+		assertThat(tuple.getByte("foo"), equalTo(b));
+		short s = 0;
+		assertThat(tuple.getShort("foo"), equalTo(s));
+		assertThat(tuple.getInt("foo"), equalTo(0));
+		assertThat(tuple.getLong("foo"), equalTo(0L));
+		assertThat(tuple.getFloat("foo"), equalTo(0f));
+		assertThat(tuple.getDouble("foo"), equalTo(0d));
+
+	}
+
+	@Test
+	public void testGetStringThatFails() {
+		Tuple tuple = TupleBuilder.tuple().of("up", "down", "charm", 2, "top", 2.0f, "black", Color.black);
+		thrown.expect(ConverterNotFoundException.class);
+		thrown.expectMessage("No converter found capable of converting from type java.awt.Color to type java.lang.String");
+		assertThat(tuple.getString("black"), equalTo("omg"));
+	}
+
+	@Test
+	public void testSelection() {
+		Tuple tuple = tuple().put("red", "rot").put("brown", "braun").put("blue", "blau").put("yellow", "gelb")
+				.put("beige", "beige").build();
+		Tuple selectedTuple = tuple.select("?[key.startsWith('b')]");
+		assertThat(selectedTuple.size(), equalTo(3));
+
+		selectedTuple = tuple.select("^[key.startsWith('b')]");
+		assertThat(selectedTuple.size(), equalTo(1));
+		assertThat(selectedTuple.getFieldNames().get(0), equalTo("brown"));
+		assertThat(selectedTuple.getString(0), equalTo("braun"));
+
+		selectedTuple = tuple.select("?[value.length() < 4]");
+		assertThat(selectedTuple.size(), equalTo(1));
+		assertThat(selectedTuple.getFieldNames().get(0), equalTo("red"));
+		assertThat(selectedTuple.getString(0), equalTo("rot"));
+	}
+
+	@Test
+	public void testReadByteWithDefault() {
+		// with a value
+		byte b = 1;
+		Tuple t = tuple().of("foo", b);
+		byte defaultByte = 2;
+		assertTrue(t.getByte(0, defaultByte) == b);
+		assertTrue(t.getByte("foo", defaultByte) == b);
+		assertTrue(t.getByte("bar", defaultByte) == defaultByte);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getByte(0, defaultByte) == defaultByte);
+		assertTrue(t.getByte("foo", defaultByte) == defaultByte);
+		assertTrue(t.getByte("bar", defaultByte) == defaultByte);
+
+	}
+
+	@Test
+	public void testReadShortWithDefault() {
+		// with a value
+		short s = 1;
+		Tuple t = tuple().of("foo", s);
+		short defaultShort = 2;
+		assertTrue(t.getShort(0, defaultShort) == s);
+		assertTrue(t.getShort("foo", defaultShort) == s);
+		assertTrue(t.getShort("bar", defaultShort) == defaultShort);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getShort(0, defaultShort) == defaultShort);
+		assertTrue(t.getShort("foo", defaultShort) == defaultShort);
+		assertTrue(t.getShort("bar", defaultShort) == defaultShort);
+
+	}
+
+	@Test
+	public void testReadIntWithDefault() {
+		// with a value
+		int i = 1;
+		Tuple t = tuple().of("foo", i);
+		int defaultInt = 2;
+		assertTrue(t.getInt(0, defaultInt) == i);
+		assertTrue(t.getInt("foo", defaultInt) == i);
+		assertTrue(t.getInt("bar", defaultInt) == defaultInt);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getInt(0, defaultInt) == defaultInt);
+		assertTrue(t.getInt("foo", defaultInt) == defaultInt);
+		assertTrue(t.getInt("bar", defaultInt) == defaultInt);
+
+	}
+
+	@Test
+	public void testReadLongWithDefault() {
+		// with a value
+		long l = 1;
+		Tuple t = tuple().of("foo", l);
+		int defaultLong = 2;
+		assertTrue(t.getLong(0, defaultLong) == l);
+		assertTrue(t.getLong("foo", defaultLong) == l);
+		assertTrue(t.getLong("bar", defaultLong) == defaultLong);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getLong(0, defaultLong) == defaultLong);
+		assertTrue(t.getLong("foo", defaultLong) == defaultLong);
+		assertTrue(t.getLong("bar", defaultLong) == defaultLong);
+
+	}
+
+	@Test
+	public void testReadFloatWithDefault() {
+		// with a value
+		float f = 1;
+		Tuple t = tuple().of("foo", f);
+		float defaultFloat = 2.0f;
+		assertTrue(t.getFloat(0, defaultFloat) == f);
+		assertTrue(t.getFloat("foo", defaultFloat) == f);
+		assertTrue(t.getFloat("bar", defaultFloat) == defaultFloat);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getFloat(0, defaultFloat) == defaultFloat);
+		assertTrue(t.getFloat("foo", defaultFloat) == defaultFloat);
+		assertTrue(t.getFloat("bar", defaultFloat) == defaultFloat);
+
+	}
+
+	@Test
+	public void testReadDoubleWithDefault() {
+		// with a value
+		double d = 1;
+		Tuple t = tuple().of("foo", d);
+		double defaultDouble = 2.0d;
+		assertTrue(t.getDouble(0, defaultDouble) == d);
+		assertTrue(t.getDouble("foo", defaultDouble) == d);
+		assertTrue(t.getDouble("bar", defaultDouble) == defaultDouble);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getDouble(0, defaultDouble) == defaultDouble);
+		assertTrue(t.getDouble("foo", defaultDouble) == defaultDouble);
+		assertTrue(t.getDouble("bar", defaultDouble) == defaultDouble);
+
+	}
+
+	@Test
+	public void testReadBigDecimalWithDefault() {
+		// with a value
+		BigDecimal bd = new BigDecimal(1);
+		Tuple t = tuple().of("foo", bd);
+		BigDecimal defaultBigDecimal = new BigDecimal(2);
+		assertTrue(t.getBigDecimal(0, defaultBigDecimal) == bd);
+		assertTrue(t.getBigDecimal("foo", defaultBigDecimal) == bd);
+		assertTrue(t.getBigDecimal("bar", defaultBigDecimal) == defaultBigDecimal);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getBigDecimal(0, defaultBigDecimal) == defaultBigDecimal);
+		assertTrue(t.getBigDecimal("foo", defaultBigDecimal) == defaultBigDecimal);
+		assertTrue(t.getBigDecimal("bar", defaultBigDecimal) == defaultBigDecimal);
+
+	}
+
+	@Test
+	public void testReadDateWithDefault() throws InterruptedException {
+		// with a value
+		Date date = new Date();
+		Tuple t = tuple().of("foo", date);
+		Thread.sleep(1000);
+		Date defaultDate = new Date();
+		assertTrue(t.getDate(0, defaultDate) == date);
+		assertTrue(t.getDate("foo", defaultDate) == date);
+		assertTrue(t.getDate("bar", defaultDate) == defaultDate);
+
+		// with a null value
+		t = tuple().of("foo", null);
+		assertTrue(t.getDate(0, defaultDate) == defaultDate);
+		assertTrue(t.getDate("foo", defaultDate) == defaultDate);
+		assertTrue(t.getDate("bar", defaultDate) == defaultDate);
+
+	}
+
+	@Test
+	public void testReadDateWithPattern() throws ParseException, InterruptedException {
+		Tuple t = tuple().of("foo", "24-12-2013");
+		Date d = t.getDateWithPattern(0, "dd-MM-yyyy");
+		SimpleDateFormat dateFormat = new SimpleDateFormat("dd-MM-yyyy");
+		assertEquals(dateFormat.parse("24-12-2013"), d);
+		Thread.sleep(1000);
+		Date defaultDate = new Date();
+		assertTrue(t.getDateWithPattern("foo", "xyz-abc", defaultDate) == defaultDate);
+		assertTrue(t.getDateWithPattern(0, "xyz-abc", defaultDate) == defaultDate);
+
+	}
+
+	@Test
+	public void testCollectionToTupleConversionFails() {
+		thrown.expect(ConverterNotFoundException.class);
+		Tuple t1 = tuple().of("hello", "world");
+		Tuple t2 = tuple().of("foo", "bar");
+		List<Tuple> list = Arrays.asList(t1, t2);
+		Tuple t = tuple().of("list", list);
+		t.getTuple("list");
+	}
+
+}
diff --git a/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/JsonBytesToTupleConverterTests.java b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/JsonBytesToTupleConverterTests.java
new file mode 100644
index 000000000..09cb024da
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/JsonBytesToTupleConverterTests.java
@@ -0,0 +1,34 @@
+/*
+ * 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 org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author David Turanski
+ * 
+ */
+public class JsonBytesToTupleConverterTests {
+
+	private JsonBytesToTupleConverter converter = new JsonBytesToTupleConverter();
+
+	@Test
+	public void testSimple() {
+		String json = "{\"symbol\":\"VMW\",\"price\":73}";
+		Tuple t = converter.convert(json.getBytes());
+		assertEquals("VMW", t.getValue("symbol"));
+	}
+}
diff --git a/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/SpringToDateConverterTests.java b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/SpringToDateConverterTests.java
new file mode 100644
index 000000000..21064cf94
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/SpringToDateConverterTests.java
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+package org.springframework.cloud.stream.tuple;
+
+import org.junit.Test;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import static org.junit.Assert.assertEquals;
+
+public class SpringToDateConverterTests {
+
+	@Test
+	public void testCtor() throws ParseException {
+		StringToDateConverter converter = new StringToDateConverter();
+		Date d = converter.convert("2013-05-02");
+
+		SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
+		assertEquals(dateFormat.parse("2013-05-02"), d);
+	}
+
+}
diff --git a/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/TupleJsonMarshallerTests.java b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/TupleJsonMarshallerTests.java
new file mode 100644
index 000000000..a74db773c
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/java/org/springframework/cloud/stream/tuple/TupleJsonMarshallerTests.java
@@ -0,0 +1,80 @@
+/*
+ * 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 org.junit.Test;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author David Turanski
+ * 
+ */
+public class TupleJsonMarshallerTests extends AbstractTupleMarshallerTests {
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.xd.tuple.AbstractTupleMarshallerTests#getMarshaller()
+	 */
+	@Override
+	protected TupleJsonMarshaller getMarshaller() {
+		return new TupleJsonMarshaller();
+	}
+
+	@Test
+	public void testComplexJson() throws IOException {
+		Resource jsonFile = new ClassPathResource("/tweet.json");
+		assertTrue(jsonFile.exists());
+
+		String source = readJson(jsonFile);
+		TupleJsonMarshaller marshaller = getMarshaller();
+		Tuple tuple = marshaller.toTuple(source);
+		assertEquals("Gabriel", tuple.getTuple("user").getString("name"));
+		List<?> mentions = (List<?>) tuple.getTuple("entities").getValue("user_mentions");
+		assertEquals(2, mentions.size());
+		assertTrue(mentions.get(0) instanceof Tuple);
+		Tuple t = (Tuple) mentions.get(0);
+		assertEquals("someoneFollowed", t.getString("screen_name"));
+	}
+
+	@Test
+	public void testJsonWithArrays() throws IOException {
+		Resource jsonFile = new ClassPathResource("/jsonWithArrays.json");
+		assertTrue(jsonFile.exists());
+		String json = readJson(jsonFile);
+
+		//Convert to tuple
+		TupleJsonMarshaller marshaller = getMarshaller();
+		Tuple tuple = marshaller.toTuple(json);
+
+		//Validate contents
+		List<Tuple> body = (List<Tuple>) tuple.getValue("body");
+		Tuple t2 = body.get(0).getTuple("har");
+		Tuple t3 = t2.getTuple("log");
+		List<?> pages = (List<?>)t3.getValue("pages");
+		assertEquals(2, pages.size());
+
+		//Convert back to json.
+		String convertedJson = prettyPrintJson(marshaller.fromTuple(tuple));
+		assertEquals(json, convertedJson);
+	}
+
+}
diff --git a/spring-cloud-stream-tuple/src/test/resources/jsonWithArrays.json b/spring-cloud-stream-tuple/src/test/resources/jsonWithArrays.json
new file mode 100644
index 000000000..ea868a54d
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/resources/jsonWithArrays.json
@@ -0,0 +1,53 @@
+{
+  "body" : [ {
+    "dataType" : "har",
+    "har" : {
+      "log" : {
+        "browser" : {
+          "name" : "Google Chrome",
+          "version" : "44.0.2403.155"
+        },
+        "creator" : {
+          "name" : "My extension",
+          "version" : "0.23.6"
+        },
+        "pages" : [ {
+          "_requestTimings" : {
+            "blocked" : -1,
+            "connect" : -1,
+            "dns" : -1,
+            "receive" : 11,
+            "send" : -1,
+            "ssl" : -1,
+            "wait" : null
+          },
+          "_requestUrl" : "https://google.com"
+        }, {
+          "_requestTimings" : {
+            "blocked" : -1,
+            "connect" : -1,
+            "dns" : -1,
+            "receive" : 11,
+            "send" : -1,
+            "ssl" : -1,
+            "wait" : 244
+          },
+          "_requestUrl" : "https://google.com"
+        } ],
+        "version" : "1.2"
+      }
+    },
+    "testId" : 1
+  } ],
+  "bodyType" : "models.MultiMessage",
+  "headers" : {
+    "appInstance" : "localhost/127.0.0.1:8080",
+    "clientIp" : "0:0:0:0:0:0:0:1",
+    "host" : "localhost:8080",
+    "requestId" : "27acf948-33ff-491c-8be7-1beb4b8c95d9",
+    "requestMethod" : "POST",
+    "requestUrl" : "http://localhost:8080/har",
+    "timestamp" : 1445914510549,
+    "userPrincipal" : "235"
+  }
+}
\ No newline at end of file
diff --git a/spring-cloud-stream-tuple/src/test/resources/tweet.json b/spring-cloud-stream-tuple/src/test/resources/tweet.json
new file mode 100644
index 000000000..d936f0157
--- /dev/null
+++ b/spring-cloud-stream-tuple/src/test/resources/tweet.json
@@ -0,0 +1 @@
+{"created_at":"Mon Jul 15 14:00:51 +0000 2013","id":1234567890000000,"id_str":"1234567890000000","text":"@someoneFollowed @ScreenName Hi","source":"web","truncated":false,"in_reply_to_status_id":1234567890,"in_reply_to_status_id_str":"1234567890","in_reply_to_user_id":12345678,"in_reply_to_user_id_str":"12345678","in_reply_to_screen_name":"someoneFollowed","user":{"id":1447674373,"id_str":"1447674373","name":"Gabriel","screen_name":"Django","location":"\u2661 Mile \u2661 Alice \u2661 Lia \u2661 ","url":null,"description":"\u25bd F\u00e3 this is a description \u25b3","protected":false,"followers_count":1569,"friends_count":1525,"listed_count":0,"created_at":"Wed May 22 01:13:27 +0000 2013","favourites_count":45,"utc_offset":-10800,"time_zone":"Brasilia","geo_enabled":false,"verified":false,"statuses_count":7930,"lang":"pt","contributors_enabled":false,"is_translator":false,"profile_background_color":"FFFFFF","profile_background_image_url":"http:\/\/a0.twimg.com\/profile_background_images\/8888000000\/abc.png","profile_background_image_url_https":"https:\/\/si0.twimg.com\/profile_background_images\/888800000013245\/abc.png","profile_background_tile":false,"profile_image_url":"http:\/\/a0.twimg.com\/profile_images\/88000000\/449e706e706ba4574eab44904c97ed0d_normal.png","profile_image_url_https":"https:\/\/si0.twimg.com\/profile_images\/8888000000888888\/0_normal.png","profile_banner_url":"https:\/\/pbs.twimg.com\/profile_banners\/12345678\/12345678","profile_link_color":"5374BA","profile_sidebar_border_color":"FFFFFF","profile_sidebar_fill_color":"E6E6E6","profile_text_color":"2EC7D7","profile_use_background_image":true,"default_profile":false,"default_profile_image":false,"following":null,"follow_request_sent":null,"notifications":null},"geo":null,"coordinates":null,"place":null,"contributors":null,"retweet_count":0,"favorite_count":0,"entities":{"hashtags":[],"symbols":[],"urls":[],"user_mentions":[{"screen_name":"someoneFollowed","name":"Someone Followed","id":12345678,"id_str":"12345678","indices":[0,13]},{"screen_name":"ScreenName","name":"My Name","id":86754321,"id_str":"86754321","indices":[14,27]}]},"favorited":false,"retweeted":false,"filter_level":"medium","lang":"und"}
\ No newline at end of file