renamed modules org.springframework.integration.* -> spring-integration-*

@Ignore'd SimpleTcpNetOutboundGatewayTests#testOutboundClose() to avoid failure; this failure is correlated to the module name change, but hard to understand how it would be caused by it

spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamReadingMessageSource.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2002-2008 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.stream;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.MessageSource;
+
+/**
+ * A pollable source for receiving bytes from an {@link InputStream}.
+ * 
+ * @author Mark Fisher
+ */
+public class ByteStreamReadingMessageSource implements MessageSource<byte[]> {
+
+	private BufferedInputStream stream;
+
+	private Object streamMonitor;
+
+	private int bytesPerMessage = 1024;
+
+	private boolean shouldTruncate = true;
+
+
+	public ByteStreamReadingMessageSource(InputStream stream) {
+		this(stream, -1);
+	}
+
+	public ByteStreamReadingMessageSource(InputStream stream, int bufferSize) {
+		this.streamMonitor = stream;
+		if (stream instanceof BufferedInputStream) {
+			this.stream = (BufferedInputStream) stream;
+		}
+		else if (bufferSize > 0) {
+			this.stream = new BufferedInputStream(stream, bufferSize);
+		}
+		else {
+			this.stream = new BufferedInputStream(stream);
+		}
+	}
+
+
+	public void setBytesPerMessage(int bytesPerMessage) {
+		this.bytesPerMessage = bytesPerMessage;
+	}
+
+	public void setShouldTruncate(boolean shouldTruncate) {
+		this.shouldTruncate = shouldTruncate;
+	}
+
+	public Message<byte[]> receive() {
+		try {
+			byte[] bytes;
+			int bytesRead = 0;
+			synchronized (this.streamMonitor) {
+				if (stream.available() == 0) {
+					return null;
+				}
+				bytes = new byte[bytesPerMessage];
+				bytesRead = stream.read(bytes, 0, bytes.length);
+			}
+			if (bytesRead <= 0) {
+				return null;
+			}
+			if (!this.shouldTruncate) {
+				return new GenericMessage<byte[]>(bytes);
+			}
+			else {
+				byte[] result = new byte[bytesRead];
+				System.arraycopy(bytes, 0, result, 0, result.length);
+				return new GenericMessage<byte[]>(result);
+			}
+		}
+		catch (IOException e) {
+			throw new MessagingException("IO failure occurred in adapter", e);
+		}
+	}
+
+}

spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamWritingMessageHandler.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2002-2008 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.stream;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.message.MessageHandler;
+
+/**
+ * A {@link MessageHandler} that writes a byte array to an {@link OutputStream}.
+ * 
+ * @author Mark Fisher
+ */
+public class ByteStreamWritingMessageHandler implements MessageHandler {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final BufferedOutputStream stream;
+
+
+	public ByteStreamWritingMessageHandler(OutputStream stream) {
+		this(stream, -1);
+	}
+
+	public ByteStreamWritingMessageHandler(OutputStream stream, int bufferSize) {
+		if (bufferSize > 0) {
+			this.stream = new BufferedOutputStream(stream, bufferSize);
+		}
+		else {
+			this.stream = new BufferedOutputStream(stream);
+		}
+	}
+
+
+	public void handleMessage(Message<?> message) {
+		Object payload = message.getPayload();
+		if (payload == null) {
+			if (logger.isWarnEnabled()) {
+				logger.warn(this.getClass().getSimpleName() + " received null object");
+			}
+			return;
+		}
+		try {
+			if (payload instanceof String) {
+				this.stream.write(((String) payload).getBytes());
+			}
+			else if (payload instanceof byte[]){
+				this.stream.write((byte[]) payload);
+			}
+			else {
+				throw new MessagingException(this.getClass().getSimpleName() +
+						" only supports byte array and String-based messages");
+			}
+			this.stream.flush();
+		}
+		catch (IOException e) {
+			throw new MessagingException("IO failure occurred in target", e);
+		}
+	}
+
+}

spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2002-2008 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.stream;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.io.UnsupportedEncodingException;
+
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.message.MessageSource;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.util.Assert;
+
+/**
+ * A pollable source for {@link Reader Readers}.
+ * 
+ * @author Mark Fisher
+ */
+public class CharacterStreamReadingMessageSource implements MessageSource<String> {
+
+	private final BufferedReader reader;
+
+	private final Object monitor;
+
+
+	public CharacterStreamReadingMessageSource(Reader reader) {
+		this(reader, -1);
+	}
+
+	public CharacterStreamReadingMessageSource(Reader reader, int bufferSize) {
+		Assert.notNull(reader, "reader must not be null");
+		this.monitor = reader;
+		if (reader instanceof BufferedReader) {
+			this.reader = (BufferedReader) reader;
+		}
+		else if (bufferSize > 0) {
+			this.reader = new BufferedReader(reader, bufferSize);
+		}
+		else {
+			this.reader = new BufferedReader(reader);
+		}
+	}
+
+
+	public StringMessage receive() {
+		try {
+			synchronized (this.monitor) {
+				if (!this.reader.ready()) {
+					return null;
+				}
+				String line = this.reader.readLine();
+				return (line != null) ? new StringMessage(line) : null;
+			}
+		}
+		catch (IOException e) {
+			throw new MessagingException("IO failure occurred in adapter", e);
+		}
+	}
+
+
+	public static final CharacterStreamReadingMessageSource stdin() {
+		return new CharacterStreamReadingMessageSource(new InputStreamReader(System.in));
+	}
+
+	public static final CharacterStreamReadingMessageSource stdin(String charsetName) {
+		try {
+			return new CharacterStreamReadingMessageSource(new InputStreamReader(System.in, charsetName));
+		}
+		catch (UnsupportedEncodingException e) {
+			throw new IllegalArgumentException("unsupported encoding: " + charsetName, e);
+		}
+	}
+
+}

spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandler.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright 2002-2008 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.stream;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
+import java.io.UnsupportedEncodingException;
+import java.io.Writer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessagingException;
+import org.springframework.integration.message.MessageHandler;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageHandler} that writes characters to a {@link Writer}.
+ * String, character array, and byte array payloads will be written directly,
+ * but for other payload types, the result of the object's {@link #toString()}
+ * method will be written. To append a new-line after each write, set the
+ * {@link #shouldAppendNewLine} flag to 'true'. It is 'false' by default.
+ * 
+ * @author Mark Fisher
+ */
+public class CharacterStreamWritingMessageHandler implements MessageHandler {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final BufferedWriter writer;
+
+	private volatile boolean shouldAppendNewLine = false;
+
+
+	public CharacterStreamWritingMessageHandler(Writer writer) {
+		this(writer, -1);
+	}
+
+	public CharacterStreamWritingMessageHandler(Writer writer, int bufferSize) {
+		Assert.notNull(writer, "writer must not be null");
+		if (writer instanceof BufferedWriter) {
+			this.writer = (BufferedWriter) writer;
+		}
+		else if (bufferSize > 0) {
+			this.writer = new BufferedWriter(writer, bufferSize);
+		}
+		else {
+			this.writer = new BufferedWriter(writer);
+		}
+	}
+
+
+	/**
+	 * Factory method that creates a target for stdout (System.out) with the
+	 * default charset encoding.
+	 */
+	public static CharacterStreamWritingMessageHandler stdout() {
+		return stdout(null);
+	}
+
+	/**
+	 * Factory method that creates a target for stdout (System.out) with the
+	 * specified charset encoding.
+	 */
+	public static CharacterStreamWritingMessageHandler stdout(String charsetName) {
+		return createTargetForStream(System.out, charsetName);
+	}
+
+	/**
+	 * Factory method that creates a target for stderr (System.err) with the
+	 * default charset encoding.
+	 */
+	public static CharacterStreamWritingMessageHandler stderr() {
+		return stderr(null);
+	}
+
+	/**
+	 * Factory method that creates a target for stderr (System.err) with the
+	 * specified charset encoding.
+	 */	
+	public static CharacterStreamWritingMessageHandler stderr(String charsetName) {
+		return createTargetForStream(System.err, charsetName);
+	}
+
+	private static CharacterStreamWritingMessageHandler createTargetForStream(OutputStream stream, String charsetName) {
+		if (charsetName == null) {
+			return new CharacterStreamWritingMessageHandler(new OutputStreamWriter(stream));
+		}
+		try {
+			return new CharacterStreamWritingMessageHandler(new OutputStreamWriter(stream, charsetName));
+		}
+		catch (UnsupportedEncodingException e) {
+			throw new IllegalArgumentException("unsupported encoding: " + charsetName, e);
+		}
+	}
+
+
+	public void setShouldAppendNewLine(boolean shouldAppendNewLine) {
+		this.shouldAppendNewLine = shouldAppendNewLine;
+	}
+
+	public void handleMessage(Message<?> message) {
+		Object payload = message.getPayload();
+		if (payload == null) {
+			if (logger.isWarnEnabled()) {
+				logger.warn("target received null payload");
+			}
+			return;
+		}
+		try {
+			if (payload instanceof String) {
+				writer.write((String) payload);
+			}
+			else if (payload instanceof char[]) {
+				this.writer.write((char[]) payload);
+			}
+			else if (payload instanceof byte[]) {
+				this.writer.write(new String((byte[]) payload));
+			}
+			else if (payload instanceof Exception) {
+				PrintWriter printWriter = new PrintWriter(this.writer, true);
+				((Exception) payload).printStackTrace(printWriter);
+			}
+			else {
+				writer.write(payload.toString());
+			}
+			if (this.shouldAppendNewLine) {
+				writer.newLine();
+			}
+			writer.flush();
+		}
+		catch (IOException e) {
+			throw new MessagingException("IO failure occurred in target", e);
+		}
+	}
+
+}

spring-integration-stream/src/main/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParser.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2002-2008 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.stream.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <stdin-channel-adapter> element.
+ * 
+ * @author Mark Fisher
+ */
+public class ConsoleInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
+
+	@Override
+	protected String parseSource(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
+				"org.springframework.integration.stream.CharacterStreamReadingMessageSource");
+		builder.setFactoryMethod("stdin");
+		String charsetName = element.getAttribute("charset");
+		if (StringUtils.hasText(charsetName)) {
+			builder.addConstructorArgValue(charsetName);
+		}
+		return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
+	}
+
+}

spring-integration-stream/src/main/java/org/springframework/integration/stream/config/ConsoleOutboundChannelAdapterParser.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2002-2008 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.stream.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the "stdout-" and "stderr-channel-adapter" elements.
+ * 
+ * @author Mark Fisher
+ */
+public class ConsoleOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+
+	@Override
+	protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
+				"org.springframework.integration.stream.CharacterStreamWritingMessageHandler");
+		if (element.getLocalName().startsWith("stderr")) {
+			builder.setFactoryMethod("stderr");
+		}
+		else {
+			builder.setFactoryMethod("stdout");
+		}
+		String charsetName = element.getAttribute("charset");
+		if (StringUtils.hasText(charsetName)) {
+			builder.addConstructorArgValue(charsetName);
+		}
+		if ("true".equals(element.getAttribute("append-newline"))) {
+			builder.addPropertyValue("shouldAppendNewLine", Boolean.TRUE);
+		}
+		return builder.getBeanDefinition();
+	}
+
+}

