diff --git a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java
index 2d8a5221c3..ecbd1247e1 100644
--- a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java
+++ b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java
@@ -74,7 +74,7 @@ public class ByteStreamWritingMessageConsumerTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
 		poller.setMaxMessagesPerPoll(3);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -90,7 +90,7 @@ public class ByteStreamWritingMessageConsumerTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
 		poller.setMaxMessagesPerPoll(2);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -106,7 +106,7 @@ public class ByteStreamWritingMessageConsumerTests {
 		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
 		poller.setMaxMessagesPerPoll(5);
 		poller.setReceiveTimeout(0);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -122,7 +122,7 @@ public class ByteStreamWritingMessageConsumerTests {
 		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
 		poller.setMaxMessagesPerPoll(2);
 		poller.setReceiveTimeout(0);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -143,7 +143,7 @@ public class ByteStreamWritingMessageConsumerTests {
 		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
 		poller.setMaxMessagesPerPoll(5);
 		poller.setReceiveTimeout(0);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -163,7 +163,7 @@ public class ByteStreamWritingMessageConsumerTests {
 		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
 		poller.setMaxMessagesPerPoll(2);
 		poller.setReceiveTimeout(0);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -183,7 +183,7 @@ public class ByteStreamWritingMessageConsumerTests {
 		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
 		poller.setMaxMessagesPerPoll(2);
 		poller.setReceiveTimeout(0);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
diff --git a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java
index 91b9637d6c..65aef64d99 100644
--- a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java
+++ b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java
@@ -58,7 +58,7 @@ public class CharacterStreamWritingMessageConsumerTests {
 	public void twoStringsAndNoNewLinesByDefault() {
 		StringWriter writer = new StringWriter();
 		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		poller.setMaxMessagesPerPoll(1);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
@@ -73,7 +73,7 @@ public class CharacterStreamWritingMessageConsumerTests {
 		StringWriter writer = new StringWriter();
 		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
 		consumer.setShouldAppendNewLine(true);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		poller.setMaxMessagesPerPoll(1);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
@@ -89,7 +89,7 @@ public class CharacterStreamWritingMessageConsumerTests {
 		StringWriter writer = new StringWriter();
 		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
 		poller.setMaxMessagesPerPoll(2);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
 		poller.run();
@@ -102,7 +102,7 @@ public class CharacterStreamWritingMessageConsumerTests {
 		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
 		poller.setMaxMessagesPerPoll(10);
 		poller.setReceiveTimeout(0);
-		poller.subscribe(consumer);		
+		poller.setConsumer(consumer);		
 		consumer.setShouldAppendNewLine(true);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
@@ -115,7 +115,7 @@ public class CharacterStreamWritingMessageConsumerTests {
 	public void singleNonStringObject() {
 		StringWriter writer = new StringWriter();
 		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		poller.setMaxMessagesPerPoll(1);
 		TestObject testObject = new TestObject("foo");
 		channel.send(new GenericMessage<TestObject>(testObject));
@@ -129,7 +129,7 @@ public class CharacterStreamWritingMessageConsumerTests {
 		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
 		poller.setReceiveTimeout(0);
 		poller.setMaxMessagesPerPoll(2);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);
@@ -145,7 +145,7 @@ public class CharacterStreamWritingMessageConsumerTests {
 		consumer.setShouldAppendNewLine(true);
 		poller.setReceiveTimeout(0);
 		poller.setMaxMessagesPerPoll(2);
-		poller.subscribe(consumer);
+		poller.setConsumer(consumer);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);