From 5821ac23a24bbc02faf7b3990e3596d3fc5c76c4 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 4 Apr 2008 16:58:57 +0000 Subject: [PATCH] Added SynchronousChannel (INT-170). --- .../dispatcher/DefaultMessageDistributor.java | 4 +- .../dispatcher/SynchronousChannel.java | 91 +++++++++ .../dispatcher/SynchronousChannelTests.java | 179 ++++++++++++++++++ 3 files changed, 272 insertions(+), 2 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java index 53e77b6674..0d40c3f1a3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.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. @@ -78,7 +78,7 @@ public class DefaultMessageDistributor implements MessageDistributor { Iterator iter = targets.iterator(); if (!iter.hasNext()) { if (logger.isWarnEnabled()) { - logger.warn("dispatcher has no active handlers"); + logger.warn("no active handlers"); } return false; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java new file mode 100644 index 0000000000..ac9591f1fe --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java @@ -0,0 +1,91 @@ +/* + * 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. + * 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.dispatcher; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.integration.adapter.PollableSource; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.SimplePayloadMessageMapper; +import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.util.CollectionUtils; + +/** + * A channel that invokes any subscribed {@link MessageHandler handler(s)} in a + * sender's thread. If a {@link PollableSource} is provided, then that source + * will likewise be polled within a receiver's thread. If no source is provided, + * then receive() will always return null. + * + * @author Mark Fisher + */ +public class SynchronousChannel extends AbstractMessageChannel { + + private final MessageDistributor distributor; + + private volatile PollableSource source; + + + public SynchronousChannel() { + this(null); + } + + public SynchronousChannel(DispatcherPolicy dispatcherPolicy) { + super(dispatcherPolicy != null ? dispatcherPolicy : new DispatcherPolicy()); + this.distributor = new DefaultMessageDistributor(this.getDispatcherPolicy()); + } + + + public void setSource(PollableSource source) { + this.source = source; + } + + public void addHandler(MessageHandler handler) { + this.distributor.addHandler(handler); + } + + @Override + protected Message doReceive(long timeout) { + if (this.source != null) { + Collection results = this.source.poll(1); + if (!CollectionUtils.isEmpty(results)) { + Object result = results.iterator().next(); + return (result instanceof Message) ? (Message) result : + new SimplePayloadMessageMapper().toMessage(result); + } + } + return null; + } + + @Override + protected boolean doSend(Message message, long timeout) { + return this.distributor.distribute(message); + } + + public List> clear() { + return new ArrayList>(); + } + + public List> purge(MessageSelector selector) { + return new ArrayList>(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java new file mode 100644 index 0000000000..8c5e776cea --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java @@ -0,0 +1,179 @@ +/* + * 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. + * 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.dispatcher; + +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 java.util.Collection; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + +import org.springframework.integration.adapter.PollableSource; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class SynchronousChannelTests { + + private static final String HANDLER_THREAD = "handler-thread"; + + + @Test + public void testSend() { + SynchronousChannel channel = new SynchronousChannel(); + channel.addHandler(new ThreadNameSettingTestHandler()); + StringMessage message = new StringMessage("test"); + assertTrue(channel.send(message)); + String handlerThreadName = message.getHeader().getProperty(HANDLER_THREAD); + assertEquals(Thread.currentThread().getName(), handlerThreadName); + } + + @Test + public void testSendWithNoHandler() { + SynchronousChannel channel = new SynchronousChannel(); + StringMessage message = new StringMessage("test"); + assertFalse(channel.send(message)); + } + + @Test + public void testSendInSeparateThread() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + final SynchronousChannel channel = new SynchronousChannel(); + channel.addHandler(new ThreadNameSettingTestHandler(latch)); + final StringMessage message = new StringMessage("test"); + new Thread(new Runnable() { + public void run() { + channel.send(message); + } + }, "test-thread").start(); + latch.await(1000, TimeUnit.MILLISECONDS); + String handlerThreadName = message.getHeader().getProperty(HANDLER_THREAD); + assertEquals("test-thread", handlerThreadName); + } + + @Test + public void testReceive() { + SynchronousChannel channel = new SynchronousChannel(); + channel.setSource(new PollableSource() { + public Collection poll(int limit) { + return Collections.singleton("foo"); + } + }); + Message message = channel.receive(); + assertNotNull(message); + assertNotNull(message.getPayload()); + assertEquals(String.class, message.getPayload().getClass()); + assertEquals("foo", message.getPayload()); + } + + @Test + public void testReceiveWithMessageResult() { + SynchronousChannel channel = new SynchronousChannel(); + channel.setSource(new MessageReturningTestSource("foo")); + Message message = channel.receive(); + assertNotNull(message); + assertNotNull(message.getPayload()); + assertEquals(String.class, message.getPayload().getClass()); + assertEquals("foo", message.getPayload()); + String handlerThreadName = message.getHeader().getProperty(HANDLER_THREAD); + assertEquals(Thread.currentThread().getName(), handlerThreadName); + } + + @Test + public void testReceiveInSeparateThread() throws InterruptedException { + final SynchronousChannel channel = new SynchronousChannel(); + channel.setSource(new MessageReturningTestSource("foo")); + final SynchronousQueue> messageHolder = new SynchronousQueue>(); + new Thread(new Runnable() { + public void run() { + Message message = channel.receive(); + assertNotNull(message); + try { + messageHolder.put(message); + } + catch (InterruptedException e) { + // will fail after timeout below + } + } + }, "test-thread").start(); + Message message = messageHolder.poll(1000, TimeUnit.MILLISECONDS); + assertNotNull(message); + assertNotNull(message.getPayload()); + assertEquals(String.class, message.getPayload().getClass()); + assertEquals("foo", message.getPayload()); + String handlerThreadName = message.getHeader().getProperty(HANDLER_THREAD); + assertEquals("test-thread", handlerThreadName); + } + + @Test + public void testReceiveWithNoSource() { + SynchronousChannel channel = new SynchronousChannel(); + assertNull(channel.receive()); + } + + + private static class ThreadNameSettingTestHandler implements MessageHandler { + + private final CountDownLatch latch; + + + ThreadNameSettingTestHandler() { + this(null); + } + + ThreadNameSettingTestHandler(CountDownLatch latch) { + this.latch = latch; + } + + public Message handle(Message message) { + message.getHeader().setProperty(HANDLER_THREAD, Thread.currentThread().getName()); + if (this.latch != null) { + this.latch.countDown(); + } + return null; + } + } + + + private static class MessageReturningTestSource implements PollableSource { + + private final String messageText; + + + MessageReturningTestSource(String messageText) { + this.messageText = messageText; + } + + public Collection poll(int limit) { + StringMessage message = new StringMessage(messageText); + message.getHeader().setProperty(HANDLER_THREAD, Thread.currentThread().getName()); + return Collections.singleton(message); + } + } + +}