Added source adapter for character-based input streams.

spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java

@@ -45,6 +45,8 @@ public abstract class AbstractSourceAdapter<T> implements SourceAdapter, Initial
 
 	private long sendTimeout = -1;
 
+	private volatile boolean initialized = false;
+
 
 	public void setChannel(MessageChannel channel) {
 		Assert.notNull(channel, "'channel' must not be null");
@@ -78,6 +80,11 @@ public abstract class AbstractSourceAdapter<T> implements SourceAdapter, Initial
 			throw new MessagingConfigurationException("'channel' is required");
 		}
 		this.initialize();
+		this.initialized = true;
+	}
+
+	protected boolean isInitialized() {
+		return this.initialized;
 	}
 
 	/**
@@ -87,6 +94,9 @@ public abstract class AbstractSourceAdapter<T> implements SourceAdapter, Initial
 	}
 
 	protected boolean sendToChannel(T object) {
+		if (!this.initialized) {
+			this.afterPropertiesSet();
+		}
 		if (object == null) {
 			if (logger.isDebugEnabled()) {
 				logger.debug("adapter attempted to send a null object");

spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java

@@ -52,6 +52,9 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
 	}
 
 	public void start() {
+		if (!this.isInitialized()) {
+			this.afterPropertiesSet();
+		}
 		this.running = true;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2002-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.adapter.stream;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.adapter.PollableSource;
+
+/**
+ * A pollable source for text-based {@link InputStream InputStreams}.
+ * 
+ * @author Mark Fisher
+ */
+public class CharacterStreamSource implements PollableSource<String> {
+
+	private BufferedReader reader;
+
+
+	public CharacterStreamSource(InputStream stream) {
+		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 = reader.readLine();
+				if (line == null) {
+					return results;
+				}
+				results.add(line);
+			}
+			catch (IOException e) {
+				throw new MessageDeliveryException("IO failure occurred in adapter", e);
+			}
+		}
+		return results;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.adapter.stream;
+
+import java.io.InputStream;
+
+import org.springframework.integration.adapter.PollingSourceAdapter;
+
+/**
+ * A polling source adapter that wraps a {@link CharacterStreamSource}.
+ * 
+ * @author Mark Fisher
+ */
+public class CharacterStreamSourceAdapter extends PollingSourceAdapter<String> {
+
+	public static final CharacterStreamSourceAdapter STDIN_ADAPTER = new CharacterStreamSourceAdapter(System.in);
+
+
+	public CharacterStreamSourceAdapter(InputStream stream) {
+		super(new CharacterStreamSource(stream));
+	}
+
+}