diff --git a/org.springframework.integration.ip/.classpath b/org.springframework.integration.ip/.classpath index 6629c34424..ddc9103c5c 100644 --- a/org.springframework.integration.ip/.classpath +++ b/org.springframework.integration.ip/.classpath @@ -1,25 +1,10 @@ - - + + + - - - - - - - - - - - - - - - - - + diff --git a/org.springframework.integration.ip/.project b/org.springframework.integration.ip/.project index 8b42dbdf68..ed78e2585a 100644 --- a/org.springframework.integration.ip/.project +++ b/org.springframework.integration.ip/.project @@ -1,6 +1,6 @@ - org.springframework.integration.ip + spring-integration-ip @@ -10,8 +10,14 @@ + + org.maven.ide.eclipse.maven2Builder + + + + org.maven.ide.eclipse.maven2Nature org.eclipse.jdt.core.javanature diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index df43440a61..35df182ca7 100644 --- a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -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> messages) { + // TODO Auto-generated method stub + + } + + public void deleteAll(Object correlationId) { + // TODO Auto-generated method stub + + } + } diff --git a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index 9dac0d85cd..d6c9ca3f53 100644 --- a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -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 message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); + Message message = MessageBuilder.withPayload("foo").build(); Message 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 message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); + Message 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 message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); + String correlationId = "X"; + Message 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 message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); + Message message = MessageBuilder.withPayload("foo").build(); message = messageStore.put(message); assertNotNull(messageStore.delete(message.getHeaders().getId())); } diff --git a/org.springframework.integration.test/src/test/java/org/springframework/integration/test/matcher/HeaderMatcherTests.java b/org.springframework.integration.test/src/test/java/org/springframework/integration/test/matcher/HeaderMatcherTests.java index 4e52d47c57..5330029f42 100644 --- a/org.springframework.integration.test/src/test/java/org/springframework/integration/test/matcher/HeaderMatcherTests.java +++ b/org.springframework.integration.test/src/test/java/org/springframework/integration/test/matcher/HeaderMatcherTests.java @@ -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)); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 5793577ddc..aebdcb75d6 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -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> all = store.list(key); + Collection> 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); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java index 9eae40b5f2..08079e06a6 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java @@ -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> list(Object correlationId); + Collection> list(Object correlationId); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/util/UUIDConverter.java b/org.springframework.integration/src/main/java/org/springframework/integration/util/UUIDConverter.java new file mode 100644 index 0000000000..cc27e7f44e --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/util/UUIDConverter.java @@ -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 { + + 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(); + + } + +}