From 98b42cbe36e9696f93ad2befaa49f3e65c9d0f4e Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 2 Sep 2013 14:11:57 +0300 Subject: [PATCH] INT-3070: Fix Delayer with MongoDbMessageStore The `DelayHandler.DelayedMessageWrapper` works correctly, as designed by INT-3049. However `DelayHandler.DelayedMessageWrapper` contains `Message` as a property and `GenericMessage` can't be reconstructed automatically. So, added new `DBObjectToGenericMessageConverter` to read `GenericMessage` for `DelayHandler.DelayedMessageWrapper#original` property. JIRA: https://jira.springsource.org/browse/INT-3070 Cherry-picked 2.2.x INT-3070: PR comments INT-3070: Remove `assert` for common state --- .../mongodb/store/MongoDbMessageStore.java | 40 ++++++- ...dlerRescheduleIntegrationTests-context.xml | 29 +++++ ...ayerHandlerRescheduleIntegrationTests.java | 111 ++++++++++++++++++ 3 files changed, 177 insertions(+), 3 deletions(-) create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index 2fded20ade..585eb7052b 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -63,6 +63,7 @@ import com.mongodb.BasicDBList; import com.mongodb.BasicDBObject; import com.mongodb.DBObject; + /** * An implementation of both the {@link MessageStore} and {@link MessageGroupStore} * strategies that relies upon MongoDB for persistence. @@ -72,6 +73,7 @@ import com.mongodb.DBObject; * @author Sean Brandt * @author Jodie StJohn * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class MongoDbMessageStore extends AbstractMessageGroupStore implements MessageStore, BeanClassLoaderAware { @@ -313,6 +315,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me customConverters.add(new UuidToDBObjectConverter()); customConverters.add(new DBObjectToUUIDConverter()); customConverters.add(new MessageHistoryToDBObjectConverter()); + customConverters.add(new DBObjectToGenericMessageConverter()); this.setCustomConversions(new CustomConversions(customConverters)); super.afterPropertiesSet(); } @@ -372,7 +375,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me } if (completeGroup != null){ - wrapper.set_Group_complete(completeGroup.booleanValue()); + wrapper.set_Group_complete(completeGroup); } return (S) wrapper; @@ -414,8 +417,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me private static class DBObjectToUUIDConverter implements Converter { public UUID convert(DBObject source) { - UUID id = UUID.fromString((String) source.get("_value")); - return id; + return UUID.fromString((String) source.get("_value")); } } @@ -438,6 +440,38 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me } } + private class DBObjectToGenericMessageConverter implements Converter> { + + @SuppressWarnings("unchecked") + public GenericMessage convert(DBObject source) { + MessageReadingMongoConverter converter = (MessageReadingMongoConverter) MongoDbMessageStore.this.template + .getConverter(); + Map headers = converter.normalizeHeaders((Map) source.get("headers")); + + Object payload = source.get("payload"); + Object payloadType = source.get(PAYLOAD_TYPE_KEY); + if (payloadType != null && payload instanceof DBObject) { + try { + Class payloadClass = ClassUtils.forName(payloadType.toString(), classLoader); + payload = converter.read(payloadClass, (DBObject) payload); + } + catch (Exception e) { + throw new IllegalStateException("failed to load class: " + payloadType, e); + } + } + + @SuppressWarnings("rawtypes") + GenericMessage message = new GenericMessage(payload, headers); + Map innerMap = (Map) new DirectFieldAccessor(message.getHeaders()).getPropertyValue("headers"); + // using reflection to set ID and TIMESTAMP since they are immutable through MessageHeaders + innerMap.put(MessageHeaders.ID, headers.get(MessageHeaders.ID)); + innerMap.put(MessageHeaders.TIMESTAMP, headers.get(MessageHeaders.TIMESTAMP)); + + return message; + } + + } + /** * Wrapper class used for storing Messages in MongoDB along with their "group" metadata. */ diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml new file mode 100644 index 0000000000..b9e2ab74fb --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java new file mode 100644 index 0000000000..97487fc79f --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java @@ -0,0 +1,111 @@ +/* + * 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.mongodb.store; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.Iterator; + +import org.hamcrest.Matchers; +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.mongodb.rules.MongoDbAvailable; +import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; +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 MongoDbAvailableTests { + + public static final String DELAYER_ID = "delayerWithMongoMS"; + + @Test + @MongoDbAvailable + @SuppressWarnings("unchecked") + public void testDelayerHandlerRescheduleWithMongoDbMessageStore() 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 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(2, messageStore.messageGroupSize(delayerMessageGroupId)); + + MessageGroup messageGroup = messageStore.getMessageGroup(delayerMessageGroupId); + Iterator> iterator = messageGroup.getMessages().iterator(); + Message messageInStore = iterator.next(); + Object payload = messageInStore.getPayload(); + + // INT-3049 + assertTrue(payload instanceof DelayHandler.DelayedMessageWrapper); + + Message original1 = (Message) ((DelayHandler.DelayedMessageWrapper) payload).getOriginal(); + messageInStore = iterator.next(); + Message original2 = (Message) ((DelayHandler.DelayedMessageWrapper) messageInStore.getPayload()).getOriginal(); + assertThat(message1, Matchers.anyOf(Matchers.is(original1), Matchers.is(original2))); + + 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(0, messageStore.messageGroupSize(delayerMessageGroupId)); + + } + +}