spring-integration-stream/src/main/java/org/springframework/integration/stream/config/StreamNamespaceHandler.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2002-2009 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.stream.config;
+
+import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
+
+/**
+ * @author Mark Fisher
+ */
+public class StreamNamespaceHandler extends AbstractIntegrationNamespaceHandler {
+
+	public void init() {
+		this.registerBeanDefinitionParser("stdin-channel-adapter", new ConsoleInboundChannelAdapterParser());
+		this.registerBeanDefinitionParser("stdout-channel-adapter", new ConsoleOutboundChannelAdapterParser());
+		this.registerBeanDefinitionParser("stderr-channel-adapter", new ConsoleOutboundChannelAdapterParser());
+	}
+
+}

spring-integration-stream/src/main/resources/META-INF/spring.handlers

@@ -0,0 +1 @@
+http\://www.springframework.org/schema/integration/stream=org.springframework.integration.stream.config.StreamNamespaceHandler

spring-integration-stream/src/main/resources/META-INF/spring.schemas

@@ -0,0 +1,3 @@
+http\://www.springframework.org/schema/integration/stream/spring-integration-stream-1.0.xsd=org/springframework/integration/stream/config/spring-integration-stream-1.0.xsd
+http\://www.springframework.org/schema/integration/stream/spring-integration-stream-2.0.xsd=org/springframework/integration/stream/config/spring-integration-stream-2.0.xsd
+http\://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd=org/springframework/integration/stream/config/spring-integration-stream-2.0.xsd

spring-integration-stream/src/main/resources/META-INF/spring.tooling

@@ -0,0 +1,4 @@
+# Tooling related information for the integration stream namespace
+http\://www.springframework.org/schema/integration/stream@name=integration stream Namespace
+http\://www.springframework.org/schema/integration/stream@prefix=int-stream
+http\://www.springframework.org/schema/integration/stream@icon=org/springframework/integration/stream/config/spring-integration-stream.gif

spring-integration-stream/src/main/resources/org/springframework/integration/stream/config/spring-integration-stream-1.0.xsd

@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xsd:schema xmlns="http://www.springframework.org/schema/integration/stream"
+		xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+		xmlns:beans="http://www.springframework.org/schema/beans"
+		xmlns:tool="http://www.springframework.org/schema/tool"
+		xmlns:integration="http://www.springframework.org/schema/integration"
+		targetNamespace="http://www.springframework.org/schema/integration/stream"
+		elementFormDefault="qualified"
+		attributeFormDefault="unqualified">
+
+	<xsd:import namespace="http://www.springframework.org/schema/beans"/>
+	<xsd:import namespace="http://www.springframework.org/schema/tool"/>
+	<xsd:import namespace="http://www.springframework.org/schema/integration"
+			schemaLocation="http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"/>
+
+	<xsd:annotation>
+		<xsd:documentation><![CDATA[
+	Defines the configuration elements for Spring Integration Stream-based Channel Adapters.
+		]]></xsd:documentation>
+	</xsd:annotation>
+
+	<xsd:element name="stdin-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+	Configures a source that reads from stdin (System.in).
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:sequence>
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+			</xsd:sequence>
+			<xsd:attribute name="id" type="xsd:string"/>
+			<xsd:attribute name="channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="charset" type="xsd:string"/>
+			<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="stdout-channel-adapter" type="consoleOutboundChannelAdapterType"/>
+
+	<xsd:element name="stderr-channel-adapter" type="consoleOutboundChannelAdapterType"/>
+
+	<xsd:complexType name="consoleOutboundChannelAdapterType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Configures an outbound Channel Adapter that writes to stdout (System.out)
+	or to stderr (System.err) depending on the element name.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="id" type="xsd:string"/>
+		<xsd:attribute name="channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="charset" type="xsd:string"/>
+		<xsd:attribute name="append-newline" type="xsd:string" default="false"/>
+		<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
+	</xsd:complexType>
+
+</xsd:schema>

spring-integration-stream/src/main/resources/org/springframework/integration/stream/config/spring-integration-stream-2.0.xsd

@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xsd:schema xmlns="http://www.springframework.org/schema/integration/stream"
+		xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+		xmlns:beans="http://www.springframework.org/schema/beans"
+		xmlns:tool="http://www.springframework.org/schema/tool"
+		xmlns:integration="http://www.springframework.org/schema/integration"
+		targetNamespace="http://www.springframework.org/schema/integration/stream"
+		elementFormDefault="qualified"
+		attributeFormDefault="unqualified">
+
+	<xsd:import namespace="http://www.springframework.org/schema/beans"/>
+	<xsd:import namespace="http://www.springframework.org/schema/tool"/>
+	<xsd:import namespace="http://www.springframework.org/schema/integration"
+			schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd"/>
+
+	<xsd:annotation>
+		<xsd:documentation><![CDATA[
+	Defines the configuration elements for Spring Integration Stream-based Channel Adapters.
+		]]></xsd:documentation>
+	</xsd:annotation>
+
+	<xsd:element name="stdin-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+	Configures a source that reads from stdin (System.in).
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:sequence>
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+			</xsd:sequence>
+			<xsd:attribute name="id" type="xsd:string"/>
+			<xsd:attribute name="channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="charset" type="xsd:string"/>
+			<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="stdout-channel-adapter" type="consoleOutboundChannelAdapterType"/>
+
+	<xsd:element name="stderr-channel-adapter" type="consoleOutboundChannelAdapterType"/>
+
+	<xsd:complexType name="consoleOutboundChannelAdapterType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Configures an outbound Channel Adapter that writes to stdout (System.out)
+	or to stderr (System.err) depending on the element name.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="id" type="xsd:string"/>
+		<xsd:attribute name="channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="charset" type="xsd:string"/>
+		<xsd:attribute name="append-newline" type="xsd:string" default="false"/>
+		<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
+	</xsd:complexType>
+
+</xsd:schema>

spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamSourceTests.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2002-2008 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.stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.io.ByteArrayInputStream;
+
+import org.junit.Test;
+
+import org.springframework.integration.core.Message;
+
+/**
+ * @author Mark Fisher
+ */
+public class ByteStreamSourceTests {
+
+	@Test
+	public void testEndOfStream() {
+		byte[] bytes = new byte[] {1,2,3};
+		ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
+		ByteStreamReadingMessageSource source = new ByteStreamReadingMessageSource(stream);
+		Message<?> message1 = source.receive();
+		byte[] payload = (byte[]) message1.getPayload();
+		assertEquals(3, payload.length);
+		assertEquals(1, payload[0]);
+		assertEquals(2, payload[1]);
+		assertEquals(3, payload[2]);
+		Message<?> message2 = source.receive();
+		assertNull(message2);
+	}
+
+	@Test
+	public void testByteArrayIsTruncated() {
+		byte[] bytes = new byte[] {0,1,2,3,4,5};
+		ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
+		ByteStreamReadingMessageSource source = new ByteStreamReadingMessageSource(stream);
+		source.setBytesPerMessage(4);
+		Message<?> message1 = source.receive();
+		assertEquals(4, ((byte[]) message1.getPayload()).length);
+		Message<?> message2 = source.receive();
+		assertEquals(2, ((byte[]) message2.getPayload()).length);
+		Message<?> message3 = source.receive();
+		assertNull(message3);
+	}
+
+	@Test
+	public void testByteArrayIsNotTruncated() {
+		byte[] bytes = new byte[] {0,1,2,3,4,5};
+		ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
+		ByteStreamReadingMessageSource source = new ByteStreamReadingMessageSource(stream);
+		source.setBytesPerMessage(4);
+		source.setShouldTruncate(false);
+		Message<?> message1 = source.receive();
+		assertEquals(4, ((byte[]) message1.getPayload()).length);
+		Message<?> message2 = source.receive();
+		assertEquals(4, ((byte[]) message2.getPayload()).length);
+		assertEquals(4, ((byte[]) message2.getPayload())[0]);
+		assertEquals(5, ((byte[]) message2.getPayload())[1]);
+		assertEquals(0, ((byte[]) message2.getPayload())[2]);
+		assertEquals(0, ((byte[]) message2.getPayload())[3]);
+		Message<?> message3 = source.receive();
+		assertNull(message3);
+	}
+
+}

spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java

