Merge remote-tracking branch 'upstream/master' into 4.0.0-WIP
Conflicts: spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java spring-integration-core/src/test/java/org/springframework/integration/config/xml/ControlBusTests.java spring-integration-file/src/main/java/org/springframework/integration/file/DefaultFileNameGenerator.java spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java spring-integration-http/src/test/java/org/springframework/integration/http/outbound/UriVariableExpressionTests.java spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java spring-integration-jpa/src/test/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayIntegrationTests.java spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreTests.java spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java Resolved.
This commit is contained in:
Gary Russell
@@ -41,18 +41,23 @@ public abstract class MongoDbAvailableTests {
|
||||
public MongoDbAvailableRule redisAvailableRule = new MongoDbAvailableRule();
|
||||
|
||||
|
||||
protected MongoDbFactory prepareMongoFactory(String... additionalCollectionToDrop) throws Exception{
|
||||
protected MongoDbFactory prepareMongoFactory(String... additionalCollectionsToDrop) throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
MongoTemplate template = new MongoTemplate(mongoDbFactory);
|
||||
template.dropCollection("messages");
|
||||
template.dropCollection("data");
|
||||
for (String additionalCollection : additionalCollectionToDrop) {
|
||||
template.dropCollection(additionalCollection);
|
||||
}
|
||||
cleanupCollections(mongoDbFactory, additionalCollectionsToDrop);
|
||||
return mongoDbFactory;
|
||||
}
|
||||
|
||||
public Person createPerson(){
|
||||
protected void cleanupCollections(MongoDbFactory mongoDbFactory, String... additionalCollectionsToDrop) {
|
||||
MongoTemplate template = new MongoTemplate(mongoDbFactory);
|
||||
template.dropCollection("messages");
|
||||
template.dropCollection("configurableStoreMessages");
|
||||
template.dropCollection("data");
|
||||
for (String additionalCollection : additionalCollectionsToDrop) {
|
||||
template.dropCollection(additionalCollection);
|
||||
}
|
||||
}
|
||||
|
||||
public Person createPerson() {
|
||||
Address address = new Address();
|
||||
address.setCity("Philadelphia");
|
||||
address.setStreet("2121 Rawn street");
|
||||
@@ -64,7 +69,7 @@ public abstract class MongoDbAvailableTests {
|
||||
return person;
|
||||
}
|
||||
|
||||
public Person createPerson(String name){
|
||||
public Person createPerson(String name) {
|
||||
Address address = new Address();
|
||||
address.setCity("Philadelphia");
|
||||
address.setStreet("2121 Rawn street");
|
||||
@@ -77,44 +82,61 @@ public abstract class MongoDbAvailableTests {
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private Address address;
|
||||
|
||||
private String name;
|
||||
|
||||
public Address getAddress() {
|
||||
return address;
|
||||
}
|
||||
|
||||
public void setAddress(Address address) {
|
||||
this.address = address;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Address {
|
||||
|
||||
private String street;
|
||||
|
||||
private String city;
|
||||
|
||||
private String state;
|
||||
|
||||
public String getStreet() {
|
||||
return street;
|
||||
}
|
||||
|
||||
public void setStreet(String street) {
|
||||
this.street = street;
|
||||
}
|
||||
|
||||
public String getCity() {
|
||||
return city;
|
||||
}
|
||||
|
||||
public void setCity(String city) {
|
||||
this.city = city;
|
||||
}
|
||||
|
||||
public String getState() {
|
||||
return state;
|
||||
}
|
||||
|
||||
public void setState(String state) {
|
||||
this.state = state;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class TestMongoConverter extends MappingMongoConverter {
|
||||
@@ -134,6 +156,7 @@ public abstract class MongoDbAvailableTests {
|
||||
public <S> S read(Class<S> clazz, DBObject source) {
|
||||
return super.read(clazz, source);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,505 @@
|
||||
/*
|
||||
* Copyright 2002-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.assertFalse;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Iterator;
|
||||
import java.util.Properties;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.history.MessageHistory;
|
||||
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.store.MessageStore;
|
||||
import org.springframework.integration.store.SimpleMessageGroup;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import com.mongodb.Mongo;
|
||||
|
||||
/**
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Gary Russell
|
||||
* @author Amol Nayak
|
||||
*
|
||||
*/
|
||||
public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvailableTests {
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testNonExistingEmptyMessageGroup() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = getMessageGroupStore();
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
assertNotNull(messageGroup);
|
||||
assertTrue(messageGroup instanceof SimpleMessageGroup);
|
||||
assertEquals(0, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupWithAddedMessagePrimitiveGroupId() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
MessageStore messageStore = this.getMessageStore();
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
messageGroup = store.addMessageToGroup(1, messageB);
|
||||
assertNotNull(messageGroup);
|
||||
assertEquals(2, messageGroup.size());
|
||||
Message<?> retrievedMessage = messageStore.getMessage(messageA.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId());
|
||||
// ensure that 'message_group' header that is only used internally is not propagated
|
||||
assertNull(retrievedMessage.getHeaders().get("message_group"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupWithAddedMessageUUIDGroupIdAndUUIDHeader() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
MessageStore messageStore = this.getMessageStore();
|
||||
Object id = UUID.randomUUID();
|
||||
MessageGroup messageGroup = store.getMessageGroup(id);
|
||||
UUID uuidA = UUID.randomUUID();
|
||||
Message<?> messageA = MessageBuilder.withPayload("A").setHeader("foo", uuidA).build();
|
||||
UUID uuidB = UUID.randomUUID();
|
||||
Message<?> messageB = MessageBuilder.withPayload("B").setHeader("foo", uuidB).build();
|
||||
store.addMessageToGroup(id, messageA);
|
||||
messageGroup = store.addMessageToGroup(id, messageB);
|
||||
assertNotNull(messageGroup);
|
||||
assertEquals(2, messageGroup.size());
|
||||
Message<?> retrievedMessage = messageStore.getMessage(messageA.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId());
|
||||
// ensure that 'message_group' header that is only used internally is not propagated
|
||||
assertNull(retrievedMessage.getHeaders().get("message_group"));
|
||||
Object fooHeader = retrievedMessage.getHeaders().get("foo");
|
||||
assertTrue(fooHeader instanceof UUID);
|
||||
assertEquals(uuidA, fooHeader);
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testCountMessagesInGroup() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(1, messageB);
|
||||
assertEquals(2, store.messageGroupSize(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testPollMessages() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(1, messageB);
|
||||
assertEquals(2, store.messageGroupSize(1));
|
||||
Message<?> out = store.pollMessageFromGroup(1);
|
||||
assertNotNull(out);
|
||||
assertEquals("A", out.getPayload());
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
out = store.pollMessageFromGroup(1);
|
||||
assertEquals("B", out.getPayload());
|
||||
assertEquals(0, store.messageGroupSize(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testSameMessageMultipleGroupsPoll() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(2, messageA);
|
||||
store.addMessageToGroup(3, messageA);
|
||||
store.addMessageToGroup(4, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(1, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(3);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(4);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(2);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(1);
|
||||
assertEquals(0, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testSameMessageMultipleGroupsRemove() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(2, messageA);
|
||||
store.addMessageToGroup(3, messageA);
|
||||
store.addMessageToGroup(4, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(1, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(3, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(4, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(2, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(1, messageA);
|
||||
assertEquals(0, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
messageGroup = store.addMessageToGroup(1, message);
|
||||
assertNotNull(messageGroup);
|
||||
assertEquals(1, messageGroup.size());
|
||||
long createdTimestamp = messageGroup.getTimestamp();
|
||||
long updatedTimestamp = messageGroup.getLastModified();
|
||||
assertEquals(createdTimestamp, updatedTimestamp);
|
||||
Thread.sleep(1000);
|
||||
message = new GenericMessage<String>("Hello again");
|
||||
messageGroup = store.addMessageToGroup(1, message);
|
||||
createdTimestamp = messageGroup.getTimestamp();
|
||||
updatedTimestamp = messageGroup.getLastModified();
|
||||
assertTrue(updatedTimestamp > createdTimestamp);
|
||||
assertEquals(2, messageGroup.size());
|
||||
|
||||
// make sure the store is properly rebuild from MongoDB
|
||||
store = this.getMessageGroupStore();
|
||||
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertEquals(2, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupMarkingMessage() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
messageGroup = store.addMessageToGroup(1, messageB);
|
||||
assertNotNull(messageGroup);
|
||||
assertEquals(2, messageGroup.size());
|
||||
|
||||
messageGroup = store.removeMessageFromGroup(1, messageA);
|
||||
assertEquals(1, messageGroup.size());
|
||||
|
||||
// validate that the updates were propagated to Mongo as well
|
||||
store = this.getMessageGroupStore();
|
||||
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertEquals(1, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testRemoveMessageGroup() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
MessageStore messageStore = this.getMessageStore();
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
UUID id = message.getHeaders().getId();
|
||||
messageGroup = store.addMessageToGroup(1, message);
|
||||
assertNotNull(messageGroup);
|
||||
assertEquals(1, messageGroup.size());
|
||||
message = messageStore.getMessage(id);
|
||||
assertNotNull(message);
|
||||
|
||||
store.removeMessageGroup(1);
|
||||
MessageGroup messageGroupA = store.getMessageGroup(1);
|
||||
assertEquals(0, messageGroupA.size());
|
||||
assertFalse(messageGroupA.equals(messageGroup));
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testCompleteMessageGroup() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
assertNotNull(messageGroup);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
store.addMessageToGroup(messageGroup.getGroupId(), message);
|
||||
store.completeGroup(messageGroup.getGroupId());
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertTrue(messageGroup.isComplete());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testLastReleasedSequenceNumber() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
assertNotNull(messageGroup);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
store.addMessageToGroup(messageGroup.getGroupId(), message);
|
||||
store.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), 5);
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testRemoveMessageFromTheGroup() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("2");
|
||||
store.addMessageToGroup(1, new GenericMessage<String>("1"));
|
||||
store.addMessageToGroup(1, message);
|
||||
messageGroup = store.addMessageToGroup(1, new GenericMessage<String>("3"));
|
||||
assertNotNull(messageGroup);
|
||||
assertEquals(3, messageGroup.size());
|
||||
|
||||
messageGroup = store.removeMessageFromGroup(1, message);
|
||||
assertEquals(2, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMultipleMessageStores() throws Exception{
|
||||
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store1 = this.getMessageGroupStore();
|
||||
MessageGroupStore store2 = this.getMessageGroupStore();
|
||||
|
||||
Message<?> message = new GenericMessage<String>("1");
|
||||
store1.addMessageToGroup(1, message);
|
||||
store2.addMessageToGroup(1, new GenericMessage<String>("2"));
|
||||
store1.addMessageToGroup(1, new GenericMessage<String>("3"));
|
||||
|
||||
MessageGroupStore store3 = this.getMessageGroupStore();
|
||||
|
||||
MessageGroup messageGroup = store3.getMessageGroup(1);
|
||||
|
||||
assertNotNull(messageGroup);
|
||||
assertEquals(3, messageGroup.size());
|
||||
|
||||
store3.removeMessageFromGroup(1, message);
|
||||
|
||||
messageGroup = store2.getMessageGroup(1);
|
||||
assertEquals(2, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupIterator() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store1 = this.getMessageGroupStore();
|
||||
MessageGroupStore store2 = this.getMessageGroupStore();
|
||||
|
||||
Message<?> message = new GenericMessage<String>("1");
|
||||
store2.addMessageToGroup(1, message);
|
||||
store1.addMessageToGroup(2, new GenericMessage<String>("2"));
|
||||
store2.addMessageToGroup(3, new GenericMessage<String>("3"));
|
||||
|
||||
MessageGroupStore store3 = this.getMessageGroupStore();
|
||||
Iterator<MessageGroup> iterator = store3.iterator();
|
||||
assertNotNull(iterator);
|
||||
int counter = 0;
|
||||
while (iterator.hasNext()) {
|
||||
iterator.next();
|
||||
counter++;
|
||||
}
|
||||
assertEquals(3, counter);
|
||||
|
||||
store2.removeMessageFromGroup(1, message);
|
||||
|
||||
iterator = store3.iterator();
|
||||
counter = 0;
|
||||
while (iterator.hasNext()) {
|
||||
iterator.next();
|
||||
counter++;
|
||||
}
|
||||
assertEquals(2, counter);
|
||||
}
|
||||
|
||||
// @Test
|
||||
// @MongoDbAvailable
|
||||
// public void testConcurrentModifications() throws Exception{
|
||||
// MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
// final MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory);
|
||||
// final MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory);
|
||||
//
|
||||
// final Message<?> message = new GenericMessage<String>("1");
|
||||
//
|
||||
// ExecutorService executor = null;
|
||||
//
|
||||
// final List<Object> failures = new ArrayList<Object>();
|
||||
//
|
||||
// for (int i = 0; i < 100; i++) {
|
||||
// executor = Executors.newCachedThreadPool();
|
||||
//
|
||||
// executor.execute(new Runnable() {
|
||||
// public void run() {
|
||||
// MessageGroup group = store1.addMessageToGroup(1, message);
|
||||
// if (group.getUnmarked().size() != 1){
|
||||
// failures.add("ADD");
|
||||
// throw new AssertionFailedError("Failed on ADD");
|
||||
// }
|
||||
// }
|
||||
// });
|
||||
// executor.execute(new Runnable() {
|
||||
// public void run() {
|
||||
// MessageGroup group = store2.removeMessageFromGroup(1, message);
|
||||
// if (group.getUnmarked().size() != 0){
|
||||
// failures.add("REMOVE");
|
||||
// throw new AssertionFailedError("Failed on Remove");
|
||||
// }
|
||||
// }
|
||||
// });
|
||||
//
|
||||
// executor.shutdown();
|
||||
// executor.awaitTermination(10, TimeUnit.SECONDS);
|
||||
// store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle
|
||||
// }
|
||||
// assertTrue(failures.size() == 0);
|
||||
// }
|
||||
|
||||
|
||||
protected void testWithAggregatorWithShutdown(String config) throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
|
||||
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(config, this.getClass());
|
||||
context.refresh();
|
||||
|
||||
MessageChannel input = context.getBean("inputChannel", MessageChannel.class);
|
||||
QueueChannel output = context.getBean("outputChannel", QueueChannel.class);
|
||||
|
||||
Message<?> m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build();
|
||||
Message<?> m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build();
|
||||
input.send(m1);
|
||||
assertNull(output.receive(1000));
|
||||
input.send(m2);
|
||||
assertNull(output.receive(1000));
|
||||
context.close();
|
||||
|
||||
context = new ClassPathXmlApplicationContext(config, this.getClass());
|
||||
input = context.getBean("inputChannel", MessageChannel.class);
|
||||
output = context.getBean("outputChannel", QueueChannel.class);
|
||||
|
||||
Message<?> m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build();
|
||||
input.send(m3);
|
||||
assertNotNull(output.receive(2000));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithMessageHistory() throws Exception{
|
||||
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageGroupStore store = this.getMessageGroupStore();
|
||||
|
||||
store.getMessageGroup(1);
|
||||
|
||||
Message<?> message = new GenericMessage<String>("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);
|
||||
MessageGroup group = store.getMessageGroup(1);
|
||||
assertNotNull(group);
|
||||
Collection<Message<?>> messages = group.getMessages();
|
||||
assertTrue(!messages.isEmpty());
|
||||
message = messages.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"));
|
||||
}
|
||||
|
||||
protected abstract MessageGroupStore getMessageGroupStore() throws Exception;
|
||||
|
||||
protected abstract MessageStore getMessageStore() throws Exception;
|
||||
|
||||
}
|
||||
@@ -0,0 +1,295 @@
|
||||
/*
|
||||
* Copyright 2002-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.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Properties;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.history.MessageHistory;
|
||||
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
|
||||
import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
|
||||
import org.springframework.integration.store.MessageStore;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
import com.mongodb.Mongo;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Artem Bilan
|
||||
* @author Amol Nayak
|
||||
*
|
||||
*/
|
||||
public abstract class AbstractMongoDbMessageStoreTests extends MongoDbAvailableTests {
|
||||
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testAddGetWithStringPayload() throws Exception {
|
||||
cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageStore store = getMessageStore();
|
||||
Message<?> messageToStore = MessageBuilder.withPayload("Hello").build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testAddThenRemoveWithStringPayload() throws Exception {
|
||||
cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageStore store = getMessageStore();
|
||||
Message<?> messageToStore = MessageBuilder.withPayload("Hello").build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
store.removeMessage(retrievedMessage.getHeaders().getId());
|
||||
retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNull(retrievedMessage);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testAddGetWithObjectDefaultConstructorPayload() throws Exception {
|
||||
cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageStore store = getMessageStore();
|
||||
Person p = new Person();
|
||||
p.setFname("John");
|
||||
p.setLname("Doe");
|
||||
Message<?> messageToStore = MessageBuilder.withPayload(p).build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithMessageHistory() throws Exception{
|
||||
cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageStore store = getMessageStore();
|
||||
Foo foo = new Foo();
|
||||
foo.setName("foo");
|
||||
Message<?> message = MessageBuilder.withPayload(foo).
|
||||
setHeader("foo", foo).
|
||||
setHeader("bar", new Bar("bar")).
|
||||
setHeader("baz", new Baz()).
|
||||
setHeader("abc", new Abc()).
|
||||
setHeader("xyz", new Xyz()).
|
||||
build();
|
||||
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());
|
||||
assertNotNull(message);
|
||||
assertTrue(message.getHeaders().get("foo") instanceof Foo);
|
||||
assertTrue(message.getHeaders().get("bar") instanceof Bar);
|
||||
assertTrue(message.getHeaders().get("baz") instanceof Baz);
|
||||
assertTrue(message.getHeaders().get("abc") instanceof Abc);
|
||||
assertTrue(message.getHeaders().get("xyz") instanceof Xyz);
|
||||
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
|
||||
@MongoDbAvailable
|
||||
public void testInt3153SequenceDetails() throws Exception{
|
||||
cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
MessageStore store = getMessageStore();
|
||||
Message<?> messageToStore = MessageBuilder.withPayload("test")
|
||||
.pushSequenceDetails(UUID.randomUUID(), 1, 1)
|
||||
.pushSequenceDetails(UUID.randomUUID(), 1, 1)
|
||||
.build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
|
||||
public static class Foo implements Serializable {
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final long serialVersionUID = 1L;
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Bar implements Serializable{
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final String name;
|
||||
|
||||
public Bar(String name){
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Baz implements Serializable {
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final long serialVersionUID = 1L;
|
||||
private final String name = "baz";
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Abc implements Serializable {
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private final String name = "abx";
|
||||
|
||||
private Abc(){}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Xyz implements Serializable {
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
private final String name = "xyz";
|
||||
|
||||
private Xyz(){}
|
||||
}
|
||||
|
||||
|
||||
public static class Person implements Serializable{
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private String fname;
|
||||
|
||||
private String lname;
|
||||
|
||||
public String getFname() {
|
||||
return fname;
|
||||
}
|
||||
|
||||
public void setFname(String fname) {
|
||||
this.fname = fname;
|
||||
}
|
||||
|
||||
public String getLname() {
|
||||
return lname;
|
||||
}
|
||||
|
||||
public void setLname(String lname) {
|
||||
this.lname = lname;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
final int prime = 31;
|
||||
int result = 1;
|
||||
result = prime * result + ((fname == null) ? 0 : fname.hashCode());
|
||||
result = prime * result + ((lname == null) ? 0 : lname.hashCode());
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (this == obj) {
|
||||
return true;
|
||||
}
|
||||
if (obj == null) {
|
||||
return false;
|
||||
}
|
||||
if (getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
Person other = (Person) obj;
|
||||
if (fname == null) {
|
||||
if (other.fname != null) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
else if (!fname.equals(other.fname)) {
|
||||
return false;
|
||||
}
|
||||
if (lname == null) {
|
||||
if (other.lname != null) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
else if (!lname.equals(other.lname)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
protected abstract MessageStore getMessageStore() throws Exception;
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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 java.util.Map;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.core.convert.converter.Converter;
|
||||
import org.springframework.data.mongodb.MongoDbFactory;
|
||||
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
|
||||
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
|
||||
import org.springframework.integration.store.MessageStore;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
import com.mongodb.DBObject;
|
||||
import com.mongodb.Mongo;
|
||||
|
||||
/**
|
||||
* @author Amol Nayak
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGroupStoreTests {
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.integration.mongodb.store.AbstractMongoDbMessageGroupStoreTests#getMessageGroupStore()
|
||||
*/
|
||||
@Override
|
||||
protected ConfigurableMongoDbMessageStore getMessageGroupStore() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
ConfigurableMongoDbMessageStore mongoDbMessageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory);
|
||||
GenericApplicationContext testApplicationContext = TestUtils.createTestApplicationContext();
|
||||
testApplicationContext.refresh();
|
||||
mongoDbMessageStore.setApplicationContext(testApplicationContext);
|
||||
mongoDbMessageStore.afterPropertiesSet();
|
||||
return mongoDbMessageStore;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.integration.mongodb.store.AbstractMongoDbMessageGroupStoreTests#getMessageStore()
|
||||
*/
|
||||
@Override
|
||||
protected MessageStore getMessageStore() throws Exception {
|
||||
return this.getMessageGroupStore();
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithAggregatorWithShutdown() throws Exception {
|
||||
super.testWithAggregatorWithShutdown("mongo-aggregator-confugurable-config.xml");
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithCustomConverter() throws Exception {
|
||||
this.prepareMongoFactory("testConfigurableMongoDbMessageStore");
|
||||
ClassPathXmlApplicationContext context =
|
||||
new ClassPathXmlApplicationContext("ConfigurableMongoDbMessageStore-CustomConverter.xml", this.getClass());
|
||||
context.refresh();
|
||||
|
||||
TestGateway gateway = context.getBean(TestGateway.class);
|
||||
String result = gateway.service("foo");
|
||||
assertEquals("FOO", result);
|
||||
|
||||
}
|
||||
|
||||
public static interface TestGateway {
|
||||
|
||||
String service(String payload);
|
||||
|
||||
}
|
||||
|
||||
public static class MessageReadConverter implements Converter<DBObject, Message<?>> {
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public Message<?> convert(DBObject source) {
|
||||
return MessageBuilder.withPayload(source.get("payload")).copyHeaders((Map<String,?>) source.get("headers")).build();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans:beans xmlns="http://www.springframework.org/schema/integration"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:beans="http://www.springframework.org/schema/beans"
|
||||
xmlns:mongo="http://www.springframework.org/schema/data/mongo"
|
||||
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
|
||||
http://www.springframework.org/schema/data/mongo
|
||||
http://www.springframework.org/schema/data/mongo/spring-mongo.xsd">
|
||||
|
||||
<mongo:db-factory dbname="test"/>
|
||||
|
||||
<beans:bean id="messageStore" class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore">
|
||||
<beans:constructor-arg ref="mongoDbFactory"/>
|
||||
<beans:constructor-arg>
|
||||
<mongo:mapping-converter>
|
||||
<mongo:custom-converters>
|
||||
<mongo:converter>
|
||||
<beans:bean class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageGroupStoreTests$MessageReadConverter"/>
|
||||
</mongo:converter>
|
||||
</mongo:custom-converters>
|
||||
</mongo:mapping-converter>
|
||||
</beans:constructor-arg>
|
||||
<beans:constructor-arg value="testConfigurableMongoDbMessageStore"/>
|
||||
</beans:bean>
|
||||
|
||||
<gateway id="gateway"
|
||||
service-interface="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageGroupStoreTests$TestGateway"
|
||||
default-request-channel="preInputChannel"/>
|
||||
|
||||
<header-enricher input-channel="preInputChannel" output-channel="inputChannel">
|
||||
<header-channels-to-string/>
|
||||
</header-enricher>
|
||||
|
||||
<channel id="inputChannel">
|
||||
<queue message-store="messageStore"/>
|
||||
</channel>
|
||||
|
||||
<transformer input-channel="inputChannel" expression="payload.toUpperCase()">
|
||||
<poller fixed-delay="1000"/>
|
||||
</transformer>
|
||||
|
||||
</beans:beans>
|
||||
@@ -0,0 +1,123 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.assertThat;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import org.hamcrest.Matchers;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.data.mongodb.MongoDbFactory;
|
||||
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
|
||||
import org.springframework.integration.message.AdviceMessage;
|
||||
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
|
||||
import org.springframework.integration.store.MessageStore;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.support.ErrorMessage;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import com.mongodb.Mongo;
|
||||
|
||||
/**
|
||||
* @author Amol Nayak
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
public class ConfigurableMongoDbMessageStoreTests extends AbstractMongoDbMessageStoreTests {
|
||||
|
||||
@Override
|
||||
protected MessageStore getMessageStore() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
ConfigurableMongoDbMessageStore mongoDbMessageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory);
|
||||
GenericApplicationContext testApplicationContext = TestUtils.createTestApplicationContext();
|
||||
testApplicationContext.refresh();
|
||||
mongoDbMessageStore.setApplicationContext(testApplicationContext);
|
||||
mongoDbMessageStore.afterPropertiesSet();
|
||||
return mongoDbMessageStore;
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testInt3076MessageAsPayload() throws Exception{
|
||||
MessageStore store = this.getMessageStore();
|
||||
Person p = new Person();
|
||||
p.setFname("John");
|
||||
p.setLname("Doe");
|
||||
Message<?> messageToStore = new GenericMessage<Message<?>>(MessageBuilder.withPayload(p).build());
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertTrue(retrievedMessage.getPayload() instanceof GenericMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(((Message<?>) messageToStore.getPayload()).getPayload(), p);
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testInt3076AdviceMessage() throws Exception{
|
||||
MessageStore store = this.getMessageStore();
|
||||
Person p = new Person();
|
||||
p.setFname("John");
|
||||
p.setLname("Doe");
|
||||
Message<Person> inputMessage = MessageBuilder.withPayload(p).build();
|
||||
Message<?> messageToStore = new AdviceMessage("foo", inputMessage);
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertTrue(retrievedMessage instanceof AdviceMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(inputMessage, ((AdviceMessage) retrievedMessage).getInputMessage());
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testInt3076ErrorMessage() throws Exception{
|
||||
MessageStore store = this.getMessageStore();
|
||||
Person p = new Person();
|
||||
p.setFname("John");
|
||||
p.setLname("Doe");
|
||||
Message<Person> failedMessage = MessageBuilder.withPayload(p).build();
|
||||
MessagingException messagingException;
|
||||
try {
|
||||
throw new RuntimeException("intentional");
|
||||
}
|
||||
catch (Exception e) {
|
||||
messagingException = new MessagingException(failedMessage, "intentional MessagingException", e);
|
||||
}
|
||||
Message<?> messageToStore = new ErrorMessage(messagingException);
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertTrue(retrievedMessage instanceof ErrorMessage);
|
||||
assertThat(retrievedMessage.getPayload(), Matchers.instanceOf(MessagingException.class));
|
||||
assertEquals("intentional MessagingException", ((MessagingException) retrievedMessage.getPayload()).getMessage());
|
||||
assertEquals(failedMessage, ((MessagingException) retrievedMessage.getPayload()).getFailedMessage());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
<?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="mongoConnectionFactory" class="org.springframework.data.mongodb.core.SimpleMongoDbFactory">
|
||||
<beans:constructor-arg>
|
||||
<beans:bean class="com.mongodb.Mongo"/>
|
||||
</beans:constructor-arg>
|
||||
<beans:constructor-arg value="test"/>
|
||||
</beans:bean>
|
||||
|
||||
<beans:bean id="messageStore" class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore">
|
||||
<beans:constructor-arg ref="mongoConnectionFactory"/>
|
||||
</beans:bean>
|
||||
|
||||
<channel id="output">
|
||||
<queue/>
|
||||
</channel>
|
||||
|
||||
<delayer id="#{T (org.springframework.integration.mongodb.store.DelayerHandlerRescheduleIntegrationTests).DELAYER_ID}"
|
||||
input-channel="input"
|
||||
output-channel="output"
|
||||
default-delay="10000"
|
||||
message-store="messageStore"/>
|
||||
|
||||
</beans:beans>
|
||||
@@ -55,10 +55,19 @@ public class DelayerHandlerRescheduleIntegrationTests extends MongoDbAvailableTe
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithMongoDbMessageStore() throws Exception {
|
||||
this.testDelayerHandlerRescheduleWithMongoDbMessageStore("DelayerHandlerRescheduleIntegrationTests-context.xml");
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithConfigurableMongoDbMessageStore() throws Exception {
|
||||
this.testDelayerHandlerRescheduleWithMongoDbMessageStore("DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml");
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testDelayerHandlerRescheduleWithMongoDbMessageStore() throws Exception {
|
||||
AbstractApplicationContext context = new ClassPathXmlApplicationContext(
|
||||
"DelayerHandlerRescheduleIntegrationTests-context.xml", this.getClass());
|
||||
private void testDelayerHandlerRescheduleWithMongoDbMessageStore(String config) throws Exception {
|
||||
AbstractApplicationContext context = new ClassPathXmlApplicationContext(config, this.getClass());
|
||||
MessageChannel input = context.getBean("input", MessageChannel.class);
|
||||
MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class);
|
||||
|
||||
|
||||
@@ -15,470 +15,35 @@
|
||||
*/
|
||||
package org.springframework.integration.mongodb.store;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.util.Iterator;
|
||||
import java.util.Properties;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
import org.springframework.data.mongodb.MongoDbFactory;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.history.MessageHistory;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
|
||||
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.SimpleMessageGroup;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.store.MessageStore;
|
||||
|
||||
import com.mongodb.Mongo;
|
||||
|
||||
/**
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests {
|
||||
public class MongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGroupStoreTests {
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testNonExistingEmptyMessageGroup() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
@Override
|
||||
protected MongoDbMessageStore getMessageGroupStore() throws Exception {
|
||||
return new MongoDbMessageStore( new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
}
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
assertNotNull(messageGroup);
|
||||
assertTrue(messageGroup instanceof SimpleMessageGroup);
|
||||
assertEquals(0, messageGroup.size());
|
||||
@Override
|
||||
protected MessageStore getMessageStore() throws Exception {
|
||||
return this.getMessageGroupStore();
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupWithAddedMessagePrimitiveGroupId() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
messageGroup = store.addMessageToGroup(1, messageB);
|
||||
assertEquals(2, messageGroup.size());
|
||||
Message<?> retrievedMessage = store.getMessage(messageA.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId());
|
||||
// ensure that 'message_group' header that is only used internally is not propagated
|
||||
assertNull(retrievedMessage.getHeaders().get("message_group"));
|
||||
public void testWithAggregatorWithShutdown() throws Exception {
|
||||
super.testWithAggregatorWithShutdown("mongo-aggregator-config.xml");
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupWithAddedMessageUUIDGroupIdAndUUIDHeader() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
Object id = UUID.randomUUID();
|
||||
MessageGroup messageGroup = store.getMessageGroup(id);
|
||||
UUID uuidA = UUID.randomUUID();
|
||||
Message<?> messageA = MessageBuilder.withPayload("A").setHeader("foo", uuidA).build();
|
||||
UUID uuidB = UUID.randomUUID();
|
||||
Message<?> messageB = MessageBuilder.withPayload("B").setHeader("foo", uuidB).build();
|
||||
store.addMessageToGroup(id, messageA);
|
||||
messageGroup = store.addMessageToGroup(id, messageB);
|
||||
assertEquals(2, messageGroup.size());
|
||||
Message<?> retrievedMessage = store.getMessage(messageA.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId());
|
||||
// ensure that 'message_group' header that is only used internally is not propagated
|
||||
assertNull(retrievedMessage.getHeaders().get("message_group"));
|
||||
Object fooHeader = retrievedMessage.getHeaders().get("foo");
|
||||
assertTrue(fooHeader instanceof UUID);
|
||||
assertEquals(uuidA, fooHeader);
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testCountMessagesInGroup() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(1, messageB);
|
||||
assertEquals(2, store.messageGroupSize(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testPollMessages() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(1, messageB);
|
||||
assertEquals(2, store.messageGroupSize(1));
|
||||
Message<?> out = store.pollMessageFromGroup(1);
|
||||
assertEquals("A", out.getPayload());
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
out = store.pollMessageFromGroup(1);
|
||||
assertEquals("B", out.getPayload());
|
||||
assertEquals(0, store.messageGroupSize(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testSameMessageMultipleGroupsPoll() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(2, messageA);
|
||||
store.addMessageToGroup(3, messageA);
|
||||
store.addMessageToGroup(4, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(1, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(3);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(4);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(2);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.pollMessageFromGroup(1);
|
||||
assertEquals(0, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testSameMessageMultipleGroupsRemove() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
store.addMessageToGroup(2, messageA);
|
||||
store.addMessageToGroup(3, messageA);
|
||||
store.addMessageToGroup(4, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(1, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(3, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(1, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(4, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(1, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(2, messageA);
|
||||
assertEquals(1, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
store.removeMessageFromGroup(1, messageA);
|
||||
assertEquals(0, store.messageGroupSize(1));
|
||||
assertEquals(0, store.messageGroupSize(2));
|
||||
assertEquals(0, store.messageGroupSize(3));
|
||||
assertEquals(0, store.messageGroupSize(4));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
messageGroup = store.addMessageToGroup(1, message);
|
||||
assertEquals(1, messageGroup.size());
|
||||
long createdTimestamp = messageGroup.getTimestamp();
|
||||
long updatedTimestamp = messageGroup.getLastModified();
|
||||
assertEquals(createdTimestamp, updatedTimestamp);
|
||||
Thread.sleep(1000);
|
||||
message = new GenericMessage<String>("Hello again");
|
||||
messageGroup = store.addMessageToGroup(1, message);
|
||||
createdTimestamp = messageGroup.getTimestamp();
|
||||
updatedTimestamp = messageGroup.getLastModified();
|
||||
assertTrue(updatedTimestamp > createdTimestamp);
|
||||
assertEquals(2, messageGroup.size());
|
||||
|
||||
// make sure the store is properly rebuild from MongoDB
|
||||
store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertEquals(2, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupMarkingMessage() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> messageA = new GenericMessage<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("B");
|
||||
store.addMessageToGroup(1, messageA);
|
||||
messageGroup = store.addMessageToGroup(1, messageB);
|
||||
assertEquals(2, messageGroup.size());
|
||||
|
||||
messageGroup = store.removeMessageFromGroup(1, messageA);
|
||||
assertEquals(1, messageGroup.size());
|
||||
|
||||
// validate that the updates were propagated to Mongo as well
|
||||
store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertEquals(1, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testRemoveMessageGroup() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
UUID id = message.getHeaders().getId();
|
||||
messageGroup = store.addMessageToGroup(1, message);
|
||||
assertEquals(1, messageGroup.size());
|
||||
message = store.getMessage(id);
|
||||
assertNotNull(message);
|
||||
|
||||
store.removeMessageGroup(1);
|
||||
MessageGroup messageGroupA = store.getMessageGroup(1);
|
||||
assertEquals(0, messageGroupA.size());
|
||||
assertFalse(messageGroupA.equals(messageGroup));
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testCompleteMessageGroup() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
store.addMessageToGroup(messageGroup.getGroupId(), message);
|
||||
store.completeGroup(messageGroup.getGroupId());
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertTrue(messageGroup.isComplete());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testLastReleasedSequenceNumber() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("Hello");
|
||||
store.addMessageToGroup(messageGroup.getGroupId(), message);
|
||||
store.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), 5);
|
||||
messageGroup = store.getMessageGroup(1);
|
||||
assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testRemoveMessageFromTheGroup() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store.getMessageGroup(1);
|
||||
Message<?> message = new GenericMessage<String>("2");
|
||||
store.addMessageToGroup(1, new GenericMessage<String>("1"));
|
||||
store.addMessageToGroup(1, message);
|
||||
messageGroup = store.addMessageToGroup(1, new GenericMessage<String>("3"));
|
||||
|
||||
assertEquals(3, messageGroup.size());
|
||||
|
||||
messageGroup = store.removeMessageFromGroup(1, message);
|
||||
assertEquals(2, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMultipleMessageStores() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory);
|
||||
MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Message<?> message = new GenericMessage<String>("1");
|
||||
store1.addMessageToGroup(1, message);
|
||||
store2.addMessageToGroup(1, new GenericMessage<String>("2"));
|
||||
store1.addMessageToGroup(1, new GenericMessage<String>("3"));
|
||||
|
||||
MongoDbMessageStore store3 = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
MessageGroup messageGroup = store3.getMessageGroup(1);
|
||||
|
||||
assertEquals(3, messageGroup.size());
|
||||
|
||||
store3.removeMessageFromGroup(1, message);
|
||||
|
||||
messageGroup = store2.getMessageGroup(1);
|
||||
assertEquals(2, messageGroup.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testMessageGroupIterator() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory);
|
||||
MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Message<?> message = new GenericMessage<String>("1");
|
||||
store2.addMessageToGroup(1, message);
|
||||
store1.addMessageToGroup(2, new GenericMessage<String>("2"));
|
||||
store2.addMessageToGroup(3, new GenericMessage<String>("3"));
|
||||
|
||||
MongoDbMessageStore store3 = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Iterator<MessageGroup> iterator = store3.iterator();
|
||||
int counter = 0;
|
||||
while (iterator.hasNext()) {
|
||||
iterator.next();
|
||||
counter++;
|
||||
}
|
||||
assertEquals(3, counter);
|
||||
|
||||
store2.removeMessageFromGroup(1, message);
|
||||
|
||||
iterator = store3.iterator();
|
||||
counter = 0;
|
||||
while (iterator.hasNext()) {
|
||||
iterator.next();
|
||||
counter++;
|
||||
}
|
||||
assertEquals(2, counter);
|
||||
}
|
||||
|
||||
// @Test
|
||||
// @MongoDbAvailable
|
||||
// public void testConcurrentModifications() throws Exception{
|
||||
// MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
// final MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory);
|
||||
// final MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory);
|
||||
//
|
||||
// final Message<?> message = new GenericMessage<String>("1");
|
||||
//
|
||||
// ExecutorService executor = null;
|
||||
//
|
||||
// final List<Object> failures = new ArrayList<Object>();
|
||||
//
|
||||
// for (int i = 0; i < 100; i++) {
|
||||
// executor = Executors.newCachedThreadPool();
|
||||
//
|
||||
// executor.execute(new Runnable() {
|
||||
// public void run() {
|
||||
// MessageGroup group = store1.addMessageToGroup(1, message);
|
||||
// if (group.getUnmarked().size() != 1){
|
||||
// failures.add("ADD");
|
||||
// throw new AssertionFailedError("Failed on ADD");
|
||||
// }
|
||||
// }
|
||||
// });
|
||||
// executor.execute(new Runnable() {
|
||||
// public void run() {
|
||||
// MessageGroup group = store2.removeMessageFromGroup(1, message);
|
||||
// if (group.getUnmarked().size() != 0){
|
||||
// failures.add("REMOVE");
|
||||
// throw new AssertionFailedError("Failed on Remove");
|
||||
// }
|
||||
// }
|
||||
// });
|
||||
//
|
||||
// executor.shutdown();
|
||||
// executor.awaitTermination(10, TimeUnit.SECONDS);
|
||||
// store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle
|
||||
// }
|
||||
// assertTrue(failures.size() == 0);
|
||||
// }
|
||||
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithAggregatorWithShutdown() throws Exception{
|
||||
this.prepareMongoFactory(); // for this test it only ensures that DB was flushed before test
|
||||
|
||||
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("mongo-aggregator-config.xml", this.getClass());
|
||||
MessageChannel input = context.getBean("inputChannel", MessageChannel.class);
|
||||
QueueChannel output = context.getBean("outputChannel", QueueChannel.class);
|
||||
|
||||
Message<?> m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build();
|
||||
Message<?> m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build();
|
||||
input.send(m1);
|
||||
assertNull(output.receive(1000));
|
||||
input.send(m2);
|
||||
assertNull(output.receive(1000));
|
||||
context.close();
|
||||
|
||||
context = new ClassPathXmlApplicationContext("mongo-aggregator-config.xml", this.getClass());
|
||||
input = context.getBean("inputChannel", MessageChannel.class);
|
||||
output = context.getBean("outputChannel", QueueChannel.class);
|
||||
|
||||
Message<?> m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build();
|
||||
input.send(m3);
|
||||
assertNotNull(output.receive(2000));
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithMessageHistory() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
store.getMessageGroup(1);
|
||||
|
||||
Message<?> message = new GenericMessage<String>("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"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2011 the original author or authors.
|
||||
* Copyright 2002-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.
|
||||
@@ -16,7 +16,12 @@
|
||||
|
||||
package org.springframework.integration.mongodb.store;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.data.mongodb.MongoDbFactory;
|
||||
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
|
||||
import org.springframework.messaging.Message;
|
||||
@@ -28,18 +33,17 @@ import org.springframework.integration.transformer.ClaimCheckOutTransformer;
|
||||
|
||||
import com.mongodb.Mongo;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
/**
|
||||
* @author Mark Fisher
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvailableTests{
|
||||
public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvailableTests {
|
||||
|
||||
@Test
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void stringPayload() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
MongoDbMessageStore messageStore = new MongoDbMessageStore(mongoDbFactory);
|
||||
MongoDbMessageStore messageStore = new MongoDbMessageStore(mongoDbFactory);
|
||||
ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore);
|
||||
ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore);
|
||||
Message<?> originalMessage = MessageBuilder.withPayload("test1").build();
|
||||
@@ -49,12 +53,9 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila
|
||||
assertEquals(claimCheckMessage.getPayload(), checkedOutMessage.getHeaders().getId());
|
||||
assertEquals(originalMessage.getPayload(), checkedOutMessage.getPayload());
|
||||
assertEquals(originalMessage, checkedOutMessage);
|
||||
//System.out.println("original: " + originalMessage);
|
||||
//System.out.println("claimcheck: " + claimCheckMessage);
|
||||
//System.out.println("checkedout: " + checkedOutMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void objectPayload() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
@@ -74,14 +75,52 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila
|
||||
assertEquals(originalMessage, checkedOutMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void stringPayloadConfigurable() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
ConfigurableMongoDbMessageStore messageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory);
|
||||
messageStore.afterPropertiesSet();
|
||||
ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore);
|
||||
ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore);
|
||||
Message<?> originalMessage = MessageBuilder.withPayload("test1").build();
|
||||
Message<?> claimCheckMessage = checkin.transform(originalMessage);
|
||||
assertEquals(originalMessage.getHeaders().getId(), claimCheckMessage.getPayload());
|
||||
Message<?> checkedOutMessage = checkout.transform(claimCheckMessage);
|
||||
assertEquals(claimCheckMessage.getPayload(), checkedOutMessage.getHeaders().getId());
|
||||
assertEquals(originalMessage.getPayload(), checkedOutMessage.getPayload());
|
||||
assertEquals(originalMessage, checkedOutMessage);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
private static class Beverage {
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void objectPayloadConfigurable() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
ConfigurableMongoDbMessageStore messageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory);
|
||||
messageStore.afterPropertiesSet();
|
||||
ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore);
|
||||
ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore);
|
||||
Beverage payload = new Beverage();
|
||||
payload.setName("latte");
|
||||
payload.setShots(3);
|
||||
payload.setIced(false);
|
||||
Message<?> originalMessage = MessageBuilder.withPayload(payload).build();
|
||||
Message<?> claimCheckMessage = checkin.transform(originalMessage);
|
||||
assertEquals(originalMessage.getHeaders().getId(), claimCheckMessage.getPayload());
|
||||
Message<?> checkedOutMessage = checkout.transform(claimCheckMessage);
|
||||
assertEquals(originalMessage.getPayload(), checkedOutMessage.getPayload());
|
||||
assertEquals(claimCheckMessage.getPayload(), checkedOutMessage.getHeaders().getId());
|
||||
assertEquals(originalMessage, checkedOutMessage);
|
||||
}
|
||||
|
||||
@SuppressWarnings("serial")
|
||||
private static class Beverage implements Serializable {
|
||||
|
||||
private String name;
|
||||
private int shots;
|
||||
private int shots;
|
||||
private boolean iced;
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
@@ -90,6 +129,7 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public int getShots() {
|
||||
return shots;
|
||||
}
|
||||
@@ -98,6 +138,7 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila
|
||||
this.shots = shots;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public boolean isIced() {
|
||||
return iced;
|
||||
}
|
||||
@@ -144,6 +185,7 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -13,247 +13,23 @@
|
||||
* 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.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.util.Properties;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.data.mongodb.MongoDbFactory;
|
||||
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.history.MessageHistory;
|
||||
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
|
||||
import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.integration.store.MessageStore;
|
||||
|
||||
import com.mongodb.Mongo;
|
||||
|
||||
/**
|
||||
* @author Mark Fisher
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
public class MongoDbMessageStoreTests extends MongoDbAvailableTests{
|
||||
public class MongoDbMessageStoreTests extends AbstractMongoDbMessageStoreTests {
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void addGetWithStringPayload() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
Message<?> messageToStore = MessageBuilder.withPayload("Hello").build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void addThenRemoveWithStringPayload() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
Message<?> messageToStore = MessageBuilder.withPayload("Hello").build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
store.removeMessage(retrievedMessage.getHeaders().getId());
|
||||
retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNull(retrievedMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void addGetWithObjectDefaultConstructorPayload() throws Exception {
|
||||
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test");
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
Person p = new Person();
|
||||
p.setFname("John");
|
||||
p.setLname("Doe");
|
||||
Message<?> messageToStore = MessageBuilder.withPayload(p).build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void testWithMessageHistory() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Foo foo = new Foo();
|
||||
foo.setName("foo");
|
||||
Message<?> message = MessageBuilder.withPayload(foo).
|
||||
setHeader("foo", foo).
|
||||
setHeader("bar", new Bar("bar")).
|
||||
setHeader("baz", new Baz()).
|
||||
setHeader("abc", new Abc()).
|
||||
setHeader("xyz", new Xyz()).
|
||||
build();
|
||||
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());
|
||||
assertTrue(message.getHeaders().get("foo") instanceof Foo);
|
||||
assertTrue(message.getHeaders().get("bar") instanceof Bar);
|
||||
assertTrue(message.getHeaders().get("baz") instanceof Baz);
|
||||
assertTrue(message.getHeaders().get("abc") instanceof Abc);
|
||||
assertTrue(message.getHeaders().get("xyz") instanceof Xyz);
|
||||
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
|
||||
@MongoDbAvailable
|
||||
public void testInt3153SequenceDetails() throws Exception{
|
||||
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
|
||||
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
|
||||
Message<?> messageToStore = MessageBuilder.withPayload("test")
|
||||
.pushSequenceDetails(UUID.randomUUID(), 1, 1)
|
||||
.pushSequenceDetails(UUID.randomUUID(), 1, 1)
|
||||
.build();
|
||||
store.addMessage(messageToStore);
|
||||
Message<?> retrievedMessage = store.getMessage(messageToStore.getHeaders().getId());
|
||||
assertNotNull(retrievedMessage);
|
||||
assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload());
|
||||
assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders());
|
||||
assertEquals(messageToStore, retrievedMessage);
|
||||
}
|
||||
|
||||
public static class Foo{
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Bar{
|
||||
private final String name;
|
||||
|
||||
public Bar(String name){
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Baz{
|
||||
private final String name = "baz";
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Abc{
|
||||
private final String name = "abx";
|
||||
|
||||
private Abc(){}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Xyz{
|
||||
@SuppressWarnings("unused")
|
||||
private final String name = "xyz";
|
||||
|
||||
private Xyz(){}
|
||||
}
|
||||
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String fname;
|
||||
|
||||
private String lname;
|
||||
|
||||
public String getFname() {
|
||||
return fname;
|
||||
}
|
||||
|
||||
public void setFname(String fname) {
|
||||
this.fname = fname;
|
||||
}
|
||||
|
||||
public String getLname() {
|
||||
return lname;
|
||||
}
|
||||
|
||||
public void setLname(String lname) {
|
||||
this.lname = lname;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
final int prime = 31;
|
||||
int result = 1;
|
||||
result = prime * result + ((fname == null) ? 0 : fname.hashCode());
|
||||
result = prime * result + ((lname == null) ? 0 : lname.hashCode());
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (this == obj) {
|
||||
return true;
|
||||
}
|
||||
if (obj == null) {
|
||||
return false;
|
||||
}
|
||||
if (getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
Person other = (Person) obj;
|
||||
if (fname == null) {
|
||||
if (other.fname != null) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
else if (!fname.equals(other.fname)) {
|
||||
return false;
|
||||
}
|
||||
if (lname == null) {
|
||||
if (other.lname != null) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
else if (!lname.equals(other.lname)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
@Override
|
||||
protected MessageStore getMessageStore() throws Exception {
|
||||
return new MongoDbMessageStore(new SimpleMongoDbFactory(new Mongo(), "test"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:int="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">
|
||||
|
||||
<int:aggregator input-channel="inputChannel" output-channel="outputChannel" message-store="mongoStore"/>
|
||||
|
||||
<int:channel id="outputChannel">
|
||||
<int:queue/>
|
||||
</int:channel>
|
||||
|
||||
<bean id="mongoStore" class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore">
|
||||
<constructor-arg ref="mongoConnectionFactory"/>
|
||||
</bean>
|
||||
|
||||
<bean id="mongoConnectionFactory" class="org.springframework.data.mongodb.core.SimpleMongoDbFactory">
|
||||
<constructor-arg>
|
||||
<bean class="com.mongodb.Mongo"/>
|
||||
</constructor-arg>
|
||||
<constructor-arg value="test"/>
|
||||
</bean>
|
||||
|
||||
</beans>
|
||||
Reference in New Issue
Block a user