INT-3078, INT-3079: Redis & Gemfire Delayer Tests
Add Redis & Gemfire Delayer integration tests to be sure that Delayer works correctly JIRA: https://jira.springsource.org/browse/INT-3078 https://jira.springsource.org/browse/INT-3079
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
afe56ca6d5
commit
bb71fd8a37
@@ -0,0 +1,22 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns="http://www.springframework.org/schema/integration"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
|
||||
|
||||
<beans:bean id="messageStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
|
||||
<beans:constructor-arg value="#{T (org.springframework.integration.gemfire.store.DelayerHandlerRescheduleIntegrationTests).cache}"/>
|
||||
</beans:bean>
|
||||
|
||||
<channel id="output">
|
||||
<queue/>
|
||||
</channel>
|
||||
|
||||
<delayer id="#{T (org.springframework.integration.gemfire.store.DelayerHandlerRescheduleIntegrationTests).DELAYER_ID}"
|
||||
input-channel="input"
|
||||
output-channel="output"
|
||||
default-delay="500"
|
||||
message-store="messageStore"/>
|
||||
|
||||
</beans:beans>
|
||||
@@ -0,0 +1,120 @@
|
||||
/*
|
||||
* Copyright 2013 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.gemfire.store;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNotSame;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.context.support.AbstractApplicationContext;
|
||||
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.core.PollableChannel;
|
||||
import org.springframework.integration.handler.DelayHandler;
|
||||
import org.springframework.integration.store.MessageGroup;
|
||||
import org.springframework.integration.store.MessageGroupStore;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
import com.gemstone.gemfire.cache.Cache;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
* @since 3.0
|
||||
*/
|
||||
public class DelayerHandlerRescheduleIntegrationTests {
|
||||
|
||||
public static final String DELAYER_ID = "delayerWithGemfireMS";
|
||||
|
||||
public static Cache cache;
|
||||
|
||||
@BeforeClass
|
||||
public static void startUp() throws Exception {
|
||||
CacheFactoryBean cacheFactoryBean = new CacheFactoryBean();
|
||||
cache = cacheFactoryBean.getObject();
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void cleanUp() {
|
||||
cache.close();
|
||||
Assert.isTrue(cache.isClosed(), "Cache did not close after close() call");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDelayerHandlerRescheduleWithGemfireMessageStore() throws Exception {
|
||||
AbstractApplicationContext context = new ClassPathXmlApplicationContext(
|
||||
"DelayerHandlerRescheduleIntegrationTests-context.xml", this.getClass());
|
||||
MessageChannel input = context.getBean("input", MessageChannel.class);
|
||||
MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class);
|
||||
|
||||
String delayerMessageGroupId = DELAYER_ID + ".messageGroupId";
|
||||
|
||||
Message<String> message1 = MessageBuilder.withPayload("test1").build();
|
||||
input.send(message1);
|
||||
input.send(MessageBuilder.withPayload("test2").build());
|
||||
|
||||
// Emulate restart and check Cache state before next start
|
||||
context.destroy();
|
||||
|
||||
Thread.sleep(100);
|
||||
|
||||
try {
|
||||
context.getBean("input", MessageChannel.class);
|
||||
fail("IllegalStateException expected");
|
||||
}
|
||||
catch (Exception e) {
|
||||
assertTrue(e instanceof IllegalStateException);
|
||||
assertTrue(e.getMessage().contains("BeanFactory not initialized or already closed - call 'refresh'"));
|
||||
}
|
||||
|
||||
assertEquals(1, messageStore.getMessageGroupCount());
|
||||
assertEquals(delayerMessageGroupId, messageStore.iterator().next().getGroupId());
|
||||
assertEquals(2, messageStore.messageGroupSize(delayerMessageGroupId));
|
||||
assertEquals(2, messageStore.getMessageCountForAllMessageGroups());
|
||||
MessageGroup messageGroup = messageStore.getMessageGroup(delayerMessageGroupId);
|
||||
Message<?> messageInStore = messageGroup.getMessages().iterator().next();
|
||||
Object payload = messageInStore.getPayload();
|
||||
|
||||
// INT-3049
|
||||
assertTrue(payload instanceof DelayHandler.DelayedMessageWrapper);
|
||||
assertEquals(message1, ((DelayHandler.DelayedMessageWrapper) payload).getOriginal());
|
||||
|
||||
context.refresh();
|
||||
|
||||
PollableChannel output = context.getBean("output", PollableChannel.class);
|
||||
|
||||
Message<?> message = output.receive(10000);
|
||||
assertNotNull(message);
|
||||
|
||||
Object payload1 = message.getPayload();
|
||||
|
||||
message = output.receive(10000);
|
||||
assertNotNull(message);
|
||||
Object payload2 = message.getPayload();
|
||||
assertNotSame(payload1, payload2);
|
||||
|
||||
assertEquals(1, messageStore.getMessageGroupCount());
|
||||
assertEquals(0, messageStore.messageGroupSize(delayerMessageGroupId));
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns="http://www.springframework.org/schema/integration"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
|
||||
|
||||
<beans:bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
|
||||
<beans:property name="port" value="7379" />
|
||||
</beans:bean>
|
||||
|
||||
<beans:bean id="messageStore" class="org.springframework.integration.redis.store.RedisMessageStore">
|
||||
<beans:constructor-arg ref="redisConnectionFactory"/>
|
||||
</beans:bean>
|
||||
|
||||
<channel id="output">
|
||||
<queue/>
|
||||
</channel>
|
||||
|
||||
<delayer id="#{T (org.springframework.integration.redis.store.DelayerHandlerRescheduleIntegrationTests).DELAYER_ID}"
|
||||
input-channel="input"
|
||||
output-channel="output"
|
||||
default-delay="500"
|
||||
message-store="messageStore"/>
|
||||
|
||||
</beans:beans>
|
||||
@@ -0,0 +1,107 @@
|
||||
/*
|
||||
* Copyright 2013 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.redis.store;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNotSame;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.context.support.AbstractApplicationContext;
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
import org.springframework.integration.Message;
|
||||
import org.springframework.integration.MessageChannel;
|
||||
import org.springframework.integration.core.PollableChannel;
|
||||
import org.springframework.integration.handler.DelayHandler;
|
||||
import org.springframework.integration.redis.rules.RedisAvailable;
|
||||
import org.springframework.integration.redis.rules.RedisAvailableTests;
|
||||
import org.springframework.integration.store.MessageGroup;
|
||||
import org.springframework.integration.store.MessageGroupStore;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
* @since 3.0
|
||||
*/
|
||||
public class DelayerHandlerRescheduleIntegrationTests extends RedisAvailableTests {
|
||||
|
||||
public static final String DELAYER_ID = "delayerWithRedisMS";
|
||||
|
||||
|
||||
|
||||
@Test
|
||||
@RedisAvailable
|
||||
public void testDelayerHandlerRescheduleWithRedisMessageStore() throws Exception {
|
||||
AbstractApplicationContext context = new ClassPathXmlApplicationContext(
|
||||
"DelayerHandlerRescheduleIntegrationTests-context.xml", this.getClass());
|
||||
MessageChannel input = context.getBean("input", MessageChannel.class);
|
||||
MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class);
|
||||
|
||||
String delayerMessageGroupId = DELAYER_ID + ".messageGroupId";
|
||||
|
||||
messageStore.removeMessageGroup(delayerMessageGroupId);
|
||||
|
||||
Message<String> message1 = MessageBuilder.withPayload("test1").build();
|
||||
input.send(message1);
|
||||
input.send(MessageBuilder.withPayload("test2").build());
|
||||
|
||||
// Emulate restart and check DB state before next start
|
||||
context.destroy();
|
||||
|
||||
Thread.sleep(100);
|
||||
|
||||
try {
|
||||
context.getBean("input", MessageChannel.class);
|
||||
fail("IllegalStateException expected");
|
||||
}
|
||||
catch (Exception e) {
|
||||
assertTrue(e instanceof IllegalStateException);
|
||||
assertTrue(e.getMessage().contains("BeanFactory not initialized or already closed - call 'refresh'"));
|
||||
}
|
||||
|
||||
assertEquals(1, messageStore.getMessageGroupCount());
|
||||
assertEquals(delayerMessageGroupId, messageStore.iterator().next().getGroupId());
|
||||
assertEquals(2, messageStore.messageGroupSize(delayerMessageGroupId));
|
||||
assertEquals(2, messageStore.getMessageCountForAllMessageGroups());
|
||||
MessageGroup messageGroup = messageStore.getMessageGroup(delayerMessageGroupId);
|
||||
Message<?> messageInStore = messageGroup.getMessages().iterator().next();
|
||||
Object payload = messageInStore.getPayload();
|
||||
|
||||
// INT-3049
|
||||
assertTrue(payload instanceof DelayHandler.DelayedMessageWrapper);
|
||||
assertEquals(message1, ((DelayHandler.DelayedMessageWrapper) payload).getOriginal());
|
||||
|
||||
context.refresh();
|
||||
|
||||
PollableChannel output = context.getBean("output", PollableChannel.class);
|
||||
|
||||
Message<?> message = output.receive(10000);
|
||||
assertNotNull(message);
|
||||
|
||||
Object payload1 = message.getPayload();
|
||||
|
||||
message = output.receive(10000);
|
||||
assertNotNull(message);
|
||||
Object payload2 = message.getPayload();
|
||||
assertNotSame(payload1, payload2);
|
||||
|
||||
assertEquals(1, messageStore.getMessageGroupCount());
|
||||
assertEquals(0, messageStore.messageGroupSize(delayerMessageGroupId));
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user