From 1e19a96a38fd8fbb0520b1e9368a887bea2c8b91 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 3 Mar 2008 21:56:59 +0000 Subject: [PATCH] Added WireTap (INT-138). --- .../channel/interceptor/WireTap.java | 129 +++++++++++++ .../integration/message/MessageHeader.java | 3 +- .../channel/interceptor/WireTapTests.java | 173 ++++++++++++++++++ 3 files changed, 304 insertions(+), 1 deletion(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java new file mode 100644 index 0000000000..dad074d193 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java @@ -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 selectors = new CopyOnWriteArrayList(); + + private volatile boolean running = true; + + + /** + * Create a new wire tap with no {@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 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 not be sent to the secondary channel. + */ + private boolean selectorsAccept(Message message) { + for (MessageSelector selector : this.selectors) { + if (!selector.accept(message)) { + return false; + } + } + return true; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java index 020effcb8e..28bcc2041c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java @@ -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 + "]"; } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java new file mode 100644 index 0000000000..7c172a537a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java @@ -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 selectors = new ArrayList(); + 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 selectors = new ArrayList(); + 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; + } + } + +}