From f9834daa15f754b8956916ce2ad31fa01c23ee75 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 3 Apr 2008 00:21:04 +0000 Subject: [PATCH] CharacterStreamSourceAdapter now takes a Reader, and CharacterStreamTargetAdapter takes a Writer (INT-148). --- .../adapter/stream/CharacterStreamSource.java | 36 +++++----- .../stream/CharacterStreamSourceAdapter.java | 12 ++-- .../stream/CharacterStreamTargetAdapter.java | 31 +++++---- .../CharacterStreamSourceAdapterTests.java | 22 +++---- .../CharacterStreamTargetAdapterTests.java | 66 ++++++++----------- 5 files changed, 83 insertions(+), 84 deletions(-) diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java index 0e71df1ce4..c09621112e 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java @@ -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 { - 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 { 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; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java index 69cf8621cf..82bc48f867 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java @@ -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 { - 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 { * 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; } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapter.java index e5cace6b03..f3ec152eb5 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapter.java @@ -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)); } diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java index 52d32815b5..1bbf2455e9 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java @@ -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(); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index 6297fe3c48..6303241227 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -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)); 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(testObject1), 0); channel.send(new GenericMessage(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(testObject1), 0); channel.send(new GenericMessage(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()); }