spring-integration/src/reference/asciidoc/stream.adoc

[[stream]]
== Stream Support

In many cases, application data is obtained from a stream.
It is not recommended sending a reference to a stream as a message payload to a consumer.
Instead, messages are created from data that is read from an input stream, and message payloads are written to an output stream one by one.

You need to include this dependency into your project:

====
[source, xml, subs="normal", role="primary"]
.Maven
----
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stream</artifactId>
    <version>{project-version}</version>
</dependency>
----
[source, groovy, subs="normal", role="secondary"]
.Gradle
----
compile "org.springframework.integration:spring-integration-stream:{project-version}"
----
====

[[stream-reading]]
=== Reading from Streams

Spring Integration provides two adapters for streams.
Both `ByteStreamReadingMessageSource` and `CharacterStreamReadingMessageSource` implement `MessageSource`.
By configuring one of these within a channel-adapter element, the polling period can be configured and the message bus can automatically detect and schedule them.
The byte stream version requires an `InputStream`, and the character stream version requires a `Reader` as the single constructor argument.
The `ByteStreamReadingMessageSource` also accepts the 'bytesPerMessage' property to determine how many bytes it tries to read into each `Message`.
The default value is `1024`.
The following example creates an input stream that creates messages that each contain 2048 bytes:

====
[source,xml]
----
<bean class="org.springframework.integration.stream.ByteStreamReadingMessageSource">
  <constructor-arg ref="someInputStream"/>
  <property name="bytesPerMessage" value="2048"/>
</bean>

<bean class="org.springframework.integration.stream.CharacterStreamReadingMessageSource">
  <constructor-arg ref="someReader"/>
</bean>
----
====

The `CharacterStreamReadingMessageSource` wraps the reader in a `BufferedReader` (if it is not one already).
You can set the buffer size used by the buffered reader in the second constructor argument.
Starting with version 5.0, a third constructor argument (`blockToDetectEOF`) controls the behavior of the `CharacterStreamReadingMessageSource`.
When `false` (the default), the `receive()` method checks whether the reader is `ready()` and returns null if not.
EOF (end of file) is not detected in this case.
When `true`, the `receive()` method blocks until data is available or EOF is detected on the underlying stream.
When EOF is detected, a `StreamClosedEvent` (application event) is published.
You can consume this event with a bean that implements `ApplicationListener<StreamClosedEvent>`.

NOTE: To facilitate EOF detection, the poller thread blocks in the `receive()` method until either data arrives or EOF is detected.

IMPORTANT: The poller continues to publish an event on each poll once EOF has been detected.
The application listener can stop the adapter to prevent this.
The event is published on the poller thread.
Stopping the adapter causes the thread to be interrupted.
If you intend to perform some interruptible task after stopping the adapter, you must either perform the `stop()` on a different thread or use a different thread for those downstream activities.
Note that sending to a `QueueChannel` is interruptible, so, if you wish to send a message from the listener, do it before stopping the adapter.

This facilitates "`piping`" or redirecting data to `stdin`, as the following two examples shows:

====
[source]
----
cat myfile.txt | java -jar my.jar
----

[source]
----
java -jar my.jar < foo.txt
----
====

This approach lets the application stop when the pipe is closed.

Four convenient factory methods are available:

====
[source, java]
----
public static final CharacterStreamReadingMessageSource stdin() { ... }

public static final CharacterStreamReadingMessageSource stdin(String charsetName) { ... }

public static final CharacterStreamReadingMessageSource stdinPipe() { ... }

public static final CharacterStreamReadingMessageSource stdinPipe(String charsetName) { ... }
----
====

[[stream-writing]]
=== Writing to Streams

For target streams, you can use either of two implementations: `ByteStreamWritingMessageHandler` or `CharacterStreamWritingMessageHandler`.
Each requires a single constructor argument (`OutputStream` for byte streams or `Writer` for character streams), and each provides a second constructor that adds the optional 'bufferSize'.
Since both of these ultimately implement the `MessageHandler` interface, you can reference them from a `channel-adapter` configuration, as described in <<./channel-adapter.adoc#channel-adapter,Channel Adapter>>.

====
[source,xml]
----
<bean class="org.springframework.integration.stream.ByteStreamWritingMessageHandler">
  <constructor-arg ref="someOutputStream"/>
  <constructor-arg value="1024"/>
</bean>

<bean class="org.springframework.integration.stream.CharacterStreamWritingMessageHandler">
  <constructor-arg ref="someWriter"/>
</bean>
----
====

[[stream-namespace]]
=== Stream Namespace Support

Spring Integration defines a namespace to reduce the configuration needed for stream-related channel adapters.
The following schema locations are needed to use it:

====
[source,xml]
----
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns:int-stream="http://www.springframework.org/schema/integration/stream"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:beans="http://www.springframework.org/schema/beans"
  xsi:schemaLocation="http://www.springframework.org/schema/beans
      https://www.springframework.org/schema/beans/spring-beans.xsd
      http://www.springframework.org/schema/integration/stream
      https://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
----
====

The following code snippet shows the different configuration options that are supported to configure the inbound channel adapter:

====
[source,xml]
----
<int-stream:stdin-channel-adapter id="adapterWithDefaultCharset"/>

<int-stream:stdin-channel-adapter id="adapterWithProvidedCharset" charset="UTF-8"/>
----
====

Starting with version 5.0, you can set the `detect-eof` attribute, which sets the `blockToDetectEOF` property.
See <<stream-reading>> for more information.

To configure the outbound channel adapter, you can use the namespace support as well.
The following example shows the different configuration for an outbound channel adapters:

====
[source,xml]
----
<int-stream:stdout-channel-adapter id="stdoutAdapterWithDefaultCharset"
    channel="testChannel"/>

<int-stream:stdout-channel-adapter id="stdoutAdapterWithProvidedCharset" charset="UTF-8"
    channel="testChannel"/>

<int-stream:stderr-channel-adapter id="stderrAdapter" channel="testChannel"/>

<int-stream:stdout-channel-adapter id="newlineAdapter" append-newline="true"
    channel="testChannel"/>
----
====