@@ -0,0 +1,264 @@
+/*
+ * Copyright 2002-2009 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.stream;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Date;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.PollingConsumer;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.scheduling.Trigger;
+import org.springframework.scheduling.TriggerContext;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/**
+ * @author Mark Fisher
+ */
+public class ByteStreamWritingMessageHandlerTests {
+
+	private ByteArrayOutputStream stream;
+
+	private ByteStreamWritingMessageHandler handler;
+
+	private QueueChannel channel;
+
+	private PollingConsumer endpoint;
+
+	private TestTrigger trigger = new TestTrigger();
+
+	private ThreadPoolTaskScheduler scheduler; 
+
+
+	@Before
+	public void initialize() {
+		stream = new ByteArrayOutputStream();
+		handler = new ByteStreamWritingMessageHandler(stream);
+		this.channel = new QueueChannel(10);
+		this.endpoint = new PollingConsumer(channel, handler);
+		scheduler = new ThreadPoolTaskScheduler();
+		this.endpoint.setTaskScheduler(scheduler);
+		scheduler.afterPropertiesSet();
+		trigger.reset();
+		endpoint.setTrigger(trigger);
+	}
+
+	@After
+	public void stop() throws Exception {
+		scheduler.destroy();
+	}
+
+
+	@Test
+	public void singleByteArray() {
+		handler.handleMessage(new GenericMessage<byte[]>(new byte[] {1,2,3}));
+		byte[] result = stream.toByteArray();
+		assertEquals(3, result.length);
+		assertEquals(1, result[0]);
+		assertEquals(2, result[1]);
+		assertEquals(3, result[2]);
+	}
+
+	@Test
+	public void singleString() {
+		handler.handleMessage(new StringMessage("foo"));
+		byte[] result = stream.toByteArray();
+		assertEquals(3, result.length);
+		assertEquals("foo", new String(result));
+	}
+
+	@Test
+	public void maxMessagesPerTaskSameAsMessageCount() {
+		endpoint.setMaxMessagesPerPoll(3);
+		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);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result = stream.toByteArray();
+		assertEquals(9, result.length);
+		assertEquals(1, result[0]);
+		assertEquals(9, result[8]);
+	}
+
+	@Test
+	public void maxMessagesPerTaskLessThanMessageCount() {
+		endpoint.setMaxMessagesPerPoll(2);
+		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);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result = stream.toByteArray();
+		assertEquals(6, result.length);
+		assertEquals(1, result[0]);
+	}
+
+	@Test
+	public void maxMessagesPerTaskExceedsMessageCount() {
+		endpoint.setMaxMessagesPerPoll(5);
+		endpoint.setReceiveTimeout(0);
+		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);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result = stream.toByteArray();
+		assertEquals(9, result.length);
+		assertEquals(1, result[0]);
+	}
+
+	@Test
+	public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() {
+		endpoint.setMaxMessagesPerPoll(2);
+		endpoint.setReceiveTimeout(0);
+		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);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result1 = stream.toByteArray();
+		assertEquals(6, result1.length);
+		assertEquals(1, result1[0]);
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result2 = stream.toByteArray();
+		assertEquals(9, result2.length);
+		assertEquals(1, result2[0]);
+		assertEquals(7, result2[6]);
+	}
+
+	@Test
+	public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() {
+		endpoint.setMaxMessagesPerPoll(5);
+		endpoint.setReceiveTimeout(0);
+		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);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result1 = stream.toByteArray();
+		assertEquals(9, result1.length);
+		assertEquals(1, result1[0]);
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result2 = stream.toByteArray();
+		assertEquals(9, result2.length);	
+		assertEquals(1, result2[0]);
+	}
+
+	@Test
+	public void testStreamResetBetweenDispatches() {
+		endpoint.setMaxMessagesPerPoll(2);
+		endpoint.setReceiveTimeout(0);
+		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);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result1 = stream.toByteArray();
+		assertEquals(6, result1.length);
+		stream.reset();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result2 = stream.toByteArray();
+		assertEquals(3, result2.length);
+		assertEquals(7, result2[0]);
+	}
+
+	@Test
+	public void testStreamWriteBetweenDispatches() throws IOException {
+		endpoint.setMaxMessagesPerPoll(2);
+		endpoint.setReceiveTimeout(0);
+		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);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result1 = stream.toByteArray();
+		assertEquals(6, result1.length);
+		stream.write(new byte[] {123});
+		stream.flush();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		byte[] result2 = stream.toByteArray();
+		assertEquals(10, result2.length);
+		assertEquals(1, result2[0]);
+		assertEquals(123, result2[6]);
+		assertEquals(7, result2[7]);
+	}
+
+
+	private static class TestTrigger implements Trigger {
+
+		private final AtomicBoolean hasRun = new AtomicBoolean();
+
+		private volatile CountDownLatch latch = new CountDownLatch(1);
+
+		public Date nextExecutionTime(TriggerContext triggerContext) {
+			if (!hasRun.getAndSet(true)) {
+				return new Date();
+			}
+			this.latch.countDown();
+			return null;
+		}
+
+		public void reset() {
+			this.latch = new CountDownLatch(1);
+			this.hasRun.set(false);
+		}
+
+		public void await() {
+			try {
+				this.latch.await(3000, TimeUnit.MILLISECONDS);
+				if (latch.getCount() != 0) {
+					throw new RuntimeException("test timeout");
+				}
+			}
+			catch (InterruptedException e) {
+				throw new RuntimeException("test latch.await() interrupted");
+			}
+		}
+	}
+
+}

spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamSourceTests.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-2008 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.stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.io.StringReader;
+
+import org.junit.Test;
+
+import org.springframework.integration.core.Message;
+
+/**
+ * @author Mark Fisher
+ */
+public class CharacterStreamSourceTests {
+
+	@Test
+	public void testEndOfStream() {
+		StringReader reader = new StringReader("test");
+		CharacterStreamReadingMessageSource source = new CharacterStreamReadingMessageSource(reader);
+		Message<?> message1 = source.receive();
+		assertEquals("test", message1.getPayload());
+		Message<?> message2 = source.receive();
+		assertNull(message2);
+	}
+
+}

spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java

