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

Added WireTap (INT-138).

Browse Source
This commit is contained in:
Mark Fisher
2008-03-03 21:56:59 +00:00
parent e2929b4d7e
commit 1e19a96a38
3 changed files with 304 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java Normal file
View File

@@ -0,0 +1,129 @@
/*
* Copyright 2002-2007 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.channel.interceptor;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.util.Assert;
/**
* A {@link ChannelInterceptor} that publishes a copy of the intercepted message
* to a secondary channel while still sending the original message to the main channel.
*
* @author Mark Fisher
*/
public class WireTap extends ChannelInterceptorAdapter implements Lifecycle {
/** key for the attribute containing the original Message's id */
public final static String ORIGINAL_MESSAGE_ID_KEY = "_wireTap.originalMessageId";
private final Log logger = LogFactory.getLog(this.getClass());
private final MessageChannel secondaryChannel;
private final List<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
private volatile boolean running = true;
/**
* Create a new wire tap with <em>no</em> {@link MessageSelector MessageSelectors}.
*
* @param secondaryChannel the channel to which duplicate messages will be sent
*/
public WireTap(MessageChannel secondaryChannel) {
Assert.notNull(secondaryChannel, "'secondaryChannel' must not be null");
this.secondaryChannel = secondaryChannel;
}
/**
* Create a new wire tap with {@link MessageSelector MessageSelectors}.
*
* @param secondaryChannel the channel to which duplicate messages will be sent
* @param selectors the list of selectors that must accept a message for it to
* be sent to the secondary channel
*/
public WireTap(MessageChannel secondaryChannel, List<MessageSelector> selectors) {
this(secondaryChannel);
if (selectors != null) {
this.selectors.addAll(selectors);
}
}
/**
* Check whether the wire tap is currently running.
*/
public boolean isRunning() {
return this.running;
}
/**
* Restart the wire tap if it has been stopped. It is running by default.
*/
public void start() {
this.running = true;
}
/**
* Stop the wire tap. To restart, invoke {@link #start()}.
*/
public void stop() {
this.running = false;
}
@Override
public boolean preSend(Message<?> message, MessageChannel channel) {
if (this.running && this.selectorsAccept(message)) {
Message<?> duplicate = new GenericMessage(message.getPayload(), message.getHeader());
duplicate.getHeader().setAttribute(ORIGINAL_MESSAGE_ID_KEY, message.getId());
if (!this.secondaryChannel.send(duplicate, 0)) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to send message to secondary channel '" + this.secondaryChannel.getName()
+ "'. Check its capacity and whether it has any subscribers.");
}
}
}
return true;
}
/**
* 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 secondary channel.
*/
private boolean selectorsAccept(Message<?> message) {
for (MessageSelector selector : this.selectors) {
if (!selector.accept(message)) {
return false;
}
}
return true;
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java
View File

@@ -149,7 +149,8 @@ public class MessageHeader {
}
public String toString() {
return "[Properties=" + this.properties + "][Attributes=" + this.attributes + "]";
return "[Properties=" + this.properties + "][Attributes=" + this.attributes +
"][Timestamp=" + this.timestamp + "]";
}
}

173
spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java Normal file
View File

