Migrated the stream-based adapters from the "adapters" module to "org.springframework.integration.stream" (INT-375).

org.springframework.integration.stream/src/main/java/org/springframework/integration/stream/ByteStreamSource.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.message.GenericMessage;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.PollableSource;
+
+/**
+ * A pollable source for receiving bytes from an {@link InputStream}.
+ * 
+ * @author Mark Fisher
+ */
+public class ByteStreamSource implements PollableSource<byte[]> {
+
+	private BufferedInputStream stream;
+
+	private Object streamMonitor;
+
+	private int bytesPerMessage = 1024;
+
+	private boolean shouldTruncate = true;
+
+
+	public ByteStreamSource(InputStream stream) {
+		this(stream, -1);
+	}
+
+	public ByteStreamSource(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);
+		}
+	}
+
+}

org.springframework.integration.stream/src/main/java/org/springframework/integration/stream/ByteStreamTarget.java

@@ -0,0 +1,83 @@
+/*
+ * 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.endpoint.AbstractMessageConsumingEndpoint;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessagingException;
+
+/**
+ * A target that writes a byte array to an {@link OutputStream}.
+ * 
+ * @author Mark Fisher
+ */
+public class ByteStreamTarget extends AbstractMessageConsumingEndpoint {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final BufferedOutputStream stream;
+
+
+	public ByteStreamTarget(OutputStream stream) {
+		this(stream, -1);
+	}
+
+	public ByteStreamTarget(OutputStream stream, int bufferSize) {
+		if (bufferSize > 0) {
+			this.stream = new BufferedOutputStream(stream, bufferSize);
+		}
+		else {
+			this.stream = new BufferedOutputStream(stream);
+		}
+	}
+
+
+	@Override
+	public void processMessage(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);
+		}
+	}
+
+}

org.springframework.integration.stream/src/main/java/org/springframework/integration/stream/CharacterStreamSource.java

