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

OPEN - issue INT-1063: MessageStore: correlation and grouping API

Browse Source 
https://jira.springsource.org/browse/INT-1063
This commit is contained in:
David Syer
2010-04-28 08:00:00 +00:00
parent d634a4a423
commit 548e81bb7c
8 changed files with 178 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
org.springframework.integration.ip/.classpath
View File

@@ -1,25 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
<classpathentry kind="src" path="src/main/java"/>
<classpathentry kind="src" path="src/main/resources"/>
<classpathentry kind="src" output="target/classes" path="src/main/java"/>
<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources"/>
<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry kind="src" path="src/test/resources"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.logging/1.1.1/com.springsource.org.apache.commons.logging-1.1.1.jar" sourcepath="/SI_IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.logging/1.1.1/com.springsource.org.apache.commons.logging-sources-1.1.1.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.junit/com.springsource.org.junit/4.7.0/com.springsource.org.junit-4.7.0.jar" sourcepath="/SI_IVY_CACHE/org.junit/com.springsource.org.junit/4.7.0/com.springsource.org.junit-sources-4.7.0.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.springframework/org.springframework.asm/3.0.2.RELEASE/org.springframework.asm-3.0.2.RELEASE.jar" sourcepath="/SI_IVY_CACHE/org.springframework/org.springframework.asm/3.0.2.RELEASE/org.springframework.asm-sources-3.0.2.RELEASE.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.springframework/org.springframework.beans/3.0.2.RELEASE/org.springframework.beans-3.0.2.RELEASE.jar" sourcepath="/SI_IVY_CACHE/org.springframework/org.springframework.beans/3.0.2.RELEASE/org.springframework.beans-sources-3.0.2.RELEASE.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.springframework/org.springframework.context/3.0.2.RELEASE/org.springframework.context-3.0.2.RELEASE.jar" sourcepath="/SI_IVY_CACHE/org.springframework/org.springframework.context/3.0.2.RELEASE/org.springframework.context-sources-3.0.2.RELEASE.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.springframework/org.springframework.core/3.0.2.RELEASE/org.springframework.core-3.0.2.RELEASE.jar" sourcepath="/SI_IVY_CACHE/org.springframework/org.springframework.core/3.0.2.RELEASE/org.springframework.core-sources-3.0.2.RELEASE.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.springframework/org.springframework.expression/3.0.2.RELEASE/org.springframework.expression-3.0.2.RELEASE.jar" sourcepath="/SI_IVY_CACHE/org.springframework/org.springframework.expression/3.0.2.RELEASE/org.springframework.expression-sources-3.0.2.RELEASE.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.springframework/org.springframework.test/3.0.2.RELEASE/org.springframework.test-3.0.2.RELEASE.jar" sourcepath="/SI_IVY_CACHE/org.springframework/org.springframework.test/3.0.2.RELEASE/org.springframework.test-sources-3.0.2.RELEASE.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.easymock/com.springsource.org.easymock/2.3.0/com.springsource.org.easymock-2.3.0.jar" sourcepath="/SI_IVY_CACHE/org.easymock/com.springsource.org.easymock/2.3.0/com.springsource.org.easymock-sources-2.3.0.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/net.sourceforge.cglib/com.springsource.net.sf.cglib/2.1.3/com.springsource.net.sf.cglib-2.1.3.jar" sourcepath="/SI_IVY_CACHE/net.sourceforge.cglib/com.springsource.net.sf.cglib/2.1.3/com.springsource.net.sf.cglib-sources-2.1.3.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.mockito/com.springsource.org.mockito/1.8.0/com.springsource.org.mockito-1.8.0.jar" sourcepath="/SI_IVY_CACHE/org.mockito/com.springsource.org.mockito/1.8.0/com.springsource.org.mockito-sources-1.8.0.jar"/>
<classpathentry kind="var" path="SI_IVY_CACHE/org.objenesis/com.springsource.org.objenesis/1.0.0/com.springsource.org.objenesis-1.0.0.jar" sourcepath="/SI_IVY_CACHE/org.objenesis/com.springsource.org.objenesis/1.0.0/com.springsource.org.objenesis-sources-1.0.0.jar"/>
<classpathentry combineaccessrules="false" kind="src" path="/org.springframework.integration"/>
<classpathentry combineaccessrules="false" kind="src" path="/org.springframework.integration.test"/>
<classpathentry combineaccessrules="false" kind="src" path="/org.springframework.integration.adapter"/>
<classpathentry combineaccessrules="false" kind="src" path="/org.springframework.integration.stream"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
<classpathentry kind="output" path="target/classes"/>
</classpath>