@@ -0,0 +1,173 @@
/*
* Copyright 2002-2007 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.channel.interceptor;
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.ArrayList;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.MessageSelector;
/**
* @author Mark Fisher
*/
public class WireTapTests {
@Test
public void testWireTapWithNoSelectors() {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
mainChannel.addInterceptor(new WireTap(secondaryChannel));
mainChannel.send(new StringMessage("testing"));
Message<?> original = mainChannel.receive(0);
assertNotNull(original);
Message<?> duplicate = secondaryChannel.receive(0);
assertNotNull(duplicate);
}
@Test
public void testWireTapWithRejectingSelector() {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
List<MessageSelector> selectors = new ArrayList<MessageSelector>();
selectors.add(new TestSelector(true));
selectors.add(new TestSelector(false));
mainChannel.addInterceptor(new WireTap(secondaryChannel, selectors));
mainChannel.send(new StringMessage("testing"));
Message<?> original = mainChannel.receive(0);
assertNotNull(original);
Message<?> duplicate = secondaryChannel.receive(0);
assertNull(duplicate);
}
@Test
public void testWireTapWithAcceptingSelectors() {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
List<MessageSelector> selectors = new ArrayList<MessageSelector>();
selectors.add(new TestSelector(true));
selectors.add(new TestSelector(true));
mainChannel.addInterceptor(new WireTap(secondaryChannel, selectors));
mainChannel.send(new StringMessage("testing"));
Message<?> original = mainChannel.receive(0);
assertNotNull(original);
Message<?> duplicate = secondaryChannel.receive(0);
assertNotNull(duplicate);
}
@Test
public void testNewMessageIdGeneratedForDuplicate() {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
mainChannel.addInterceptor(new WireTap(secondaryChannel));
mainChannel.send(new StringMessage("testing"));
Message<?> original = mainChannel.receive(0);
Message<?> duplicate = secondaryChannel.receive(0);
Object duplicateId = duplicate.getId();
assertNotNull(duplicateId);
assertFalse("message ids should not match", original.getId().equals(duplicateId));
}
@Test
public void testOriginalIdStoredAsAttribute() {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
mainChannel.addInterceptor(new WireTap(secondaryChannel));
mainChannel.send(new StringMessage("testing"));
Message<?> original = mainChannel.receive(0);
Message<?> duplicate = secondaryChannel.receive(0);
Object originalIdAttribute = duplicate.getHeader().getAttribute(WireTap.ORIGINAL_MESSAGE_ID_KEY);
assertNotNull(originalIdAttribute);
assertEquals(original.getId(), originalIdAttribute);
}
@Test
public void testNewTimestampGeneratedForDuplicate() throws InterruptedException {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
mainChannel.addInterceptor(new WireTap(secondaryChannel));
Message<?> message = new StringMessage("testing");
Thread.sleep(3);
mainChannel.send(message);
Message<?> original = mainChannel.receive(0);
Message<?> duplicate = secondaryChannel.receive(0);
assertTrue("original timestamp should precede duplicate",
original.getHeader().getTimestamp().before(duplicate.getHeader().getTimestamp()));
}
public void testDuplicateMessageContainsAttribute() {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
mainChannel.addInterceptor(new WireTap(secondaryChannel));
Message<?> message = new StringMessage("testing");
String attributeKey = "testAttribute";
Integer attributeValue = new Integer(123);
message.getHeader().setAttribute(attributeKey, attributeValue);
mainChannel.send(message);
Message<?> original = mainChannel.receive(0);
Message<?> duplicate = secondaryChannel.receive(0);
Object originalAttribute = original.getHeader().getAttribute(attributeKey);
Object duplicateAttribute = duplicate.getHeader().getAttribute(attributeKey);
assertNotNull(originalAttribute);
assertNotNull(duplicateAttribute);
assertEquals(originalAttribute, duplicateAttribute);
}
@Test
public void testDuplicateMessageContainsProperty() {
SimpleChannel mainChannel = new SimpleChannel();
SimpleChannel secondaryChannel = new SimpleChannel();
mainChannel.addInterceptor(new WireTap(secondaryChannel));
Message<?> message = new StringMessage("testing");
String propertyKey = "testProperty";
String propertyValue = "foo";
message.getHeader().setProperty(propertyKey, propertyValue);
mainChannel.send(message);
Message<?> original = mainChannel.receive(0);
Message<?> duplicate = secondaryChannel.receive(0);
String originalProperty = original.getHeader().getProperty(propertyKey);
String duplicateProperty = duplicate.getHeader().getProperty(propertyKey);
assertNotNull(originalProperty);
assertNotNull(duplicateProperty);
assertEquals(originalProperty, duplicateProperty);
}
private static class TestSelector implements MessageSelector {
private boolean shouldAccept;
public TestSelector(boolean shouldAccept) {
this.shouldAccept = shouldAccept;
}
public boolean accept(Message<?> message) {
return this.shouldAccept;
}
}
}