From 39a6667a760c57e0872e8492d049b0af2c527175 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 11 Jan 2008 20:36:55 +0000 Subject: [PATCH] Added source adapter for character-based input streams. --- .../adapter/AbstractSourceAdapter.java | 10 ++ .../adapter/PollingSourceAdapter.java | 3 + .../adapter/stream/CharacterStreamSource.java | 65 +++++++++++ .../stream/CharacterStreamSourceAdapter.java | 37 ++++++ .../CharacterStreamSourceAdapterTests.java | 110 ++++++++++++++++++ 5 files changed, 225 insertions(+) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java index 3fc6bebb9a..9f50397e46 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java @@ -45,6 +45,8 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial private long sendTimeout = -1; + private volatile boolean initialized = false; + public void setChannel(MessageChannel channel) { Assert.notNull(channel, "'channel' must not be null"); @@ -78,6 +80,11 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial throw new MessagingConfigurationException("'channel' is required"); } this.initialize(); + this.initialized = true; + } + + protected boolean isInitialized() { + return this.initialized; } /** @@ -87,6 +94,9 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial } protected boolean sendToChannel(T object) { + if (!this.initialized) { + this.afterPropertiesSet(); + } if (object == null) { if (logger.isDebugEnabled()) { logger.debug("adapter attempted to send a null object"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 90156b46b2..a91a6e2a2e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -52,6 +52,9 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements } public void start() { + if (!this.isInitialized()) { + this.afterPropertiesSet(); + } this.running = true; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java new file mode 100644 index 0000000000..575b3d9f72 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java @@ -0,0 +1,65 @@ +/* + * Copyright 2002-2007 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.adapter.stream; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.adapter.PollableSource; + +/** + * A pollable source for text-based {@link InputStream InputStreams}. + * + * @author Mark Fisher + */ +public class CharacterStreamSource implements PollableSource { + + private BufferedReader reader; + + + public CharacterStreamSource(InputStream stream) { + this.reader = new BufferedReader(new InputStreamReader(stream)); + } + + public Collection poll(int limit) { + List results = new ArrayList(); + while (results.size() < limit) { + try { + boolean isReady = reader.ready(); + if (!isReady) { + return results; + } + String line = reader.readLine(); + if (line == null) { + return results; + } + results.add(line); + } + catch (IOException e) { + throw new MessageDeliveryException("IO failure occurred in adapter", e); + } + } + return results; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java new file mode 100644 index 0000000000..df24d9436c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java @@ -0,0 +1,37 @@ +/* + * Copyright 2002-2007 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.adapter.stream; + +import java.io.InputStream; + +import org.springframework.integration.adapter.PollingSourceAdapter; + +/** + * A polling source adapter that wraps a {@link CharacterStreamSource}. + * + * @author Mark Fisher + */ +public class CharacterStreamSourceAdapter extends PollingSourceAdapter { + + public static final CharacterStreamSourceAdapter STDIN_ADAPTER = new CharacterStreamSourceAdapter(System.in); + + + public CharacterStreamSourceAdapter(InputStream stream) { + super(new CharacterStreamSource(stream)); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java new file mode 100644 index 0000000000..d5fa7a836f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java @@ -0,0 +1,110 @@ +/* + * Copyright 2002-2007 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.adapter.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.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.message.Message; + +/** + * @author Mark Fisher + */ +public class CharacterStreamSourceAdapterTests { + + @Test + public void testEndOfStream() { + byte[] bytes = "test".getBytes(); + ByteArrayInputStream stream = new ByteArrayInputStream(bytes); + MessageChannel channel = new SimpleChannel(); + CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream); + adapter.setChannel(channel); + adapter.start(); + int count = adapter.dispatch(); + assertEquals(1, count); + Message message1 = channel.receive(0); + assertEquals("test", message1.getPayload()); + Message message2 = channel.receive(0); + assertNull(message2); + adapter.dispatch(); + Message message3 = channel.receive(0); + assertNull(message3); + } + + @Test + public void testEndOfStreamWithMaxMessagesPerTask() { + byte[] bytes = "test".getBytes(); + ByteArrayInputStream stream = new ByteArrayInputStream(bytes); + MessageChannel channel = new SimpleChannel(); + CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream); + adapter.setChannel(channel); + adapter.setMaxMessagesPerTask(5); + adapter.start(); + int count = adapter.dispatch(); + assertEquals(1, count); + Message message1 = channel.receive(0); + assertEquals("test", message1.getPayload()); + Message message2 = channel.receive(0); + assertNull(message2); + } + + @Test + public void testMultipleLinesWithSingleMessagePerTask() { + String s = "test1" + System.getProperty("line.separator") + "test2"; + ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes()); + MessageChannel channel = new SimpleChannel(); + CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream); + adapter.setMaxMessagesPerTask(1); + adapter.setChannel(channel); + adapter.start(); + int count = adapter.dispatch(); + assertEquals(1, count); + Message message1 = channel.receive(0); + assertEquals("test1", message1.getPayload()); + Message message2 = channel.receive(0); + assertNull(message2); + adapter.dispatch(); + Message message3 = channel.receive(0); + assertEquals("test2", message3.getPayload()); + } + + @Test + public void testLessThanMaxMessagesAvailable() { + String s = "test1" + System.getProperty("line.separator") + "test2"; + ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes()); + MessageChannel channel = new SimpleChannel(); + CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream); + adapter.setChannel(channel); + adapter.setMaxMessagesPerTask(5); + adapter.start(); + int count = adapter.dispatch(); + assertEquals(2, count); + Message message1 = channel.receive(0); + assertEquals("test1", message1.getPayload()); + Message message2 = channel.receive(0); + assertEquals("test2", message2.getPayload()); + Message message3 = channel.receive(0); + assertNull(message3); + } + +}