From 8fcb10d9be1cb95a57cecd1f3f488f01389cdc98 Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Sat, 12 Jan 2008 00:32:33 +0000
Subject: [PATCH] Added configurable buffer size and synchronization on the
 underlying stream.

---
 .../adapter/stream/CharacterStreamSource.java | 26 +++++++++++++++----
 1 file changed, 21 insertions(+), 5 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
index 575b3d9f72..9d779c1ab6 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
@@ -36,20 +36,36 @@ public class CharacterStreamSource implements PollableSource<String> {
 
 	private BufferedReader reader;
 
+	private Object streamMonitor;
+
 
 	public CharacterStreamSource(InputStream stream) {
-		this.reader = new BufferedReader(new InputStreamReader(stream));
+		this(stream, -1);
 	}
 
+	public CharacterStreamSource(InputStream stream, int bufferSize) {
+		this.streamMonitor = stream;
+		if (bufferSize > 0) {
+			this.reader = new BufferedReader(new InputStreamReader(stream), bufferSize);
+		}
+		else {
+			this.reader = new BufferedReader(new InputStreamReader(stream));
+		}
+	}
+
+
 	public Collection<String> poll(int limit) {
 		List<String> results = new ArrayList<String>();
 		while (results.size() < limit) {
 			try {
-				boolean isReady = reader.ready();
-				if (!isReady) {
-					return results;
+				String line = null;
+				synchronized (this.streamMonitor) {
+					boolean isReady = reader.ready();
+					if (!isReady) {
+						return results;
+					}
+					line = reader.readLine();
 				}
-				String line = reader.readLine();
 				if (line == null) {
 					return results;
 				}