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

CharacterStreamSourceAdapter now takes a Reader, and CharacterStreamTargetAdapter takes a Writer (INT-148).

Browse Source
This commit is contained in:
Mark Fisher
2008-04-03 00:21:04 +00:00
parent cbf63d7e90
commit f9834daa15
5 changed files with 83 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -18,38 +18,42 @@ package org.springframework.integration.adapter.stream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.util.Assert;
/**
* A pollable source for text-based {@link InputStream InputStreams}.
* A pollable source for {@link Reader Readers}.
*
* @author Mark Fisher
*/
public class CharacterStreamSource implements PollableSource<String> {
private BufferedReader reader;
private final BufferedReader reader;
private Object streamMonitor;
private final Object monitor;
public CharacterStreamSource(InputStream stream) {
this(stream, -1);
public CharacterStreamSource(Reader reader) {
this(reader, -1);
}
public CharacterStreamSource(InputStream stream, int bufferSize) {
this.streamMonitor = stream;
if (bufferSize > 0) {
this.reader = new BufferedReader(new InputStreamReader(stream), bufferSize);
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(new InputStreamReader(stream));
this.reader = new BufferedReader(reader);
}
}
@@ -59,12 +63,12 @@ public class CharacterStreamSource implements PollableSource<String> {
while (results.size() < limit) {
try {
String line = null;
synchronized (this.streamMonitor) {
boolean isReady = reader.ready();
synchronized (this.monitor) {
boolean isReady = this.reader.ready();
if (!isReady) {
return results;
}
line = reader.readLine();
line = this.reader.readLine();
}
if (line == null) {
return results;

12
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -16,7 +16,8 @@
package org.springframework.integration.adapter.stream;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.channel.MessageChannel;
@@ -28,8 +29,8 @@ import org.springframework.integration.channel.MessageChannel;
*/
public class CharacterStreamSourceAdapter extends PollingSourceAdapter<String> {
public CharacterStreamSourceAdapter(InputStream stream) {
super(new CharacterStreamSource(stream));
public CharacterStreamSourceAdapter(Reader reader) {
super(new CharacterStreamSource(reader));
}
@@ -37,7 +38,8 @@ public class CharacterStreamSourceAdapter extends PollingSourceAdapter<String> {
* Factory method that creates an adapter for stdin (System.in).
*/
public static CharacterStreamSourceAdapter stdinAdapter(MessageChannel channel) {
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(System.in);
CharacterStreamSourceAdapter adapter =
new CharacterStreamSourceAdapter(new InputStreamReader(System.in));
adapter.setChannel(channel);
return adapter;
}

31
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -18,14 +18,15 @@ package org.springframework.integration.adapter.stream;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import org.springframework.integration.adapter.AbstractTargetAdapter;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.util.Assert;
/**
* A target adapter that writes to an {@link OutputStream}. String-based
* A target adapter that writes to a {@link Writer}. String-based
* objects will be written directly, but if the object is not itself a
* {@link String}, the adapter will write the result of the object's
* {@link #toString()} method. To append a new-line after each write, set the
@@ -36,21 +37,25 @@ import org.springframework.integration.message.MessageHandlingException;
*/
public class CharacterStreamTargetAdapter extends AbstractTargetAdapter {
private BufferedWriter writer;
private final BufferedWriter writer;
private boolean shouldAppendNewLine = false;
private volatile boolean shouldAppendNewLine = false;
public CharacterStreamTargetAdapter(OutputStream stream) {
this(stream, -1);
public CharacterStreamTargetAdapter(Writer writer) {
this(writer, -1);
}
public CharacterStreamTargetAdapter(OutputStream stream, int bufferSize) {
if (bufferSize > 0) {
this.writer = new BufferedWriter(new OutputStreamWriter(stream), bufferSize);
public CharacterStreamTargetAdapter(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(new OutputStreamWriter(stream));
this.writer = new BufferedWriter(writer);
}
}
@@ -59,14 +64,14 @@ public class CharacterStreamTargetAdapter extends AbstractTargetAdapter {
* Factory method that creates an adapter for stdout (System.out).
*/
public static CharacterStreamTargetAdapter stdoutAdapter() {
return new CharacterStreamTargetAdapter(System.out);
return new CharacterStreamTargetAdapter(new OutputStreamWriter(System.out));
}
/**
* Factory method that creates an adapter for stderr (System.err).
*/
public static CharacterStreamTargetAdapter stderrAdapter() {
return new CharacterStreamTargetAdapter(System.err);
return new CharacterStreamTargetAdapter(new OutputStreamWriter(System.err));
}

22
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -19,7 +19,7 @@ package org.springframework.integration.adapter.stream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import java.io.ByteArrayInputStream;
import java.io.StringReader;
import org.junit.Test;
@@ -34,10 +34,9 @@ public class CharacterStreamSourceAdapterTests {
@Test
public void testEndOfStream() {
byte[] bytes = "test".getBytes();
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
StringReader reader = new StringReader("test");
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
adapter.setChannel(channel);
adapter.start();
int count = adapter.processMessages();
@@ -53,10 +52,9 @@ public class CharacterStreamSourceAdapterTests {
@Test
public void testEndOfStreamWithMaxMessagesPerTask() {
byte[] bytes = "test".getBytes();
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
StringReader reader = new StringReader("test");
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
adapter.setChannel(channel);
adapter.setMaxMessagesPerTask(5);
adapter.start();
@@ -71,9 +69,9 @@ public class CharacterStreamSourceAdapterTests {
@Test
public void testMultipleLinesWithSingleMessagePerTask() {
String s = "test1" + System.getProperty("line.separator") + "test2";
ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes());
StringReader reader = new StringReader(s);
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
adapter.setInitialDelay(10000);
adapter.setMaxMessagesPerTask(1);
adapter.setChannel(channel);
@@ -92,9 +90,9 @@ public class CharacterStreamSourceAdapterTests {
@Test
public void testLessThanMaxMessagesAvailable() {
String s = "test1" + System.getProperty("line.separator") + "test2";
ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes());
StringReader reader = new StringReader(s);
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
adapter.setChannel(channel);
adapter.setMaxMessagesPerTask(5);
adapter.start();

66
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -18,7 +18,7 @@ package org.springframework.integration.adapter.stream;
import static org.junit.Assert.assertEquals;
import java.io.ByteArrayOutputStream;
import java.io.StringWriter;
import org.junit.Test;
@@ -41,53 +41,48 @@ public class CharacterStreamTargetAdapterTests {
@Test
public void testSingleString() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
adapter.handle(new StringMessage("foo"));
String result = new String(stream.toByteArray());
assertEquals("foo", result);
assertEquals("foo", writer.toString());
}
@Test
public void testTwoStringsAndNoNewLinesByDefault() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
MessageChannel channel = new SimpleChannel();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(1, dispatcher.dispatch());
String result1 = new String(stream.toByteArray());
assertEquals("foo", result1);
assertEquals("foo", writer.toString());
assertEquals(1, dispatcher.dispatch());
String result2 = new String(stream.toByteArray());
assertEquals("foobar", result2);
assertEquals("foobar", writer.toString());
}
@Test
public void testTwoStringsWithNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
MessageChannel channel = new SimpleChannel();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
adapter.setShouldAppendNewLine(true);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(1, dispatcher.dispatch());
String result1 = new String(stream.toByteArray());
String newLine = System.getProperty("line.separator");
assertEquals("foo" + newLine, result1);
assertEquals("foo" + newLine, writer.toString());
assertEquals(1, dispatcher.dispatch());
String result2 = new String(stream.toByteArray());
assertEquals("foo" + newLine + "bar" + newLine, result2);
assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
}
@Test
public void testMaxMessagesPerTaskSameAsMessageCount() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(2);
SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy);
@@ -96,14 +91,13 @@ public class CharacterStreamTargetAdapterTests {
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(2, dispatcher.dispatch());
String result = new String(stream.toByteArray());
assertEquals("foobar", result);
assertEquals("foobar", writer.toString());
}
@Test
public void testMaxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(10);
dispatcherPolicy.setReceiveTimeout(0);
@@ -114,30 +108,28 @@ public class CharacterStreamTargetAdapterTests {
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(2, dispatcher.dispatch());
String result = new String(stream.toByteArray());
String newLine = System.getProperty("line.separator");
assertEquals("foo" + newLine + "bar" + newLine, result);
assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
}
@Test
public void testSingleNonStringObject() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
MessageChannel channel = new SimpleChannel();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
TestObject testObject = new TestObject("foo");
channel.send(new GenericMessage<TestObject>(testObject));
int count = dispatcher.dispatch();
assertEquals(1, count);
String result = new String(stream.toByteArray());
assertEquals("foo", result);
assertEquals("foo", writer.toString());
}
@Test
public void testTwoNonStringObjectWithOutNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setReceiveTimeout(0);
dispatcherPolicy.setMaxMessagesPerTask(2);
@@ -149,14 +141,13 @@ public class CharacterStreamTargetAdapterTests {
channel.send(new GenericMessage<TestObject>(testObject1), 0);
channel.send(new GenericMessage<TestObject>(testObject2), 0);
assertEquals(2, dispatcher.dispatch());
String result = new String(stream.toByteArray());
assertEquals("foobar", result);
assertEquals("foobar", writer.toString());
}
@Test
public void testTwoNonStringObjectWithNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
StringWriter writer = new StringWriter();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setReceiveTimeout(0);
dispatcherPolicy.setMaxMessagesPerTask(2);
@@ -169,9 +160,8 @@ public class CharacterStreamTargetAdapterTests {
channel.send(new GenericMessage<TestObject>(testObject1), 0);
channel.send(new GenericMessage<TestObject>(testObject2), 0);
dispatcher.dispatch();
String result = new String(stream.toByteArray());
String newLine = System.getProperty("line.separator");
assertEquals("foo" + newLine + "bar" + newLine, result);
assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
}