From f9ddefec2cca10bb8f1a0d030cb1aacf79b140e8 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 22 Sep 2016 11:54:00 -0400 Subject: [PATCH] INT-4118: Detect EOF on stdin 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 --- .../CharacterStreamReadingMessageSource.java | 103 +++++++++++++++++- .../integration/stream/StreamClosedEvent.java | 35 ++++++ .../ConsoleInboundChannelAdapterParser.java | 14 ++- .../config/spring-integration-stream-5.0.xsd | 21 +++- .../stream/CharacterStreamSourceTests.java | 62 ++++++++++- ...nsoleInboundChannelAdapterParserTests.java | 26 ++++- ...onsoleInboundChannelAdapterParserTests.xml | 13 ++- src/reference/asciidoc/stream.adoc | 48 +++++++- src/reference/asciidoc/whats-new.adoc | 5 + 9 files changed, 306 insertions(+), 21 deletions(-) create mode 100644 spring-integration-stream/src/main/java/org/springframework/integration/stream/StreamClosedEvent.java diff --git a/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java b/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java index 5c22e8db7e..b8fc4beccf 100644 --- a/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java +++ b/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java @@ -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 { +public class CharacterStreamReadingMessageSource extends IntegrationObjectSupport implements MessageSource, + 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 + * 
+	 *     cat foo.txt | java -jar my.jar
+	 *
+ * or + * 
+	 *     java -jar my.jar < foo.txt
+	 *
+ * @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 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(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); + } + } + } diff --git a/spring-integration-stream/src/main/java/org/springframework/integration/stream/StreamClosedEvent.java b/spring-integration-stream/src/main/java/org/springframework/integration/stream/StreamClosedEvent.java new file mode 100644 index 0000000000..20c885b8ea --- /dev/null +++ b/spring-integration-stream/src/main/java/org/springframework/integration/stream/StreamClosedEvent.java @@ -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); + } + +} diff --git a/spring-integration-stream/src/main/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParser.java b/spring-integration-stream/src/main/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParser.java index a04d302919..61fe6e1253 100644 --- a/spring-integration-stream/src/main/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParser.java +++ b/spring-integration-stream/src/main/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParser.java @@ -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 <stdin-channel-adapter> 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); diff --git a/spring-integration-stream/src/main/resources/org/springframework/integration/stream/config/spring-integration-stream-5.0.xsd b/spring-integration-stream/src/main/resources/org/springframework/integration/stream/config/spring-integration-stream-5.0.xsd index abde524207..81bdcf8835 100644 --- a/spring-integration-stream/src/main/resources/org/springframework/integration/stream/config/spring-integration-stream-5.0.xsd +++ b/spring-integration-stream/src/main/resources/org/springframework/integration/stream/config/spring-integration-stream-5.0.xsd @@ -32,7 +32,26 @@ - + + + + The charset to use when converting the byte stream from stdin to + String. + + + + + + + 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. + + + + + + diff --git a/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamSourceTests.java b/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamSourceTests.java index 08fb3d6305..0e38cb9ddc 100644 --- a/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamSourceTests.java +++ b/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamSourceTests.java @@ -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(); + } + } diff --git a/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParserTests.java b/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParserTests.java index 28869548bb..062b264bdb 100644 --- a/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParserTests.java +++ b/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/ConsoleInboundChannelAdapterParserTests.java @@ -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(); } diff --git a/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/consoleInboundChannelAdapterParserTests.xml b/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/consoleInboundChannelAdapterParserTests.xml index d8c25cf996..eee30b094f 100644 --- a/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/consoleInboundChannelAdapterParserTests.xml +++ b/spring-integration-stream/src/test/java/org/springframework/integration/stream/config/consoleInboundChannelAdapterParserTests.xml @@ -10,12 +10,17 @@ http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd"> - + - + - + - + + + + + diff --git a/src/reference/asciidoc/stream.adoc b/src/reference/asciidoc/stream.adoc index 11f26b6ce4..21938a83c1 100644 --- a/src/reference/asciidoc/stream.adoc +++ b/src/reference/asciidoc/stream.adoc @@ -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] ---- @@ -27,7 +28,50 @@ The default value is 1024 +---- +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`. + +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 ---- +Starting with _version 5.0_ you can set the `detect-eof` attribute which sets the `blockToDetectEOF` property - see <> 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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 70c601c2ea..738fed8ae6 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -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 <> 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 <> for more information.