From e3f7bf46816ffb1b5f30c0347e9e78bcd3eee011 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 31 Jan 2008 18:30:21 +0000 Subject: [PATCH] Added 'clear()' and 'purge()' methods to channel and added 'isExpired()' method to Message. --- .../integration/channel/MessageChannel.java | 12 ++++ .../integration/channel/SimpleChannel.java | 22 +++++++ .../integration/message/GenericMessage.java | 7 +++ .../integration/message/Message.java | 2 + .../channel/SimpleChannelTests.java | 50 +++++++++++++++- .../message/GenericMessageTests.java | 58 +++++++++++++++++++ 6 files changed, 150 insertions(+), 1 deletion(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/message/GenericMessageTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index ef46200346..5f0688cfe3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -16,6 +16,8 @@ package org.springframework.integration.channel; +import java.util.List; + import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.message.Message; @@ -83,4 +85,14 @@ public interface MessageChannel { */ Message receive(long timeout); + /** + * Remove all {@link Message Messages} from this channel. + */ + List clear(); + + /** + * Remove any expired {@link Message Messages} from this channel. + */ + List purge(); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index 98d3c8b0b8..ca5b2232ba 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -16,6 +16,8 @@ package org.springframework.integration.channel; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; @@ -103,4 +105,24 @@ public class SimpleChannel extends AbstractMessageChannel { } } + public List clear() { + List clearedMessages = new ArrayList(); + this.queue.drainTo(clearedMessages); + return clearedMessages; + } + + public List purge() { + List purgedMessages = new ArrayList(); + // take a snapshot + Object[] array = this.queue.toArray(); + for (Object o : array) { + Message message = (Message) o; + if (message.isExpired() && this.queue.remove(message)) { + // message was still in the queue + purgedMessages.add(message); + } + } + return purgedMessages; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java b/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java index 3dadf39a90..e790fba3f8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java @@ -16,6 +16,8 @@ package org.springframework.integration.message; +import java.util.Date; + import org.springframework.integration.util.RandomGuidUidGenerator; import org.springframework.integration.util.UidGenerator; import org.springframework.util.Assert; @@ -82,4 +84,9 @@ public class GenericMessage implements Message { this.payload = newPayload; } + public boolean isExpired() { + Date expiration = this.header.getExpiration(); + return (expiration != null) ? expiration.getTime() < System.currentTimeMillis() : false; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java b/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java index 7294978662..5d83ded23f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java @@ -29,4 +29,6 @@ public interface Message { T getPayload(); + boolean isExpired(); + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java index 0ddbf51468..38782d927b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java @@ -16,9 +16,13 @@ package org.springframework.integration.channel; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import java.util.Date; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.Executors; @@ -27,8 +31,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; -import org.springframework.integration.message.Message; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher @@ -189,4 +194,47 @@ public class SimpleChannelTests { assertTrue(sendInterrupted.get()); } + @Test + public void testClear() { + SimpleChannel channel = new SimpleChannel(2); + StringMessage message1 = new StringMessage("test1"); + StringMessage message2 = new StringMessage("test2"); + StringMessage message3 = new StringMessage("test3"); + assertTrue(channel.send(message1)); + assertTrue(channel.send(message2)); + assertFalse(channel.send(message3, 0)); + List clearedMessages = channel.clear(); + assertNotNull(clearedMessages); + assertEquals(2, clearedMessages.size()); + assertTrue(channel.send(message3)); + } + + @Test + public void testClearEmptyChannel() { + SimpleChannel channel = new SimpleChannel(); + List clearedMessages = channel.clear(); + assertNotNull(clearedMessages); + assertEquals(0, clearedMessages.size()); + } + + @Test + public void testPurge() { + SimpleChannel channel = new SimpleChannel(2); + long minute = 60 * 1000; + long time = System.currentTimeMillis(); + Date past = new Date(time - minute); + Date future = new Date(time + minute); + StringMessage expiredMessage = new StringMessage("test1"); + expiredMessage.getHeader().setExpiration(past); + StringMessage unexpiredMessage = new StringMessage("test2"); + unexpiredMessage.getHeader().setExpiration(future); + assertTrue(channel.send(expiredMessage, 0)); + assertTrue(channel.send(unexpiredMessage, 0)); + assertFalse(channel.send(new StringMessage("atCapacity"), 0)); + List purgedMessages = channel.purge(); + assertNotNull(purgedMessages); + assertEquals(1, purgedMessages.size()); + assertTrue(channel.send(new StringMessage("roomAvailable"), 0)); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/GenericMessageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/GenericMessageTests.java new file mode 100644 index 0000000000..cb44669a8e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/GenericMessageTests.java @@ -0,0 +1,58 @@ +/* + * 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.message; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.Date; + +import org.junit.Test; + +/** + * @author Mark Fisher + */ +public class GenericMessageTests { + + private static final long ONE_MINUTE = 60 * 1000; + + + @Test + public void testExpiredMessage() { + GenericMessage expiredMessage = new GenericMessage(1); + Date past = new Date(System.currentTimeMillis() - ONE_MINUTE); + expiredMessage.getHeader().setExpiration(past); + assertTrue(expiredMessage.isExpired()); + } + + @Test + public void testUnexpiredMessage() { + GenericMessage unexpiredMessage = new GenericMessage(1); + Date future = new Date(System.currentTimeMillis() + ONE_MINUTE); + unexpiredMessage.getHeader().setExpiration(future); + assertFalse(unexpiredMessage.isExpired()); + } + + @Test + public void testMessageWithNullExpirationNeverExpires() { + GenericMessage message = new GenericMessage(1); + assertNull(message.getHeader().getExpiration()); + assertFalse(message.isExpired()); + } + +}