diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java index e267a5ceef..93dbfea660 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.gemfire.store; import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -33,7 +34,9 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.gemfire.CacheFactoryBean; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroup; @@ -206,6 +209,33 @@ public class GemfireGroupStoreTests { assertEquals(1, messageGroup.getMessages().size()); } + @Test + public void testWithMessageHistory() throws Exception{ + GemfireMessageStore store = new GemfireMessageStore(this.cache); + store.afterPropertiesSet(); + + store.getMessageGroup(1); + + Message message = new GenericMessage("Hello"); + DirectChannel fooChannel = new DirectChannel(); + fooChannel.setBeanName("fooChannel"); + DirectChannel barChannel = new DirectChannel(); + barChannel.setBeanName("barChannel"); + + message = MessageHistory.write(message, fooChannel); + message = MessageHistory.write(message, barChannel); + store.addMessageToGroup(1, message); + + message = store.getMessageGroup(1).getMessages().iterator().next(); + + MessageHistory messageHistory = MessageHistory.read(message); + assertNotNull(messageHistory); + assertEquals(2, messageHistory.size()); + Properties fooChannelHistory = messageHistory.get(0); + assertEquals("fooChannel", fooChannelHistory.get("name")); + assertEquals("channel", fooChannelHistory.get("type")); + } + @Test public void testIteratorOfMessageGroups() throws Exception{ GemfireMessageStore store1 = new GemfireMessageStore(this.cache); diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java index b9e0c08683..2884467f26 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java @@ -16,17 +16,24 @@ package org.springframework.integration.gemfire.store; +import java.util.Properties; + import org.junit.After; import org.junit.Before; import org.junit.Test; + import org.springframework.data.gemfire.CacheFactoryBean; import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.history.MessageHistory; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; import com.gemstone.gemfire.cache.Cache; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; /** * @author Mark Fisher @@ -46,6 +53,29 @@ public class GemfireMessageStoreTests { Message retrieved = store.getMessage(message.getHeaders().getId()); assertEquals(message, retrieved); } + + @Test + public void testWithMessageHistory() throws Exception{ + GemfireMessageStore store = new GemfireMessageStore(this.cache); + store.afterPropertiesSet(); + + Message message = new GenericMessage("Hello"); + DirectChannel fooChannel = new DirectChannel(); + fooChannel.setBeanName("fooChannel"); + DirectChannel barChannel = new DirectChannel(); + barChannel.setBeanName("barChannel"); + + message = MessageHistory.write(message, fooChannel); + message = MessageHistory.write(message, barChannel); + store.addMessage(message); + message = store.getMessage(message.getHeaders().getId()); + MessageHistory messageHistory = MessageHistory.read(message); + assertNotNull(messageHistory); + assertEquals(2, messageHistory.size()); + Properties fooChannelHistory = messageHistory.get(0); + assertEquals("fooChannel", fooChannelHistory.get("name")); + assertEquals("channel", fooChannelHistory.get("type")); + } @Before public void init() throws Exception{ diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index d5d2aad8ab..ad9b6220e3 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -16,20 +16,12 @@ package org.springframework.integration.jdbc; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNotSame; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertSame; -import static org.junit.Assert.assertThat; -import static org.junit.Assert.assertTrue; -import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders; - import java.io.BufferedReader; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import java.io.OutputStream; +import java.util.Properties; import java.util.UUID; import javax.sql.DataSource; @@ -37,10 +29,13 @@ import javax.sql.DataSource; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupCallback; @@ -50,6 +45,16 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.transaction.annotation.Transactional; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders; + @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) public class JdbcMessageStoreTests { @@ -83,6 +88,28 @@ public class JdbcMessageStoreTests { assertNotNull(result.getHeaders().get(JdbcMessageStore.SAVED_KEY)); assertNotNull(result.getHeaders().get(JdbcMessageStore.CREATED_DATE_KEY)); } + + @Test + @Transactional + public void testWithMessageHistory() throws Exception{ + + Message message = new GenericMessage("Hello"); + DirectChannel fooChannel = new DirectChannel(); + fooChannel.setBeanName("fooChannel"); + DirectChannel barChannel = new DirectChannel(); + barChannel.setBeanName("barChannel"); + + message = MessageHistory.write(message, fooChannel); + message = MessageHistory.write(message, barChannel); + messageStore.addMessage(message); + message = messageStore.getMessage(message.getHeaders().getId()); + MessageHistory messageHistory = MessageHistory.read(message); + assertNotNull(messageHistory); + assertEquals(2, messageHistory.size()); + Properties fooChannelHistory = messageHistory.get(0); + assertEquals("fooChannel", fooChannelHistory.get("name")); + assertEquals("channel", fooChannelHistory.get("type")); + } @Test @Transactional diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java index 94f5815190..863da108f4 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.redis.store; import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -26,11 +27,14 @@ import junit.framework.AssertionFailedError; import org.junit.Ignore; import org.junit.Test; + import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; @@ -184,6 +188,34 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); } + + @Test + @RedisAvailable + public void testWithMessageHistory() throws Exception{ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + + store.getMessageGroup(1); + + Message message = new GenericMessage("Hello"); + DirectChannel fooChannel = new DirectChannel(); + fooChannel.setBeanName("fooChannel"); + DirectChannel barChannel = new DirectChannel(); + barChannel.setBeanName("barChannel"); + + message = MessageHistory.write(message, fooChannel); + message = MessageHistory.write(message, barChannel); + store.addMessageToGroup(1, message); + + message = store.getMessageGroup(1).getMessages().iterator().next(); + + MessageHistory messageHistory = MessageHistory.read(message); + assertNotNull(messageHistory); + assertEquals(2, messageHistory.size()); + Properties fooChannelHistory = messageHistory.get(0); + assertEquals("fooChannel", fooChannelHistory.get("name")); + assertEquals("channel", fooChannelHistory.get("type")); + } @Test @RedisAvailable public void testRemoveNonExistingMessageFromTheGroup() throws Exception{ diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java index 418ddcc77f..a8b76609af 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java @@ -16,11 +16,15 @@ package org.springframework.integration.redis.store; import java.io.Serializable; +import java.util.Properties; import java.util.UUID; import org.junit.Test; + import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; @@ -115,6 +119,30 @@ public class RedisMessageStoreTests extends RedisAvailableTests { assertNull(store.getMessage(stringMessage.getHeaders().getId())); } + @Test + @RedisAvailable + public void testWithMessageHistory() throws Exception{ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + + Message message = new GenericMessage("Hello"); + DirectChannel fooChannel = new DirectChannel(); + fooChannel.setBeanName("fooChannel"); + DirectChannel barChannel = new DirectChannel(); + barChannel.setBeanName("barChannel"); + + message = MessageHistory.write(message, fooChannel); + message = MessageHistory.write(message, barChannel); + store.addMessage(message); + message = store.getMessage(message.getHeaders().getId()); + MessageHistory messageHistory = MessageHistory.read(message); + assertNotNull(messageHistory); + assertEquals(2, messageHistory.size()); + Properties fooChannelHistory = messageHistory.get(0); + assertEquals("fooChannel", fooChannelHistory.get("name")); + assertEquals("channel", fooChannelHistory.get("type")); + } + @SuppressWarnings("serial") public static class Person implements Serializable{ private Address address;