@@ -0,0 +1,230 @@
+/*
+ * Copyright 2002-2009 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.stream;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.StringWriter;
+import java.util.Date;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.PollingConsumer;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.scheduling.Trigger;
+import org.springframework.scheduling.TriggerContext;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/**
+ * @author Mark Fisher
+ */
+public class CharacterStreamWritingMessageHandlerTests {
+
+	private StringWriter writer;
+
+	private CharacterStreamWritingMessageHandler handler;
+
+	private QueueChannel channel;
+
+	private PollingConsumer endpoint;
+
+	private TestTrigger trigger = new TestTrigger();
+
+	private ThreadPoolTaskScheduler scheduler;
+
+
+	@Before
+	public void initialize() {
+		writer = new StringWriter();
+		handler = new CharacterStreamWritingMessageHandler(writer);
+		this.channel = new QueueChannel(10);
+		trigger.reset();
+		this.endpoint = new PollingConsumer(channel, handler);
+		scheduler = new ThreadPoolTaskScheduler();
+		this.endpoint.setTaskScheduler(scheduler);
+		scheduler.afterPropertiesSet();
+		trigger.reset();
+		endpoint.setTrigger(trigger);
+	}
+
+	@After
+	public void stop() throws Exception {
+		scheduler.destroy();
+	}
+
+
+	@Test
+	public void singleString() {
+		handler.handleMessage(new StringMessage("foo"));
+		assertEquals("foo", writer.toString());
+	}
+
+	@Test
+	public void twoStringsAndNoNewLinesByDefault() {
+		endpoint.setMaxMessagesPerPoll(1);
+		channel.send(new StringMessage("foo"), 0);
+		channel.send(new StringMessage("bar"), 0);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		assertEquals("foo", writer.toString());
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		assertEquals("foobar", writer.toString());
+	}
+
+	@Test
+	public void twoStringsWithNewLines() {
+		handler.setShouldAppendNewLine(true);
+		endpoint.setMaxMessagesPerPoll(1);
+		channel.send(new StringMessage("foo"), 0);
+		channel.send(new StringMessage("bar"), 0);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		String newLine = System.getProperty("line.separator");
+		assertEquals("foo" + newLine, writer.toString());
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
+	}
+
+	@Test
+	public void maxMessagesPerTaskSameAsMessageCount() {
+		endpoint.setMaxMessagesPerPoll(2);
+		channel.send(new StringMessage("foo"), 0);
+		channel.send(new StringMessage("bar"), 0);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		assertEquals("foobar", writer.toString());
+	}
+
+	@Test
+	public void maxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
+		endpoint.setMaxMessagesPerPoll(10);
+		endpoint.setReceiveTimeout(0);
+		handler.setShouldAppendNewLine(true);
+		channel.send(new StringMessage("foo"), 0);
+		channel.send(new StringMessage("bar"), 0);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		String newLine = System.getProperty("line.separator");
+		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
+	}
+
+	@Test
+	public void singleNonStringObject() {
+		endpoint.setMaxMessagesPerPoll(1);
+		TestObject testObject = new TestObject("foo");
+		channel.send(new GenericMessage<TestObject>(testObject));
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		assertEquals("foo", writer.toString());
+	}
+
+	@Test
+	public void twoNonStringObjectWithOutNewLines() {
+		endpoint.setReceiveTimeout(0);
+		endpoint.setMaxMessagesPerPoll(2);
+		TestObject testObject1 = new TestObject("foo");
+		TestObject testObject2 = new TestObject("bar");
+		channel.send(new GenericMessage<TestObject>(testObject1), 0);
+		channel.send(new GenericMessage<TestObject>(testObject2), 0);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		assertEquals("foobar", writer.toString());
+	}
+
+	@Test
+	public void twoNonStringObjectWithNewLines() {
+		handler.setShouldAppendNewLine(true);
+		endpoint.setReceiveTimeout(0);
+		endpoint.setMaxMessagesPerPoll(2);
+		TestObject testObject1 = new TestObject("foo");
+		TestObject testObject2 = new TestObject("bar");
+		channel.send(new GenericMessage<TestObject>(testObject1), 0);
+		channel.send(new GenericMessage<TestObject>(testObject2), 0);
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
+		String newLine = System.getProperty("line.separator");
+		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
+	}
+
+
+	private static class TestObject {
+
+		private String text;
+
+		TestObject(String text) {
+			this.text = text;
+		}
+
+		public String toString() {
+			return this.text;
+		}
+	}
+
+
+	private static class TestTrigger implements Trigger {
+
+		private final AtomicBoolean hasRun = new AtomicBoolean();
+
+		private volatile CountDownLatch latch = new CountDownLatch(1);
+
+		public Date nextExecutionTime(TriggerContext triggerContext) {
+			if (!hasRun.getAndSet(true)) {
+				return new Date();
+			}
+			this.latch.countDown();
+			return null;
+		}
+
+		public void reset() {
+			this.latch = new CountDownLatch(1);
+			this.hasRun.set(false);
+		}
+
+		public void await() {
+			try {
+				this.latch.await(1000, TimeUnit.MILLISECONDS);
+				if (latch.getCount() != 0) {
+					throw new RuntimeException("test timeout");
+				}
+			}
+			catch (InterruptedException e) {
+				throw new RuntimeException("test latch.await() interrupted");
+			}
+		}
+	}
+
+}

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParserTests.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2002-2008 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.stream.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.io.BufferedReader;
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.io.UnsupportedEncodingException;
+import java.nio.charset.Charset;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.message.MessageSource;
+
+/**
+ * @author Mark Fisher
+ */
+public class ConsoleInboundChannelAdapterParserTests {
+
+	@Before
+	public void writeTestInput() {
+		ByteArrayInputStream stream = new ByteArrayInputStream("foo".getBytes());
+		System.setIn(stream);
+	}
+
+	@Test
+	public void adapterWithDefaultCharset() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"consoleInboundChannelAdapterParserTests.xml", ConsoleInboundChannelAdapterParserTests.class);
+		SourcePollingChannelAdapter adapter =
+				(SourcePollingChannelAdapter) context.getBean("adapterWithDefaultCharset.adapter");
+		MessageSource<?> source = (MessageSource<?>) new DirectFieldAccessor(adapter).getPropertyValue("source");
+		DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(source);
+		Reader bufferedReader = (Reader) sourceAccessor.getPropertyValue("reader");
+		assertEquals(BufferedReader.class, bufferedReader.getClass());
+		DirectFieldAccessor bufferedReaderAccessor = new DirectFieldAccessor(bufferedReader);
+		Reader reader = (Reader) bufferedReaderAccessor.getPropertyValue("in");
+		assertEquals(InputStreamReader.class, reader.getClass());
+		Charset readerCharset = Charset.forName(((InputStreamReader) reader).getEncoding());
+		assertEquals(Charset.defaultCharset(), readerCharset);
+		Message<?> message = source.receive();
+		assertNotNull(message);
+		assertEquals("foo", message.getPayload());
+	}
+
+	@Test
+	public void adapterWithProvidedCharset() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"consoleInboundChannelAdapterParserTests.xml", ConsoleInboundChannelAdapterParserTests.class);
+		SourcePollingChannelAdapter adapter =
+				(SourcePollingChannelAdapter) context.getBean("adapterWithProvidedCharset.adapter");
+		MessageSource<?> source = (MessageSource<?>) new DirectFieldAccessor(adapter).getPropertyValue("source");
+		DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(source);
+		Reader bufferedReader = (Reader) sourceAccessor.getPropertyValue("reader");
+		assertEquals(BufferedReader.class, bufferedReader.getClass());
+		DirectFieldAccessor bufferedReaderAccessor = new DirectFieldAccessor(bufferedReader);
+		Reader reader = (Reader) bufferedReaderAccessor.getPropertyValue("in");
+		assertEquals(InputStreamReader.class, reader.getClass());
+		Charset readerCharset = Charset.forName(((InputStreamReader) reader).getEncoding());
+		assertEquals(Charset.forName("UTF-8"), readerCharset);
+		Message<?> message = source.receive();
+		assertNotNull(message);
+		assertEquals("foo", message.getPayload());
+	}
+
+	@Test
+	public void testConsoleSourceWithInvalidCharset() {
+		BeanCreationException beanCreationException = null;
+		try {
+			new ClassPathXmlApplicationContext(
+					"invalidConsoleInboundChannelAdapterParserTests.xml", ConsoleInboundChannelAdapterParserTests.class);
+		}
+		catch (BeanCreationException e) {
+			beanCreationException = e;
+		}
+		Throwable rootCause = beanCreationException.getRootCause();
+		assertEquals(UnsupportedEncodingException.class, rootCause.getClass());
+	}
+
+}

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/ConsoleOutboundChannelAdapterParserTests.java

