From 1f5e7b575633f1b605f96ece6b98c8892c945956 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 31 Jan 2008 22:04:35 +0000 Subject: [PATCH] SimpleChannel now creates a SynchronousQueue if the capacity is less than or equal to 0. SplitterMessageHandlerAdapter exposes a configurable 'sendTimeout' property. --- .../integration/channel/SimpleChannel.java | 8 +++- .../router/SplitterMessageHandlerAdapter.java | 8 +++- .../stream/ByteStreamTargetAdapterTests.java | 42 +++++++++---------- .../CharacterStreamTargetAdapterTests.java | 24 +++++------ .../ChannelPollingMessageRetrieverTests.java | 16 +++---- 5 files changed, 55 insertions(+), 43 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index ca5b2232ba..0a43c02a59 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; import org.springframework.integration.dispatcher.DispatcherPolicy; @@ -46,7 +47,12 @@ public class SimpleChannel extends AbstractMessageChannel { */ public SimpleChannel(int capacity, DispatcherPolicy dispatcherPolicy) { super((dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy()); - this.queue = new LinkedBlockingQueue>(capacity); + if (capacity > 0) { + this.queue = new LinkedBlockingQueue>(capacity); + } + else { + this.queue = new SynchronousQueue>(true); + } } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java index aa8c4e00df..71562af441 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java @@ -47,6 +47,8 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter private ChannelRegistry channelRegistry; + private long sendTimeout = -1; + public SplitterMessageHandlerAdapter(Object object, Method method, Map attributes) { Assert.notNull(object, "'object' must not be null"); @@ -63,6 +65,10 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter this.channelRegistry = channelRegistry; } + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + @Override protected Object doHandle(Message message, SimpleMethodInvoker invoker) { if (method.getParameterTypes().length != 1) { @@ -139,7 +145,7 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter if (logger.isDebugEnabled()) { logger.debug("sending message to channel '" + channelName + "'"); } - return channel.send(message); + return (this.sendTimeout < 0) ? channel.send(message) : channel.send(message, this.sendTimeout); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java index b85f252508..05d3cdd68b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java @@ -76,9 +76,9 @@ public class ByteStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - channel.send(new GenericMessage(new byte[] {4,5,6})); - channel.send(new GenericMessage(new byte[] {7,8,9})); + channel.send(new GenericMessage(new byte[] {1,2,3}), 0); + channel.send(new GenericMessage(new byte[] {4,5,6}), 0); + channel.send(new GenericMessage(new byte[] {7,8,9}), 0); assertEquals(3, dispatcherTask.dispatch()); byte[] result = stream.toByteArray(); assertEquals(9, result.length); @@ -95,9 +95,9 @@ public class ByteStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - channel.send(new GenericMessage(new byte[] {4,5,6})); - channel.send(new GenericMessage(new byte[] {7,8,9})); + channel.send(new GenericMessage(new byte[] {1,2,3}), 0); + channel.send(new GenericMessage(new byte[] {4,5,6}), 0); + channel.send(new GenericMessage(new byte[] {7,8,9}), 0); assertEquals(2, dispatcherTask.dispatch()); byte[] result = stream.toByteArray(); assertEquals(6, result.length); @@ -114,9 +114,9 @@ public class ByteStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - channel.send(new GenericMessage(new byte[] {4,5,6})); - channel.send(new GenericMessage(new byte[] {7,8,9})); + channel.send(new GenericMessage(new byte[] {1,2,3}), 0); + channel.send(new GenericMessage(new byte[] {4,5,6}), 0); + channel.send(new GenericMessage(new byte[] {7,8,9}), 0); assertEquals(3, dispatcherTask.dispatch()); byte[] result = stream.toByteArray(); assertEquals(9, result.length); @@ -133,9 +133,9 @@ public class ByteStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - channel.send(new GenericMessage(new byte[] {4,5,6})); - channel.send(new GenericMessage(new byte[] {7,8,9})); + channel.send(new GenericMessage(new byte[] {1,2,3}), 0); + channel.send(new GenericMessage(new byte[] {4,5,6}), 0); + channel.send(new GenericMessage(new byte[] {7,8,9}), 0); assertEquals(2, dispatcherTask.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); @@ -157,9 +157,9 @@ public class ByteStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - channel.send(new GenericMessage(new byte[] {4,5,6})); - channel.send(new GenericMessage(new byte[] {7,8,9})); + channel.send(new GenericMessage(new byte[] {1,2,3}), 0); + channel.send(new GenericMessage(new byte[] {4,5,6}), 0); + channel.send(new GenericMessage(new byte[] {7,8,9}), 0); assertEquals(3, dispatcherTask.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(9, result1.length); @@ -180,9 +180,9 @@ public class ByteStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - channel.send(new GenericMessage(new byte[] {4,5,6})); - channel.send(new GenericMessage(new byte[] {7,8,9})); + channel.send(new GenericMessage(new byte[] {1,2,3}), 0); + channel.send(new GenericMessage(new byte[] {4,5,6}), 0); + channel.send(new GenericMessage(new byte[] {7,8,9}), 0); assertEquals(2, dispatcherTask.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); @@ -203,9 +203,9 @@ public class ByteStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - channel.send(new GenericMessage(new byte[] {4,5,6})); - channel.send(new GenericMessage(new byte[] {7,8,9})); + channel.send(new GenericMessage(new byte[] {1,2,3}), 0); + channel.send(new GenericMessage(new byte[] {4,5,6}), 0); + channel.send(new GenericMessage(new byte[] {7,8,9}), 0); assertEquals(2, dispatcherTask.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index dbc798b1d9..5449db7c90 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -55,8 +55,8 @@ public class CharacterStreamTargetAdapterTests { CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new StringMessage("foo")); - channel.send(new StringMessage("bar")); + channel.send(new StringMessage("foo"), 0); + channel.send(new StringMessage("bar"), 0); assertEquals(1, dispatcherTask.dispatch()); String result1 = new String(stream.toByteArray()); assertEquals("foo", result1); @@ -73,8 +73,8 @@ public class CharacterStreamTargetAdapterTests { adapter.setShouldAppendNewLine(true); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new StringMessage("foo")); - channel.send(new StringMessage("bar")); + channel.send(new StringMessage("foo"), 0); + channel.send(new StringMessage("bar"), 0); assertEquals(1, dispatcherTask.dispatch()); String result1 = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); @@ -93,8 +93,8 @@ public class CharacterStreamTargetAdapterTests { SimpleChannel channel = new SimpleChannel(dispatcherPolicy); DispatcherTask dispatcherTask = new DispatcherTask(channel); dispatcherTask.addHandler(adapter); - channel.send(new StringMessage("foo")); - channel.send(new StringMessage("bar")); + channel.send(new StringMessage("foo"), 0); + channel.send(new StringMessage("bar"), 0); assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); assertEquals("foobar", result); @@ -111,8 +111,8 @@ public class CharacterStreamTargetAdapterTests { DispatcherTask dispatcherTask = new DispatcherTask(channel); adapter.setShouldAppendNewLine(true); dispatcherTask.addHandler(adapter); - channel.send(new StringMessage("foo")); - channel.send(new StringMessage("bar")); + channel.send(new StringMessage("foo"), 0); + channel.send(new StringMessage("bar"), 0); assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); @@ -146,8 +146,8 @@ public class CharacterStreamTargetAdapterTests { dispatcherTask.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); - channel.send(new GenericMessage(testObject1)); - channel.send(new GenericMessage(testObject2)); + channel.send(new GenericMessage(testObject1), 0); + channel.send(new GenericMessage(testObject2), 0); assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); assertEquals("foobar", result); @@ -166,8 +166,8 @@ public class CharacterStreamTargetAdapterTests { dispatcherTask.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); - channel.send(new GenericMessage(testObject1)); - channel.send(new GenericMessage(testObject2)); + channel.send(new GenericMessage(testObject1), 0); + channel.send(new GenericMessage(testObject2), 0); assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java index 30cdc53d2a..1973ea4a58 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetrieverTests.java @@ -42,8 +42,8 @@ public class ChannelPollingMessageRetrieverTests { ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); Collection> results = retriever.retrieveMessages(); assertTrue(results.isEmpty()); - channel.send(new StringMessage("test1")); - channel.send(new StringMessage("test2")); + channel.send(new StringMessage("test1"), 0); + channel.send(new StringMessage("test2"), 0); results = retriever.retrieveMessages(); assertEquals(1, results.size()); assertEquals("test1", results.iterator().next().getPayload()); @@ -61,9 +61,9 @@ public class ChannelPollingMessageRetrieverTests { ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); Collection> results = retriever.retrieveMessages(); assertTrue(results.isEmpty()); - channel.send(new StringMessage("test1")); - channel.send(new StringMessage("test2")); - channel.send(new StringMessage("test3")); + channel.send(new StringMessage("test1"), 0); + channel.send(new StringMessage("test2"), 0); + channel.send(new StringMessage("test3"), 0); results = retriever.retrieveMessages(); assertEquals(2, results.size()); Iterator> iter = results.iterator(); @@ -83,9 +83,9 @@ public class ChannelPollingMessageRetrieverTests { ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); Collection> results = retriever.retrieveMessages(); assertTrue(results.isEmpty()); - channel.send(new StringMessage("test1")); - channel.send(new StringMessage("test2")); - channel.send(new StringMessage("test3")); + channel.send(new StringMessage("test1"), 0); + channel.send(new StringMessage("test2"), 0); + channel.send(new StringMessage("test3"), 0); results = retriever.retrieveMessages(); assertEquals(1, results.size()); assertEquals("test1", results.iterator().next().getPayload());