8
org.springframework.integration.ip/.project
View File

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
<name>org.springframework.integration.ip</name>
<name>spring-integration-ip</name>
<comment></comment>
<projects>
</projects>
@@ -10,8 +10,14 @@
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.maven.ide.eclipse.maven2Builder</name>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.maven.ide.eclipse.maven2Nature</nature>
<nature>org.eclipse.jdt.core.javanature</nature>
</natures>
</projectDescription>

45
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -1,13 +1,11 @@
package org.springframework.integration.jdbc;
import java.math.BigInteger;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
@@ -19,6 +17,7 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.jdbc.util.SerializationUtils;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementSetter;
@@ -221,30 +220,7 @@ public class JdbcMessageStore implements MessageStore {
}
private String getKey(Object input) {
if (input == null) {
return null;
}
if (input instanceof UUID) {
return input.toString();
}
if (input instanceof String && ((String) input).length() < 100) {
return (String) input;
}
MessageDigest digest;
try {
digest = MessageDigest.getInstance("MD5");
}
catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("MD5 algorithm not available. Fatal (should be in the JDK).");
}
byte[] bytes = digest.digest(SerializationUtils.serialize(input));
return String.format("%032x", new BigInteger(1, bytes));
return input==null ? null : UUIDConverter.getUUID(input).toString();
}
/**
@@ -265,4 +241,19 @@ public class JdbcMessageStore implements MessageStore {
}
public void add(Object correlationId, Message<?> message) {
// TODO Auto-generated method stub
}
public void add(Object correlationId, Collection<Message<?>> messages) {
// TODO Auto-generated method stub
}
public void deleteAll(Object correlationId) {
// TODO Auto-generated method stub
}
}

18
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -24,17 +24,17 @@ import org.springframework.transaction.annotation.Transactional;
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class JdbcMessageStoreTests {
@Autowired
private DataSource dataSource;
private JdbcMessageStore messageStore;
@Before
public void init() {
messageStore = new JdbcMessageStore(dataSource);
}
@Test
@Transactional
public void testGetNonExistent() throws Exception {
@@ -45,7 +45,7 @@ public class JdbcMessageStoreTests {
@Test
@Transactional
public void testAddAndGet() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId("X").build();
Message<String> message = MessageBuilder.withPayload("foo").build();
Message<String> saved = messageStore.put(message);
assertNull(messageStore.get(message.getHeaders().getId()));
Message<?> result = messageStore.get(saved.getHeaders().getId());
@@ -58,7 +58,8 @@ public class JdbcMessageStoreTests {
@Test
@Transactional
public void testAddAndUpdate() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId("X").build();
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(
"X").build();
message = messageStore.put(message);
message = MessageBuilder.fromMessage(message).setCorrelationId("Y").build();
message = messageStore.put(message);
@@ -89,15 +90,16 @@ public class JdbcMessageStoreTests {
@Test
@Transactional
public void testAddAndListByCorrelationId() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId("X").build();
String correlationId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build();
messageStore.put(message);
assertEquals(1, messageStore.list("X").size());
assertEquals(1, messageStore.list(correlationId).size());
}
@Test
@Transactional
public void testAddAndDelete() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId("X").build();
Message<String> message = MessageBuilder.withPayload("foo").build();
message = messageStore.put(message);
assertNotNull(messageStore.delete(message.getHeaders().getId()));
}

33
org.springframework.integration.test/src/test/java/org/springframework/integration/test/matcher/HeaderMatcherTests.java
View File

@@ -16,21 +16,34 @@
package org.springframework.integration.test.matcher;
import static org.hamcrest.CoreMatchers.anything;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasAllHeaders;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasCorrelationId;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasExpirationDate;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeaderKey;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasMessageId;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasSequenceNumber;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasSequenceSize;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasTimestamp;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.hamcrest.Matcher;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import static org.hamcrest.CoreMatchers.*;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.HeaderMatcher.*;
/**
* @author Alex Peters
* @author Iwein Fuld
@@ -146,7 +159,7 @@ public class HeaderMatcherTests {
@Test
public void hasCorrelationId_() throws Exception {
Object correlationId = message.getHeaders().getId();
UUID correlationId = message.getHeaders().getId();
message = MessageBuilder.withPayload("blabla").setCorrelationId(correlationId).build();
assertThat(message, hasCorrelationId(correlationId));
}

7
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.aggregator;
import java.util.List;
import java.util.Collection;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
@@ -30,7 +30,6 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessageChannelTemplate;
@@ -293,11 +292,11 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
protected final boolean forceComplete(Object key) {
try {
if (tracker.tryLockFor(key)) {
List<Message<?>> all = store.list(key);
Collection<Message<?>> all = store.list(key);
MessageGroup group = new MessageGroup(all, completionStrategy, key, deleteOrTrackCallback());
if (all.size() > 0) {
// last chance for normal completion
MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel);
MessageChannel outputChannel = resolveReplyChannel(all.iterator().next(), this.outputChannel);
boolean processed = false;
if (group.isComplete()) {
outputProcessor.processAndSend(group, channelTemplate, outputChannel);

4
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.store;
import java.util.List;
import java.util.Collection;
import java.util.UUID;
import org.springframework.integration.core.Message;
@@ -63,6 +63,6 @@ public interface MessageStore {
* provided correlationId header value.
* @see org.springframework.integration.core.MessageHeaders#getCorrelationId()
*/
List<Message<?>> list(Object correlationId);
Collection<Message<?>> list(Object correlationId);
}

