Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2016 modified DefaultAmqpHeaderMapper to NOT copy headers such as ID, TIMESTAMP, REPLY_CHANNEL and ERRPR_CHANNEL

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-07-28 07:47:52 -04:00
parent 02735b15f1
commit 47e3b47955
4 changed files with 75 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
View File

@@ -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<String> 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

3
spring-integration-core/.springBeans
View File

@@ -1,7 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.6.0.201103160035-RELEASE]]></pluginVersion>
<pluginVersion><![CDATA[2.7.1.201107082359-RELEASE]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
@@ -10,6 +10,7 @@
<config>src/test/java/org/springframework/integration/gateway/GatewayInterfaceTest-context.xml</config>
<config>src/test/java/org/springframework/integration/gateway/InnerGatewayWithChainTests-context.xml</config>
<config>src/test/java/org/springframework/integration/history/annotated-config.xml</config>
<config>src/test/java/org/springframework/integration/aggregator/config.xml</config>
</configs>
<configSets>
</configSets>

17
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoMessageStore.java
View File

@@ -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 <T> Message<T> addMessage(Message<T> 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);
}

46
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoMessageStoreTests.java
View File

@@ -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;
}
}
}