diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java index 82960658e1..d54f7caef9 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java @@ -44,6 +44,7 @@ import org.springframework.util.StringUtils; * Constants for the AMQP header keys are defined in {@link AmqpHeaders}. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper { @@ -165,20 +166,22 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper { // now map to the user-defined headers, if any, within the AMQP MessageProperties Set headerNames = headers.keySet(); for (String headerName : headerNames) { - if (StringUtils.hasText(headerName) && !headerName.startsWith(AmqpHeaders.PREFIX)) { - Object value = headers.get(headerName); - if (value != null) { - try { - String key = this.fromHeaderName(headerName); - amqpMessageProperties.setHeader(key, value); - } - catch (Exception e) { - if (logger.isWarnEnabled()) { - logger.warn("failed to map Message header '" + headerName + "' to AMQP header", e); + if (this.shoudlMapHeader(headerName)){ + if (StringUtils.hasText(headerName) && !headerName.startsWith(AmqpHeaders.PREFIX)) { + Object value = headers.get(headerName); + if (value != null) { + try { + String key = this.fromHeaderName(headerName); + amqpMessageProperties.setHeader(key, value); + } + catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn("failed to map Message header '" + headerName + "' to AMQP header", e); + } } } } - } + } } } catch (Exception e) { @@ -187,6 +190,13 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper { } } } + + private boolean shoudlMapHeader(String headerName){ + return !headerName.equals(MessageHeaders.ERROR_CHANNEL) && + !headerName.equals(MessageHeaders.REPLY_CHANNEL) && + !headerName.equals(MessageHeaders.ID) && + !headerName.equals(MessageHeaders.TIMESTAMP); + } /** * Maps headers from an AMQP MessageProperties instance to the MessageHeaders of a diff --git a/spring-integration-core/.springBeans b/spring-integration-core/.springBeans index df15308a34..6a45d2853e 100644 --- a/spring-integration-core/.springBeans +++ b/spring-integration-core/.springBeans @@ -1,7 +1,7 @@ 1 - + @@ -10,6 +10,7 @@ src/test/java/org/springframework/integration/gateway/GatewayInterfaceTest-context.xml src/test/java/org/springframework/integration/gateway/InnerGatewayWithChainTests-context.xml src/test/java/org/springframework/integration/history/annotated-config.xml + src/test/java/org/springframework/integration/aggregator/config.xml diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoMessageStore.java index 952f6b3472..9e4c02717f 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoMessageStore.java @@ -56,6 +56,8 @@ public class MongoMessageStore implements MessageStore, BeanClassLoaderAware { private final static String DEFAULT_COLLECTION_NAME = "messages"; private final MongoTemplate template; + + private volatile String collectionName = DEFAULT_COLLECTION_NAME; //private final MongoConverter mongoConverter = new MessageReadingMongoConverter(); private final MongoConverter mongoConverter = null; @@ -76,8 +78,7 @@ public class MongoMessageStore implements MessageStore, BeanClassLoaderAware { public void setCollectionName(String collectionName) { Assert.hasText(collectionName, "collectionName must not be empty"); - //this.template.setDefaultCollectionName(collectionName); - this.template.createCollection(collectionName); + this.collectionName = collectionName; } // public void setUsername(String username) { @@ -94,12 +95,12 @@ public class MongoMessageStore implements MessageStore, BeanClassLoaderAware { } public Message addMessage(Message message) { - this.template.insert(message, "messages"); + this.template.insert(message, collectionName); return message; } public Message getMessage(UUID id) { - return this.template.findOne(this.idQuery(id), Message.class); + return this.template.findOne(this.idQuery(id), Message.class, this.collectionName); } public int getMessageCount() { @@ -108,11 +109,10 @@ public class MongoMessageStore implements MessageStore, BeanClassLoaderAware { } public Message removeMessage(UUID id) { - return this.template.findAndRemove(idQuery(id), Message.class, "messages"); + return this.template.findAndRemove(idQuery(id), Message.class, this.collectionName); } private Query idQuery(UUID id) { - System.out.println(id.toString()); return new Query(where("_id").is(id.toString())); } @@ -141,14 +141,17 @@ public class MongoMessageStore implements MessageStore, BeanClassLoaderAware { public void write(Object source, DBObject target) { if (source instanceof Message) { String payloadType = ((Message) source).getPayload().getClass().getName(); - target.put("_payloadType", payloadType); + //target.put("_payloadType", payloadType); target.put("_id", ((Message) source).getHeaders().getId().toString()); + //target.put("_id", ((Message) source).getHeaders().getId().toString()); } // // TODO: fix this (the base class should handle it via converters?) // if (source instanceof UUID) { +// System.out.println("Converting to UUID"); // target.put("uuid", ((UUID)source).toString()); // } + super.write(source, target); } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoMessageStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoMessageStoreTests.java index a4ca3e391f..9d9d3f5e86 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoMessageStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoMessageStoreTests.java @@ -17,6 +17,9 @@ package org.springframework.integration.mongodb.store; import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import org.springframework.integration.Message; import org.springframework.integration.support.MessageBuilder; @@ -29,12 +32,47 @@ import com.mongodb.Mongo; public class MongoMessageStoreTests { @Test - public void addAndRemove() throws Exception { + public void addGetWithStringPayload() throws Exception { Mongo mongo = new Mongo(); MongoMessageStore store = new MongoMessageStore(mongo, "test"); - Message message = MessageBuilder.withPayload("UUID again and again").build(); - Message claimCheck = store.addMessage(message); - message = store.removeMessage(message.getHeaders().getId()); + Message message = MessageBuilder.withPayload("Hello").build(); + System.out.println(message); + store.addMessage(message); + assertNotNull(store.getMessage(message.getHeaders().getId())); + } + + @SuppressWarnings("unchecked") + @Test + public void addGetWithObjectDefaultConstructorPayload() throws Exception { + Mongo mongo = new Mongo(); + MongoMessageStore store = new MongoMessageStore(mongo, "test"); + Person p = new Person(); + p.setFname("John"); + p.setLname("Doe"); + + Message message = MessageBuilder.withPayload(p).build(); + System.out.println(message); + store.addMessage(message); + Message m = store.getMessage(message.getHeaders().getId()); + assertNotNull(m); + //assertEquals("John", m.getPayload().getFname()); + } + + 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; + } } }