111
org.springframework.integration/src/main/java/org/springframework/integration/util/UUIDConverter.java Normal file
View File

@@ -0,0 +1,111 @@
/*
* 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.util;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.UnsupportedEncodingException;
import java.util.UUID;
import org.springframework.core.convert.converter.Converter;
import org.springframework.util.ClassUtils;
/**
* Utility to help generate UUID instances from generic objects.
*
* @author Dave Syer
*
*/
public class UUIDConverter implements Converter<Object, UUID> {
public static final String DEFAULT_CHARSET = "UTF-8";
private final String charset;
public UUIDConverter() {
this(DEFAULT_CHARSET);
}
public UUIDConverter(String charset) {
this.charset = charset;
}
public UUID convert(Object source) {
return getUUID(source, charset);
}
public static UUID getUUID(Object input) {
return getUUID(input, DEFAULT_CHARSET);
}
public static UUID getUUID(Object input, String charset) {
if (input == null) {
return null;
}
if (input instanceof UUID) {
return (UUID) input;
}
if (input instanceof String) {
try {
return UUID.fromString((String) input);
}
catch (IllegalArgumentException e) {
try {
return UUID.nameUUIDFromBytes(((String) input).getBytes(charset));
}
catch (UnsupportedEncodingException ex) {
throw new IllegalStateException("Cannot convert String using charset=" + charset, ex);
}
}
}
if (ClassUtils.isPrimitiveOrWrapper(input.getClass())) {
try {
return UUID.nameUUIDFromBytes(input.toString().getBytes(charset));
}
catch (UnsupportedEncodingException e) {
throw new IllegalStateException("Cannot convert primitive using charset=" + charset, e);
}
}
byte[] bytes = serialize(input);
return UUID.nameUUIDFromBytes(bytes);
}
private static byte[] serialize(Object object) {
if (object == null) {
return null;
}
ByteArrayOutputStream stream = new ByteArrayOutputStream();
try {
new ObjectOutputStream(stream).writeObject(object);
}
catch (IOException e) {
throw new IllegalArgumentException("Could not serialize object of type: " + object.getClass(), e);
}
return stream.toByteArray();
}
}