@@ -0,0 +1,156 @@
+/*
+ * Copyright 2002-2008 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.stream.config;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.ByteArrayOutputStream;
+import java.io.OutputStreamWriter;
+import java.io.PrintStream;
+import java.io.UnsupportedEncodingException;
+import java.io.Writer;
+import java.nio.charset.Charset;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.stream.CharacterStreamWritingMessageHandler;
+
+/**
+ * @author Mark Fisher
+ */
+public class ConsoleOutboundChannelAdapterParserTests {
+
+	private final ByteArrayOutputStream err = new ByteArrayOutputStream();
+
+	private final ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+
+	@Before
+	public void setupStreams() {
+		System.setErr(new PrintStream(this.err));
+		System.setOut(new PrintStream(this.out));
+	}
+
+	private void resetStreams() {
+		this.err.reset();
+		this.out.reset();
+	}
+
+	@Test
+	public void stdoutAdapterWithDefaultCharset() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"consoleOutboundChannelAdapterParserTests.xml", ConsoleOutboundChannelAdapterParserTests.class);
+		Object adapter = context.getBean("stdoutAdapterWithDefaultCharset");
+		CharacterStreamWritingMessageHandler handler = (CharacterStreamWritingMessageHandler)
+				new DirectFieldAccessor(adapter).getPropertyValue("handler");
+		DirectFieldAccessor accessor = new DirectFieldAccessor(handler);
+		Writer bufferedWriter = (Writer) accessor.getPropertyValue("writer");
+		assertEquals(BufferedWriter.class, bufferedWriter.getClass());
+		DirectFieldAccessor bufferedWriterAccessor = new DirectFieldAccessor(bufferedWriter);
+		Writer writer = (Writer) bufferedWriterAccessor.getPropertyValue("out");
+		assertEquals(OutputStreamWriter.class, writer.getClass());
+		Charset writerCharset = Charset.forName(((OutputStreamWriter) writer).getEncoding());
+		assertEquals(Charset.defaultCharset(), writerCharset);
+		this.resetStreams();
+		handler.handleMessage(new StringMessage("foo"));
+		assertEquals("foo", out.toString());
+		assertEquals("", err.toString());
+	}
+
+	@Test
+	public void stdoutAdapterWithProvidedCharset() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"consoleOutboundChannelAdapterParserTests.xml", ConsoleOutboundChannelAdapterParserTests.class);
+		Object adapter = context.getBean("stdoutAdapterWithProvidedCharset");
+		CharacterStreamWritingMessageHandler handler = (CharacterStreamWritingMessageHandler)
+				new DirectFieldAccessor(adapter).getPropertyValue("handler");
+		DirectFieldAccessor accessor = new DirectFieldAccessor(handler);
+		Writer bufferedWriter = (Writer) accessor.getPropertyValue("writer");
+		assertEquals(BufferedWriter.class, bufferedWriter.getClass());
+		DirectFieldAccessor bufferedWriterAccessor = new DirectFieldAccessor(bufferedWriter);
+		Writer writer = (Writer) bufferedWriterAccessor.getPropertyValue("out");
+		assertEquals(OutputStreamWriter.class, writer.getClass());
+		Charset writerCharset = Charset.forName(((OutputStreamWriter) writer).getEncoding());
+		assertEquals(Charset.forName("UTF-8"), writerCharset);
+		this.resetStreams();
+		handler.handleMessage(new StringMessage("bar"));
+		assertEquals("bar", out.toString());
+		assertEquals("", err.toString());
+	}
+
+	@Test
+	public void stdoutAdapterWithInvalidCharset() {
+		BeanCreationException beanCreationException = null;
+		try {
+			new ClassPathXmlApplicationContext(
+					"invalidConsoleOutboundChannelAdapterParserTests.xml", ConsoleOutboundChannelAdapterParserTests.class);
+		}
+		catch (BeanCreationException e) {
+			beanCreationException = e;
+		}
+		Throwable rootCause = beanCreationException.getRootCause();
+		assertEquals(UnsupportedEncodingException.class, rootCause.getClass());
+	}
+
+	@Test
+	public void stderrAdapter() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"consoleOutboundChannelAdapterParserTests.xml", ConsoleOutboundChannelAdapterParserTests.class);
+		Object adapter = context.getBean("stderrAdapter");
+		CharacterStreamWritingMessageHandler handler = (CharacterStreamWritingMessageHandler)
+				new DirectFieldAccessor(adapter).getPropertyValue("handler");
+		DirectFieldAccessor accessor = new DirectFieldAccessor(handler);
+		Writer bufferedWriter = (Writer) accessor.getPropertyValue("writer");
+		assertEquals(BufferedWriter.class, bufferedWriter.getClass());
+		DirectFieldAccessor bufferedWriterAccessor = new DirectFieldAccessor(bufferedWriter);
+		Writer writer = (Writer) bufferedWriterAccessor.getPropertyValue("out");
+		assertEquals(OutputStreamWriter.class, writer.getClass());
+		Charset writerCharset = Charset.forName(((OutputStreamWriter) writer).getEncoding());
+		assertEquals(Charset.defaultCharset(), writerCharset);
+		this.resetStreams();
+		handler.handleMessage(new StringMessage("bad"));
+		assertEquals("", out.toString());
+		assertEquals("bad", err.toString());
+	}
+
+    @Test
+	public void stdoutAdatperWithAppendNewLine() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"consoleOutboundChannelAdapterParserTests.xml", ConsoleOutboundChannelAdapterParserTests.class);
+		Object adapter = context.getBean("newlineAdapter");
+		CharacterStreamWritingMessageHandler handler = (CharacterStreamWritingMessageHandler)
+				new DirectFieldAccessor(adapter).getPropertyValue("handler");
+		DirectFieldAccessor accessor = new DirectFieldAccessor(handler);
+		Writer bufferedWriter = (Writer) accessor.getPropertyValue("writer");
+		assertEquals(BufferedWriter.class, bufferedWriter.getClass());
+		DirectFieldAccessor bufferedWriterAccessor = new DirectFieldAccessor(bufferedWriter);
+		Writer writer = (Writer) bufferedWriterAccessor.getPropertyValue("out");
+		assertEquals(OutputStreamWriter.class, writer.getClass());
+		Charset writerCharset = Charset.forName(((OutputStreamWriter) writer).getEncoding());
+		assertEquals(Charset.defaultCharset(), writerCharset);
+		this.resetStreams();
+		handler.handleMessage(new StringMessage("foo"));
+		assertEquals("foo" + System.getProperty("line.separator"), out.toString());
+	}
+
+}

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/DefaultConfigurationTests-context.xml

