From 2fbf5da2c78752defc9c8f12ffed5bc762c2e3eb Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Mon, 14 Jun 2010 15:15:48 +0000 Subject: [PATCH] INT-1174: Added CorrelatingMessageBarrier and tests --- .../aggregator/CorrelatingMessageBarrier.java | 121 ++++++++++++++ .../CorrelatingMessageBarrierTest.java | 154 ++++++++++++++++++ 2 files changed, 275 insertions(+) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java new file mode 100644 index 0000000000..79fdb92258 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java @@ -0,0 +1,121 @@ +/* + * Copyright 2002-2010 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.aggregator; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.core.Message; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.SimpleMessageStore; + +import java.util.Iterator; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * This Endpoint serves as a barrier for messages that should not be processed yet. The decision when a message can be + * processed is delegated to a {@link org.springframework.integration.aggregator.ReleaseStrategy ReleaseStrategy}. + * When a message can be processed it is up to the client to take care of the locking (potentially from the ReleaseStrategy's + * {@link org.springframework.integration.aggregator.ReleaseStrategy#canRelease(org.springframework.integration.store.MessageGroup) canRelease(..)} + * method). + *

+ * The messages will be stored in a {@link org.springframework.integration.store.MessageGroupStore MessageStore} + * for each correlation key. + * + * @author Iwein Fuld + */ +public class CorrelatingMessageBarrier extends AbstractMessageHandler implements MessageSource { + private static final Log log = LogFactory.getLog(CorrelatingMessageBarrier.class); + + private CorrelationStrategy correlationStrategy; + private ReleaseStrategy releaseStrategy; + + private final ConcurrentMap correlationLocks = new ConcurrentHashMap(); + private final MessageGroupStore store; + + public CorrelatingMessageBarrier(MessageGroupStore store) { + this.store = store; + } + + public CorrelatingMessageBarrier() { + this(new SimpleMessageStore(0)); + } + + /** + * Set the CorrelationStrategy to be used to determine the correlation key for incoming messages + */ + public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) { + this.correlationStrategy = correlationStrategy; + } + + /** + * Set the ReleaseStrategy that should be used when deciding if a group in this barrier may be released. + */ + public void setReleaseStrategy(ReleaseStrategy releaseStrategy) { + this.releaseStrategy = releaseStrategy; + } + + @Override + protected void handleMessageInternal(Message message) throws Exception { + Object correlationKey = correlationStrategy.getCorrelationKey(message); + Object lock = getLock(correlationKey); + synchronized (lock) { + store.addMessageToGroup(correlationKey, message); + } + if (log.isDebugEnabled()) { + log.debug(String.format("Handled message for key [%s]: %s.", correlationKey, message)); + } + } + + private Object getLock(Object correlationKey) { + Object existingLock = correlationLocks.putIfAbsent(correlationKey, correlationKey); + return existingLock == null ? correlationKey : existingLock; + } + + + public Message receive() { + for (Object key : correlationLocks.keySet()) { + Object lock = getLock(key); + synchronized (lock) { + MessageGroup group = store.getMessageGroup(key); + //group might be removed by another thread + if (group != null) { + if (releaseStrategy.canRelease(group)) { + Message nextMessage = null; + + Iterator> unmarked = group.getUnmarked().iterator(); + if (unmarked.hasNext()) { + nextMessage = unmarked.next(); + store.markMessageInGroup(key, nextMessage); + if (log.isDebugEnabled()) { + log.debug(String.format("Released message for key [%s]: %s.", key, nextMessage)); + } + } else { + remove(key); + } + return nextMessage; + } + } + } + } + return null; + } + + private void remove(Object key) { + correlationLocks.remove(key); + store.removeMessageGroup(key); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java new file mode 100644 index 0000000000..2e42386f3c --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTest.java @@ -0,0 +1,154 @@ +/* + * Copyright 2002-2010 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.aggregator; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.store.MessageGroup; + +import java.util.concurrent.*; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.nullValue; +import static org.junit.Assert.assertThat; +import static org.mockito.Mockito.*; + +/** + * @author Iwein Fuld + */ +@RunWith(MockitoJUnitRunner.class) +public class CorrelatingMessageBarrierTest { + + private CorrelatingMessageBarrier barrier; + @Mock + private CorrelationStrategy correlationStrategy; + @Mock + private ReleaseStrategy releaseStrategy; + + @Before + public void initializeBarrier() { + barrier = new CorrelatingMessageBarrier(); + barrier.setCorrelationStrategy(correlationStrategy); + barrier.setReleaseStrategy(releaseStrategy); + + when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn("foo"); + when(releaseStrategy.canRelease(isA(MessageGroup.class))).thenReturn(true); + } + + @Test + public void shouldPassMessage() { + Message message = testMessage(); + barrier.handleMessage(message); + assertThat(barrier.receive(), is(message)); + } + + @Test + public void shouldRemoveKeyWithoutLockingOnEmptyQueue() throws InterruptedException { + Message message = testMessage(); + Message message2 = testMessage(); + barrier.handleMessage(message); + verify(correlationStrategy).getCorrelationKey(message); + assertThat(barrier.receive(), is(notNullValue())); + barrier.handleMessage(message2); + assertThat(barrier.receive(), is(notNullValue())); + assertThat(barrier.receive(), is(nullValue())); + } + + @Test(timeout = 10000) + public void shouldNotDropMessageOrBlockSendingThread() { + OneMessagePerKeyReleaseStrategy trackingReleaseStrategy = new OneMessagePerKeyReleaseStrategy(); + barrier.setReleaseStrategy(trackingReleaseStrategy); + final CountDownLatch start = new CountDownLatch(1); + final CountDownLatch sent = new CountDownLatch(200); + for (int i = 0; i < 200; i++) { + sendAsynchronously(barrier, testMessage(), start, sent); + } + start.countDown(); + + try { + sent.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + assertThat((barrier.receive()), is(notNullValue())); + for (int i = 0; i < 199; i++) { + trackingReleaseStrategy.release("foo"); + assertThat((barrier.receive()), is(notNullValue())); + } + } + + private void sendAsynchronously(final MessageHandler handler, final Message message, final CountDownLatch start, final CountDownLatch sent) { + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + start.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + handler.handleMessage(message); + sent.countDown(); + } + }); + + } + + private Message testMessage() { + return MessageBuilder.withPayload("payload").build(); + } + + + /** + * ReleaseStrategy that emulates the use case described in INT-1068 + */ + private static class OneMessagePerKeyReleaseStrategy implements ReleaseStrategy { + private final ConcurrentMap keyLocks = new ConcurrentHashMap(); + + public boolean canRelease(MessageGroup messageGroup) { + System.out.println("Trying to release group: " + messageGroup + "\n to thread: " + Thread.currentThread()); + Object correlationKey = messageGroup.getCorrelationKey(); + Semaphore lock = lockForKey(correlationKey); + System.out.println(Thread.currentThread() + " got lock: " + lock); + return lock.tryAcquire(); + } + + private Semaphore lockForKey(Object correlationKey) { + Semaphore semaphore = keyLocks.get(correlationKey); + if (semaphore == null) { + keyLocks.putIfAbsent(correlationKey, new Semaphore(1)); + semaphore = keyLocks.get(correlationKey); + } + return semaphore; + } + + public void release(String correlationKey) { + Semaphore lock = keyLocks.get(correlationKey); + if (lock != null) { + lock.release(); + } + } + + public void releaseAll() { + for (Semaphore semaphore : keyLocks.values()) { + semaphore.release(); + } + } + } +}