Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4118: Detect EOF on stdin

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4118

Add an option to the `CharacterStreamReadingMessageSource` to detect EOF
on the stream and close the context.

Publish event instead of closing context.

Polishing - PR Comments

Polishing

Docs about stopping the poller.

* Fix typos in the docs
This commit is contained in:
Gary Russell
2016-09-22 11:54:00 -04:00
committed by Artem Bilan
parent a7171c4db9
commit f9ddefec2c
9 changed files with 306 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -22,10 +22,12 @@ import java.io.InputStreamReader;
import java.io.Reader;
import java.io.UnsupportedEncodingException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
@@ -33,19 +35,61 @@ import org.springframework.util.Assert;
* A pollable source for {@link Reader Readers}.
*
* @author Mark Fisher
* @author Gary Russell
*/
public class CharacterStreamReadingMessageSource extends IntegrationObjectSupport implements MessageSource<String> {
public class CharacterStreamReadingMessageSource extends IntegrationObjectSupport implements MessageSource<String>,
ApplicationEventPublisherAware {
private final BufferedReader reader;
private final Object monitor;
private final boolean blockToDetectEOF;
private ApplicationEventPublisher applicationEventPublisher;
/**
* Construct an instance with the provider reader.
* {@link #receive()} will return {@code null} when the reader is not ready.
* @param reader the reader.
*/
public CharacterStreamReadingMessageSource(Reader reader) {
this(reader, -1);
this(reader, -1, false);
}
/**
* Construct an instance with the provider reader and buffer size.
* {@link #receive()} will return {@code null} when the reader is not ready.
* @param reader the reader.
* @param bufferSize the buffer size.
*/
public CharacterStreamReadingMessageSource(Reader reader, int bufferSize) {
this(reader, bufferSize, false);
}
/**
* Construct an instance with the provided reader and buffer size.
* When {@code blockToDetectEOF} is {@code false},
* {@link #receive()} will return {@code null} when the reader is not ready.
* When it is {@code true}, the thread will block until data is available; when the
* underlying stream is closed, a {@link StreamClosedEvent} is published to inform
* the application via an {@link org.springframework.context.ApplicationListener}.
* This can be useful, for example, when piping stdin
* <pre class="code">
* cat foo.txt | java -jar my.jar
* </pre>
* or
* <pre class="code">
* java -jar my.jar &lt; foo.txt
* </pre>
* @param reader the reader.
* @param bufferSize the buffer size; if negative use the default in
* {@link BufferedReader}.
* @param blockToDetectEOF true to block the thread until data is available and
* publish a {@link StreamClosedEvent} at EOF.
* @since 5.0
*/
public CharacterStreamReadingMessageSource(Reader reader, int bufferSize, boolean blockToDetectEOF) {
Assert.notNull(reader, "reader must not be null");
this.monitor = reader;
if (reader instanceof BufferedReader) {
@@ -57,20 +101,30 @@ public class CharacterStreamReadingMessageSource extends IntegrationObjectSuppor
else {
this.reader = new BufferedReader(reader);
}
this.blockToDetectEOF = blockToDetectEOF;
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
public String getComponentType() {
return "stream:stdin-channel-adapter(character)";
}
@Override
public Message<String> receive() {
try {
synchronized (this.monitor) {
if (!this.reader.ready()) {
if (!this.blockToDetectEOF && !this.reader.ready()) {
return null;
}
String line = this.reader.readLine();
if (line == null && this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new StreamClosedEvent(this));
}
return (line != null) ? new GenericMessage<String>(line) : null;
}
}
@@ -80,10 +134,19 @@ public class CharacterStreamReadingMessageSource extends IntegrationObjectSuppor
}
/**
* Create a source that reads from {@link System#in}. EOF will not be detected.
* @return the stream.
*/
public static final CharacterStreamReadingMessageSource stdin() {
return new CharacterStreamReadingMessageSource(new InputStreamReader(System.in));
}
/**
* Create a source that reads from {@link System#in}. EOF will not be detected.
* @param charsetName the charset to use when converting bytes to String.
* @return the stream.
*/
public static final CharacterStreamReadingMessageSource stdin(String charsetName) {
try {
return new CharacterStreamReadingMessageSource(new InputStreamReader(System.in, charsetName));
@@ -93,4 +156,32 @@ public class CharacterStreamReadingMessageSource extends IntegrationObjectSuppor
}
}
/**
* Create a source that reads from {@link System#in}. EOF will be detected and the application
* context closed.
* @return the stream.
* @see CharacterStreamReadingMessageSource#CharacterStreamReadingMessageSource(Reader, int, boolean)
* @since 5.0
*/
public static final CharacterStreamReadingMessageSource stdinPipe() {
return new CharacterStreamReadingMessageSource(new InputStreamReader(System.in), -1, true);
}
/**
* Create a source that reads from {@link System#in}. EOF will be detected and the application
* context closed.
* @param charsetName the charset to use when converting bytes to String.
* @return the stream.
* @see CharacterStreamReadingMessageSource#CharacterStreamReadingMessageSource(Reader, int, boolean)
* @since 5.0
*/
public static final CharacterStreamReadingMessageSource stdinPipe(String charsetName) {
try {
return new CharacterStreamReadingMessageSource(new InputStreamReader(System.in, charsetName), -1, true);
}
catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("unsupported encoding: " + charsetName, e);
}
}
}

35
spring-integration-stream/src/main/java/org/springframework/integration/stream/StreamClosedEvent.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* Copyright 2016 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 org.springframework.integration.event.IntegrationEvent;
/**
* Application event published when EOF is detected on a stream.
*
* @author Gary Russell
* @since 5.0
*
*/
@SuppressWarnings("serial")
public class StreamClosedEvent extends IntegrationEvent {
public StreamClosedEvent(Object source) {
super(source);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -22,20 +22,28 @@ import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.stream.CharacterStreamReadingMessageSource;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;stdin-channel-adapter&gt; element.
*
* @author Mark Fisher
* @author Gary Russell
*/
public class ConsoleInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.stream.CharacterStreamReadingMessageSource");
builder.setFactoryMethod("stdin");
CharacterStreamReadingMessageSource.class);
String pipe = element.getAttribute("detect-eof");
if (StringUtils.hasText(pipe)) {
builder.setFactoryMethod("stdinPipe");
}
else {
builder.setFactoryMethod("stdin");
}
String charsetName = element.getAttribute("charset");
if (StringUtils.hasText(charsetName)) {
builder.addConstructorArgValue(charsetName);

21
spring-integration-stream/src/main/resources/org/springframework/integration/stream/config/spring-integration-stream-5.0.xsd
View File

@@ -32,7 +32,26 @@
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="charset" type="xsd:string"/>
<xsd:attribute name="charset" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The charset to use when converting the byte stream from stdin to
String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="detect-eof">
<xsd:annotation>
<xsd:documentation>
When 'true' an application event is published when EOF is detected
on stdin. To facilitate this, the poller thread will block until data is
present or EOF is detected.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2016 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.
@@ -17,16 +17,31 @@
package org.springframework.integration.stream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import java.io.StringReader;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.PeriodicTrigger;
/**
* @author Mark Fisher
* @author Gary Russell
*/
public class CharacterStreamSourceTests {
@@ -40,4 +55,49 @@ public class CharacterStreamSourceTests {
assertNull(message2);
}
@Test
public void testEOF() {
StringReader reader = new StringReader("test");
CharacterStreamReadingMessageSource source = new CharacterStreamReadingMessageSource(reader, -1, true);
ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class);
source.setApplicationEventPublisher(publisher);
Message<?> message1 = source.receive();
assertEquals("test", message1.getPayload());
Message<?> message2 = source.receive();
assertNull(message2);
verify(publisher).publishEvent(any(StreamClosedEvent.class));
}
@Test
public void testEOFIntegrationTest() throws Exception {
StringReader reader = new StringReader("test");
CharacterStreamReadingMessageSource source = new CharacterStreamReadingMessageSource(reader, -1, true);
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
CountDownLatch latch = new CountDownLatch(2);
source.setApplicationEventPublisher(e -> {
if (e instanceof StreamClosedEvent) {
if (latch.getCount() == 1) {
adapter.stop();
}
latch.countDown();
}
});
adapter.setSource(source);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.afterPropertiesSet();
adapter.setTaskScheduler(scheduler);
adapter.setTrigger(new PeriodicTrigger(100));
QueueChannel out = new QueueChannel();
adapter.setOutputChannel(out);
adapter.setBeanFactory(mock(BeanFactory.class));
adapter.afterPropertiesSet();
adapter.start();
Message<?> received = out.receive(10000);
assertNotNull(received);
assertEquals("test", received.getPayload());
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(adapter.isRunning());
scheduler.shutdown();
}
}

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

@@ -36,11 +36,13 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
/**
* @author Mark Fisher
* @author Gunnar Hillert
* @author Gary Russell
*/
public class ConsoleInboundChannelAdapterParserTests {
@@ -54,8 +56,8 @@ public class ConsoleInboundChannelAdapterParserTests {
public void adapterWithDefaultCharset() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"consoleInboundChannelAdapterParserTests.xml", ConsoleInboundChannelAdapterParserTests.class);
SourcePollingChannelAdapter adapter =
(SourcePollingChannelAdapter) context.getBean("adapterWithDefaultCharset.adapter");
SourcePollingChannelAdapter adapter = context.getBean("adapterWithDefaultCharset.adapter",
SourcePollingChannelAdapter.class);
MessageSource<?> source = (MessageSource<?>) new DirectFieldAccessor(adapter).getPropertyValue("source");
assertTrue(source instanceof NamedComponent);
assertEquals("adapterWithDefaultCharset.adapter", adapter.getComponentName());
@@ -72,6 +74,9 @@ public class ConsoleInboundChannelAdapterParserTests {
Message<?> message = source.receive();
assertNotNull(message);
assertEquals("foo", message.getPayload());
adapter = context.getBean("pipedAdapterNoCharset.adapter", SourcePollingChannelAdapter.class);
source = adapter.getMessageSource();
assertTrue(TestUtils.getPropertyValue(source, "blockToDetectEOF", Boolean.class));
context.close();
}
@@ -79,12 +84,13 @@ public class ConsoleInboundChannelAdapterParserTests {
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");
SourcePollingChannelAdapter adapter = context.getBean("adapterWithProvidedCharset.adapter",
SourcePollingChannelAdapter.class);
MessageSource<?> source = adapter.getMessageSource();
DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(source);
Reader bufferedReader = (Reader) sourceAccessor.getPropertyValue("reader");
assertEquals(BufferedReader.class, bufferedReader.getClass());
assertEquals(false, sourceAccessor.getPropertyValue("blockToDetectEOF"));
DirectFieldAccessor bufferedReaderAccessor = new DirectFieldAccessor(bufferedReader);
Reader reader = (Reader) bufferedReaderAccessor.getPropertyValue("in");
assertEquals(InputStreamReader.class, reader.getClass());
@@ -93,6 +99,16 @@ public class ConsoleInboundChannelAdapterParserTests {
Message<?> message = source.receive();
assertNotNull(message);
assertEquals("foo", message.getPayload());
adapter = context.getBean("pipedAdapterWithCharset.adapter", SourcePollingChannelAdapter.class);
source = adapter.getMessageSource();
assertTrue(TestUtils.getPropertyValue(source, "blockToDetectEOF", Boolean.class));
bufferedReader = (Reader) sourceAccessor.getPropertyValue("reader");
assertEquals(BufferedReader.class, bufferedReader.getClass());
bufferedReaderAccessor = new DirectFieldAccessor(bufferedReader);
reader = (Reader) bufferedReaderAccessor.getPropertyValue("in");
assertEquals(InputStreamReader.class, reader.getClass());
readerCharset = Charset.forName(((InputStreamReader) reader).getEncoding());
assertEquals(Charset.forName("UTF-8"), readerCharset);
context.close();
}

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

@@ -10,12 +10,17 @@
http://www.springframework.org/schema/integration/stream
http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
<integration:message-history/>
<integration:message-history />
<stdin-channel-adapter id="adapterWithDefaultCharset" auto-startup="false"/>
<stdin-channel-adapter id="adapterWithDefaultCharset" auto-startup="false" />
<stdin-channel-adapter id="adapterWithProvidedCharset" charset="UTF-8" auto-startup="false"/>
<stdin-channel-adapter id="adapterWithProvidedCharset" charset="UTF-8" auto-startup="false" />
<integration:poller id="poller" default="true" fixed-rate="3000"/>
<stdin-channel-adapter id="pipedAdapterNoCharset" auto-startup="false" detect-eof="true" />
<stdin-channel-adapter id="pipedAdapterWithCharset" charset="UTF-8" auto-startup="false"
detect-eof="true" />
<integration:poller id="poller" default="true" fixed-rate="3000" />
</beans:beans>

48
src/reference/asciidoc/stream.adoc
View File

@@ -16,7 +16,8 @@ Both `ByteStreamReadingMessageSource` and `CharacterStreamReadingMessageSource`
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 will attempt to read into each `Message`.
The default value is 1024
The default value is 1024.
[source,xml]
----
<bean class="org.springframework.integration.stream.ByteStreamReadingMessageSource">
@@ -27,7 +28,50 @@ The default value is 1024
<bean class="org.springframework.integration.stream.CharacterStreamReadingMessageSource">
<constructor-arg ref="someReader"/>
</bean>
----
The `CharacterStreamReadingMessageSource` wraps the reader in a `BufferedReader` (if it's 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` (default), the `receive()` method checks if the reader is `ready()` and returns null if not.
EOF 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 implementing `ApplicationListener<StreamClosedEvent>`.
NOTE: To facilitate EOF detection, the poller thread will block in the `receive()` method until either data arrives or EOF is detected.
IMPORTANT: The poller will continue 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 and stopping the adapter will cause 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`, such as...
[source]
----
cat foo.txt | java -jar my.jar
----
or
[source]
----
java -jar my.jar < foo.txt
----
allowing the application to terminate 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]]
@@ -77,6 +121,8 @@ To configure the inbound channel adapter the following code snippet shows the di
<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 code snippet shows the different configuration for an outbound channel adapters.

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -45,3 +45,8 @@ The inbound channel adapters now have a property `max-fetch-size` which is used
Since _version 4.3.2_ a new `spring.integration.readOnly.headers` global property has been added to customize the list of headers which should not be copied to a newly created `Message` by the `MessageBuilder`.
See <<global-properties>> for more information.
==== Stream Changes
There is a new option on the `CharacterStreamReadingMessageSource` to allow it to be used to "pipe" stdin and publish an application event when the pipe is closed.
See <<stream-reading>> for more information.