@@ -0,0 +1,91 @@
+/*
+ * 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.ConfigurationException;
+import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.util.Assert;
+
+/**
+ * A pollable source for {@link Reader Readers}.
+ * 
+ * @author Mark Fisher
+ */
+public class CharacterStreamSource implements PollableSource<String> {
+
+	private final BufferedReader reader;
+
+	private final Object monitor;
+
+
+	public CharacterStreamSource(Reader reader) {
+		this(reader, -1);
+	}
+
+	public CharacterStreamSource(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 CharacterStreamSource stdin() {
+		return new CharacterStreamSource(new InputStreamReader(System.in));
+	}
+
+	public static final CharacterStreamSource stdin(String charsetName) {
+		try {
+			return new CharacterStreamSource(new InputStreamReader(System.in, charsetName));
+		}
+		catch (UnsupportedEncodingException e) {
+			throw new ConfigurationException("unsupported encoding: " + charsetName, e);
+		}
+	}
+
+}

org.springframework.integration.stream/src/main/java/org/springframework/integration/stream/CharacterStreamTarget.java

@@ -0,0 +1,152 @@
+/*
+ * 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.UnsupportedEncodingException;
+import java.io.Writer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.endpoint.AbstractMessageConsumingEndpoint;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessagingException;
+import org.springframework.util.Assert;
+
+/**
+ * A target that writes to a {@link Writer}. String-based objects will be
+ * written directly, but if the object is not itself a {@link String}, the
+ * target will write the result of the object's {@link #toString()} method.
+ * To append a new-line after each write, set the {@link #shouldAppendNewLine}
+ * flag to <em>true</em>. It is <em>false</em> by default.
+ * 
+ * @author Mark Fisher
+ */
+public class CharacterStreamTarget extends AbstractMessageConsumingEndpoint {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final BufferedWriter writer;
+
+	private volatile boolean shouldAppendNewLine = false;
+
+
+	public CharacterStreamTarget(Writer writer) {
+		this(writer, -1);
+	}
+
+	public CharacterStreamTarget(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 CharacterStreamTarget stdout() {
+		return stdout(null);
+	}
+
+	/**
+	 * Factory method that creates a target for stdout (System.out) with the
+	 * specified charset encoding.
+	 */
+	public static CharacterStreamTarget 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 CharacterStreamTarget stderr() {
+		return stderr(null);
+	}
+
+	/**
+	 * Factory method that creates a target for stderr (System.err) with the
+	 * specified charset encoding.
+	 */	
+	public static CharacterStreamTarget stderr(String charsetName) {
+		return createTargetForStream(System.err, charsetName);
+	}
+
+	private static CharacterStreamTarget createTargetForStream(OutputStream stream, String charsetName) {
+		if (charsetName == null) {
+			return new CharacterStreamTarget(new OutputStreamWriter(stream));
+		}
+		try {
+			return new CharacterStreamTarget(new OutputStreamWriter(stream, charsetName));
+		}
+		catch (UnsupportedEncodingException e) {
+			throw new ConfigurationException("unsupported encoding: " + charsetName, e);
+		}
+	}
+
+
+	public void setShouldAppendNewLine(boolean shouldAppendNewLine) {
+		this.shouldAppendNewLine = shouldAppendNewLine;
+	}
+
+	@Override
+	public void processMessage(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 {
+				writer.write(payload.toString());
+			}
+			if (this.shouldAppendNewLine) {
+				writer.newLine();
+			}
+			writer.flush();
+		}
+		catch (IOException e) {
+			throw new MessagingException("IO failure occurred in target", e);
+		}
+	}
+
+}

org.springframework.integration.stream/src/main/java/org/springframework/integration/stream/config/ConsoleSourceParser.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.AbstractInboundChannelAdapterParser;
+import org.springframework.integration.stream.CharacterStreamSource;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <stdin-channel-adapter> element.
+ * 
+ * @author Mark Fisher
+ */
+public class ConsoleSourceParser extends AbstractInboundChannelAdapterParser {
+
+	@Override
+	protected String parseSource(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(CharacterStreamSource.class);
+		builder.setFactoryMethod("stdin");
+		String charsetName = element.getAttribute("charset");
+		if (StringUtils.hasText(charsetName)) {
+			builder.addConstructorArgValue(charsetName);
+		}
+		return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
+	}
+
+}

org.springframework.integration.stream/src/main/java/org/springframework/integration/stream/config/ConsoleTargetParser.java

@@ -0,0 +1,94 @@
+/*
+ * 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.BeanDefinitionStoreException;
+import org.springframework.beans.factory.config.BeanDefinitionHolder;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.stream.CharacterStreamTarget;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the "stdout-" and "stderr-channel-adapter" elements.
+ * 
+ * @author Mark Fisher
+ */
+public class ConsoleTargetParser extends AbstractSingleBeanDefinitionParser {
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return CharacterStreamTarget.class;
+	}
+
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) throws BeanDefinitionStoreException {
+		String id = element.getAttribute("id");
+		if (!element.hasAttribute("channel")) {
+			// the created channel will get the 'id', so the adapter's bean name includes a suffix
+			id = id + ".adapter";
+		}
+		else if (!StringUtils.hasText(id)) {
+			id = parserContext.getReaderContext().generateBeanName(definition);
+		}
+		return id;
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		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);
+		}
+		String channelName = element.getAttribute("channel");
+		if (StringUtils.hasText(channelName)) {
+			builder.addPropertyReference("inputChannel", channelName);
+		}
+		else {
+			builder.addPropertyReference("inputChannel", this.createDirectChannel(element, parserContext));
+		}
+	}
+
+	private String createDirectChannel(Element element, ParserContext parserContext) {
+		String channelId = element.getAttribute("id");
+		if (!StringUtils.hasText(channelId)) {
+			throw new ConfigurationException("The channel-adapter's 'id' attribute is required when no 'channel' "
+					+ "reference has been provided, because that 'id' would be used for the created channel.");
+		}
+		BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
+		BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId);
+		BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
+		return channelId;
+	}
+
+}

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

@@ -0,0 +1,32 @@
+/*
+ * 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.springframework.beans.factory.xml.NamespaceHandlerSupport;
+
+/**
+ * @author Mark Fisher
+ */
+public class StreamNamespaceHandler extends NamespaceHandlerSupport {
+
+	public void init() {
+		this.registerBeanDefinitionParser("stdin-channel-adapter", new ConsoleSourceParser());
+		this.registerBeanDefinitionParser("stdout-channel-adapter", new ConsoleTargetParser());
+		this.registerBeanDefinitionParser("stderr-channel-adapter", new ConsoleTargetParser());
+	}
+
+}

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

@@ -0,0 +1,53 @@
+<?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"/>
+
+	<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="charset" type="xsd:string"/>
+		</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:attribute name="charset" type="xsd:string"/>
+		<xsd:attribute name="append-newline" type="xsd:boolean" default="false"/>
+	</xsd:complexType>
+
+</xsd:schema>

org.springframework.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

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

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