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

Added namespace support for the WireTap (INT-323).

Browse Source
This commit is contained in:
Mark Fisher
2008-08-12 22:49:27 +00:00
parent 1e8fa31e9d
commit 744db9e5a1
9 changed files with 333 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java
View File

@@ -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<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
private volatile long timeout = 0;
private final MessageSelector selector;
private volatile boolean running = true;
/**
* Create a new wire tap with <em>no</em> {@link MessageSelector MessageSelectors}.
* Create a new wire tap with <em>no</em> {@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<MessageSelector> 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, <em>if accepted</em> by the {@link MessageSelector},
* send it to the secondary target. If this wire tap's {@link MessageSelector} is
* <code>null</code>, 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 <em>not</em> be sent to the intercepting target.
*/
private boolean selectorsAccept(Message<?> message) {
for (MessageSelector selector : this.selectors) {
if (!selector.accept(message)) {
return false;
}
}
return true;
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java
View File

@@ -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<String, BeanDefinitionRegisteringParser> getParserMap() {
return null;
Map<String, BeanDefinitionRegisteringParser> parsers = new HashMap<String, BeanDefinitionRegisteringParser>();
parsers.put("wire-tap", new WireTapParser());
return parsers;
}
}

54
org.springframework.integration/src/main/java/org/springframework/integration/config/WireTapParser.java Normal file
View File

@@ -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 &lt;wire-tap&gt; 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());
}
}

12
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -418,11 +418,23 @@
<xsd:attribute name="bean" use="required"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="wire-tap" type="wireTapType" minOccurs="0" maxOccurs="unbounded"/>
<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded"/>
</xsd:choice>
</xsd:sequence>
</xsd:complexType>
<xsd:complexType name="wireTapType">
<xsd:annotation>
<xsd:documentation>
Defines a Wire Tap Channel Interceptor.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="target" type="xsd:string" use="required"/>
<xsd:attribute name="selector" type="xsd:string" use="optional"/>
<xsd:attribute name="timeout" type="xsd:long" use="optional"/>
</xsd:complexType>
<xsd:complexType name="endpointInterceptorsType">
<xsd:annotation>
<xsd:documentation>

17
org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java
View File

@@ -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<MessageSelector> selectors = new ArrayList<MessageSelector>();
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<MessageSelector> selectors = new ArrayList<MessageSelector>();
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);

37
org.springframework.integration/src/test/java/org/springframework/integration/config/TestSelector.java Normal file
View File

@@ -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;
}
}

106
org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapParserTests.java Normal file
View File

@@ -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<String, WireTap> beans = (Map<String, WireTap>) 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);
}
}

39
org.springframework.integration/src/test/java/org/springframework/integration/config/WireTapTestTarget.java Normal file
View File

@@ -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;
}
}

44
org.springframework.integration/src/test/java/org/springframework/integration/config/wireTapParserTests.xml Normal file
View File

@@ -0,0 +1,44 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<channel id="simple">
<interceptors>
<wire-tap target="testTarget"/>
</interceptors>
</channel>
<channel id="accepting">
<interceptors>
<wire-tap target="testTarget" selector="acceptingSelector"/>
</interceptors>
</channel>
<channel id="rejecting">
<interceptors>
<wire-tap target="testTarget" selector="rejectingSelector"/>
</interceptors>
</channel>
<channel id="timeout">
<interceptors>
<wire-tap target="testTarget" timeout="1234"/>
</interceptors>
</channel>
<beans:bean id="testTarget" class="org.springframework.integration.config.WireTapTestTarget"/>
<beans:bean id="acceptingSelector" class="org.springframework.integration.config.TestSelector">
<beans:constructor-arg value="true"/>
</beans:bean>
<beans:bean id="rejectingSelector" class="org.springframework.integration.config.TestSelector">
<beans:constructor-arg value="false"/>
</beans:bean>
</beans:beans>