@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:stream="http://www.springframework.org/schema/integration/stream"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
+
+	<stream:stdout-channel-adapter id="adapter" auto-startup="false"/>
+
+</beans>

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/DefaultConfigurationTests.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2002-2009 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.stream.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.integration.channel.MessagePublishingErrorHandler;
+import org.springframework.integration.channel.NullChannel;
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Mark Fisher
+ * @since 1.0.3
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class DefaultConfigurationTests {
+
+	@Autowired
+	private ApplicationContext context;
+
+
+	@Test
+	public void verifyErrorChannel() {
+		Object errorChannel = context.getBean("errorChannel");
+		assertNotNull(errorChannel);
+		assertEquals(PublishSubscribeChannel.class, errorChannel.getClass());
+	}
+
+	@Test
+	public void verifyNullChannel() {
+		Object nullChannel = context.getBean("nullChannel");
+		assertNotNull(nullChannel);
+		assertEquals(NullChannel.class, nullChannel.getClass());
+	}
+
+	@Test
+	public void verifyTaskScheduler() {
+		Object taskScheduler = context.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME);
+		assertEquals(ThreadPoolTaskScheduler.class, taskScheduler.getClass());
+		Object errorHandler = new DirectFieldAccessor(taskScheduler).getPropertyValue("errorHandler");
+		assertEquals(MessagePublishingErrorHandler.class, errorHandler.getClass());
+		Object defaultErrorChannel = new DirectFieldAccessor(errorHandler).getPropertyValue("defaultErrorChannel");
+		assertEquals(context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME), defaultErrorChannel);
+	}
+
+}

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/consoleInboundChannelAdapterParserTests.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration/stream"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
+
+	<stdin-channel-adapter id="adapterWithDefaultCharset" auto-startup="false"/>
+
+	<stdin-channel-adapter id="adapterWithProvidedCharset" charset="UTF-8" auto-startup="false"/>
+
+	<integration:poller id="poller" default="true">
+		<integration:interval-trigger interval="3000"/>
+	</integration:poller>
+
+</beans:beans>

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/consoleOutboundChannelAdapterParserTests.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration/stream"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
+
+	<integration:channel id="testChannel"/>
+
+	<stdout-channel-adapter id="stdoutAdapterWithDefaultCharset" channel="testChannel"/>
+
+	<stdout-channel-adapter id="stdoutAdapterWithProvidedCharset" charset="UTF-8" channel="testChannel"/>
+
+	<stderr-channel-adapter id="stderrAdapter" channel="testChannel"/>
+
+	<stdout-channel-adapter id="newlineAdapter" append-newline="true" channel="testChannel"/>
+
+</beans:beans>

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/invalidConsoleInboundChannelAdapterParserTests.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration/stream"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
+
+	<stdin-channel-adapter id="adapterWithInvalidCharset" charset="invalid-charset-name"/>
+
+</beans:beans>

spring-integration-stream/src/test/java/org/springframework/integration/stream/config/invalidConsoleOutboundChannelAdapterParserTests.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration/stream"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
+
+	<stdout-channel-adapter id="adapterWithInvalidCharset" charset="invalid-charset-name"/>
+
+</beans:beans>