diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java index e452efbd32..30317172c9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java @@ -16,9 +16,6 @@ package org.springframework.integration.channel.interceptor; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -39,40 +36,51 @@ import org.springframework.util.Assert; */ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle { - private final Log logger = LogFactory.getLog(this.getClass()); + private static final Log logger = LogFactory.getLog(WireTap.class); private final MessageTarget target; - private final List selectors = new CopyOnWriteArrayList(); + private volatile long timeout = 0; + + private final MessageSelector selector; private volatile boolean running = true; /** - * Create a new wire tap with no {@link MessageSelector MessageSelectors}. + * Create a new wire tap with no {@link MessageSelector}. * * @param target the MessageTarget to which intercepted messages will be sent */ public WireTap(MessageTarget target) { - Assert.notNull(target, "target must not be null"); - this.target = target; + this(target, null); } /** - * Create a new wire tap with {@link MessageSelector MessageSelectors}. + * Create a new wire tap with the provided {@link MessageSelector}. * * @param target the target to which intercepted messages will be sent - * @param selectors the list of selectors that must accept a message for it to + * @param selector the selector that must accept a message for it to * be sent to the intercepting target */ - public WireTap(MessageTarget target, List selectors) { - this(target); - if (selectors != null) { - this.selectors.addAll(selectors); - } + public WireTap(MessageTarget target, MessageSelector selector) { + Assert.notNull(target, "target must not be null"); + this.target = target; + this.selector = selector; } + /** + * Specify the timeout value for sending to the intercepting target. Note + * that this value will only apply if the target is a {@link BlockingTarget}. + * The default value is 0. + * + * @param timeout the timeout in milliseconds + */ + public void setTimeout(long timeout) { + this.timeout = timeout; + } + /** * Check whether the wire tap is currently running. */ @@ -94,30 +102,22 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle { this.running = false; } + /** + * Intercept the Message and, if accepted by the {@link MessageSelector}, + * send it to the secondary target. If this wire tap's {@link MessageSelector} is + * null, it will accept all messages. + */ @Override public Message preSend(Message message, MessageChannel channel) { - if (this.running && this.selectorsAccept(message)) { - boolean sent = (this.target instanceof BlockingTarget) ? - ((BlockingTarget) this.target).send(message, 0) : this.target.send(message); + if (this.running && (this.selector == null || this.selector.accept(message))) { + boolean sent = (this.target instanceof BlockingTarget) + ? ((BlockingTarget) this.target).send(message, this.timeout) + : this.target.send(message); if (!sent && logger.isWarnEnabled()) { - logger.warn("failed to send message to WireTap target '" + this.target + "'"); + logger.warn("failed to send message to WireTap target '" + this.target + "'"); } } return message; } - /** - * If this wire tap has any {@link MessageSelector MessageSelectors}, check - * whether they accept the current message. If any of them do not accept it, - * the message will not be sent to the intercepting target. - */ - private boolean selectorsAccept(Message message) { - for (MessageSelector selector : this.selectors) { - if (!selector.accept(message)) { - return false; - } - } - return true; - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java index 508f6d7118..2312c1e72e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java @@ -16,6 +16,7 @@ package org.springframework.integration.config; +import java.util.HashMap; import java.util.Map; /** @@ -25,7 +26,9 @@ public class ChannelInterceptorParser extends AbstractInterceptorParser { @Override protected Map getParserMap() { - return null; + Map parsers = new HashMap(); + parsers.put("wire-tap", new WireTapParser()); + return parsers; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/WireTapParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/WireTapParser.java new file mode 100644 index 0000000000..e3a99a48e0 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/WireTapParser.java @@ -0,0 +1,54 @@ +/* + * 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.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.channel.interceptor.WireTap; +import org.springframework.util.StringUtils; + +/** + * Parser for the <wire-tap> element. + * + * @author Mark Fisher + */ +public class WireTapParser implements BeanDefinitionRegisteringParser { + + public String parse(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(WireTap.class); + String targetRef = element.getAttribute("target"); + if (!StringUtils.hasText(targetRef)) { + throw new ConfigurationException("the 'target' attribute is required"); + } + builder.addConstructorArgReference(targetRef); + String selectorRef = element.getAttribute("selector"); + if (StringUtils.hasText(selectorRef)) { + builder.addConstructorArgReference(selectorRef); + } + String timeout = element.getAttribute("timeout"); + if (StringUtils.hasText(timeout)) { + builder.addPropertyValue("timeout", Long.parseLong(timeout)); + } + return BeanDefinitionReaderUtils.registerWithGeneratedName( + builder.getBeanDefinition(), parserContext.getRegistry()); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index b254b6156c..6f5deb8295 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -418,11 +418,23 @@ + + + + + Defines a Wire Tap Channel Interceptor. + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java index c3eb8548ca..4d3c54ee46 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java @@ -20,9 +20,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; -import java.util.ArrayList; -import java.util.List; - import org.junit.Test; import org.springframework.integration.channel.QueueChannel; @@ -38,7 +35,7 @@ import org.springframework.integration.message.selector.MessageSelector; public class WireTapTests { @Test - public void wireTapWithNoSelectors() { + public void wireTapWithNoSelector() { QueueChannel mainChannel = new QueueChannel(); QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); @@ -54,10 +51,7 @@ public class WireTapTests { public void wireTapWithRejectingSelector() { QueueChannel mainChannel = new QueueChannel(); QueueChannel secondaryChannel = new QueueChannel(); - List selectors = new ArrayList(); - selectors.add(new TestSelector(true)); - selectors.add(new TestSelector(false)); - mainChannel.addInterceptor(new WireTap(secondaryChannel, selectors)); + mainChannel.addInterceptor(new WireTap(secondaryChannel, new TestSelector(false))); mainChannel.send(new StringMessage("testing")); Message original = mainChannel.receive(0); assertNotNull(original); @@ -66,13 +60,10 @@ public class WireTapTests { } @Test - public void wireTapWithAcceptingSelectors() { + public void wireTapWithAcceptingSelector() { QueueChannel mainChannel = new QueueChannel(); QueueChannel secondaryChannel = new QueueChannel(); - List selectors = new ArrayList(); - selectors.add(new TestSelector(true)); - selectors.add(new TestSelector(true)); - mainChannel.addInterceptor(new WireTap(secondaryChannel, selectors)); + mainChannel.addInterceptor(new WireTap(secondaryChannel, new TestSelector(true))); mainChannel.send(new StringMessage("testing")); Message original = mainChannel.receive(0); assertNotNull(original); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestSelector.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestSelector.java new file mode 100644 index 0000000000..82909a5c89 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestSelector.java @@ -0,0 +1,37 @@ +/* + * 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.config; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * @author Mark Fisher + */ +public class TestSelector implements MessageSelector { + + private final boolean accept; + + public TestSelector(boolean accept) { + this.accept = accept; + } + + public boolean accept(Message message) { + return this.accept; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java new file mode 100644 index 0000000000..2adc154cd7 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java @@ -0,0 +1,106 @@ +/* + * 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.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.util.Map; + +import org.junit.Test; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.interceptor.WireTap; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class WireTapParserTests { + + @Test + public void simpleWireTap() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "wireTapParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("simple"); + WireTapTestTarget target = (WireTapTestTarget) context.getBean("testTarget"); + assertNull(target.getLastMessage()); + Message original = new StringMessage("test"); + channel.send(original); + Message intercepted = target.getLastMessage(); + assertNotNull(intercepted); + assertEquals(original, intercepted); + } + + @Test + public void wireTapWithAcceptingSelector() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "wireTapParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("accepting"); + WireTapTestTarget target = (WireTapTestTarget) context.getBean("testTarget"); + assertNull(target.getLastMessage()); + Message original = new StringMessage("test"); + channel.send(original); + Message intercepted = target.getLastMessage(); + assertNotNull(intercepted); + assertEquals(original, intercepted); + } + + @Test + public void wireTapWithRejectingSelector() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "wireTapParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("rejecting"); + WireTapTestTarget target = (WireTapTestTarget) context.getBean("testTarget"); + assertNull(target.getLastMessage()); + Message original = new StringMessage("test"); + channel.send(original); + Message intercepted = target.getLastMessage(); + assertNull(intercepted); + } + + @Test + @SuppressWarnings("unchecked") + public void wireTapTimeouts() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "wireTapParserTests.xml", this.getClass()); + Map beans = (Map) context.getBeansOfType(WireTap.class); + int defaultTimeoutCount = 0; + int expectedTimeoutCount = 0; + int otherTimeoutCount = 0; + for (WireTap wireTap : beans.values()) { + long timeout = ((Long) new DirectFieldAccessor(wireTap).getPropertyValue("timeout")).longValue(); + if (timeout == 0) { + defaultTimeoutCount++; + } + else if (timeout == 1234) { + expectedTimeoutCount++; + } + else { + otherTimeoutCount++; + } + } + assertEquals(3, defaultTimeoutCount); + assertEquals(1, expectedTimeoutCount); + assertEquals(0, otherTimeoutCount); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapTestTarget.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapTestTarget.java new file mode 100644 index 0000000000..63e99f4f44 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapTestTarget.java @@ -0,0 +1,39 @@ +/* + * 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.config; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; + +/** + * @author Mark Fisher + */ +public class WireTapTestTarget implements MessageTarget { + + private volatile Message lastMessage; + + + public Message getLastMessage() { + return this.lastMessage; + } + + public boolean send(Message message) { + this.lastMessage= message; + return true; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml new file mode 100644 index 0000000000..b2b3045d5a --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +