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

INT-2460: Remove Message Modification Logic in MS

Browse Source 
JIRA: https://jira.spring.io/browse/INT-2460,
https://jira.spring.io/browse/INT-4122

Since the main purpose of the `MessageStore` to persist message for durability and only,
it doesn't make sense to modify `Message` for additional headers like `SAVED` and `CREATED_DATE`.
Such a logic should be a part of metadata stored together with the message.
And it is provided by the out-of-the-box `MessageStore` implementation.
In addition we free ourselves from the reflection operations to retain `ID` and `TIMESTAMP` headers when we add `SAVED` and `CREATED_DATE`

* Control "already saved" logic in the `JdbcMessageStore`s via `DuplicateKeyException` on the `INSERT`.
This is much effective then additional `SELECT` in case of `SAVED` before
* Control "already saved" logic in the  `AbstractConfigurableMongoDbMessageStore` via `DuplicateKeyException` on the `INSERT`.
Since `MongoDbMessageStore` doesn't provide extra `messageId` field, perform extra `SELECT` before store document.
Anyway the `MongoDbMessageStore` isn't recommended for use.
We may consider to deprecate it
* Control "already saved" logic  in the `AbstractKeyValueMessageStore` via `putIfAbsent` operation

With this fix we persist message in the store as is without any modifications when we perform standard serialization procedure.
Any custom serializers should consider to use `MutableMessageBuilder` if there is a requirement to retain `ID` and `TIMESTAMP`

Rework `MongoDbMetadataStore.putIfAbsent()` to normal `findAndModify()` with particular `$setOnInsert`.
Technically the MongoDB query looks like:
```
db.collection.findAndModify({
  query: { _id: $key },
  update: {
    $setOnInsert: { value: $value } // perform modification only on upsert
  },
  new: false,   // don't return new doc if one is upserted
  upsert: true // insert the document if it does not exist
})
```

Move single import to the appropriate JavaDoc

Polishing after rebase
DEBUG messages in `doStoreIfAbsent()` implementations

* To keep track of the extra message information in the `MessageStore`, without `Message` modification, introduce `MessageMetadata` and `MessageHolder`
* Add `MessageStore#getMessageMetadata()`
* Modify MongoDb `MessageStore` to add extra `timestamp` for individual message
* Fix `ConcurrentAggregatorTests` race condition.
Since currently the default release strategy is `SimpleSequenceSizeReleaseStrategy` which is just based on the `MessageGroup` size, there is no guaranty which messages will complete the group in concurrent environment.
The test is really based on the `SequenceAwareMessageGroup` logic to discard the message with the same `correlationId`

Fix `@Copyright` format

Move cast to `MessageHolder` after `Assert.isInstanceOf(MessageHolder.class, messageHolder)`

Retain backward compatibility in the `AbstractKeyValueMessageStore`

Polishing
This commit is contained in:
Artem Bilan
2016-09-29 18:57:16 -04:00
committed by Gary Russell
parent 7c12288d2e
commit a1f554c04d
23 changed files with 436 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java
View File

@@ -20,14 +20,11 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
/**
@@ -44,15 +41,50 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
protected static final String MESSAGE_GROUP_KEY_PREFIX = "MESSAGE_GROUP_";
/**
* Represents the time when the message has been added to the store.
* @deprecated since 5.0. This constant isn't used any more.
*/
@Deprecated
protected static final String CREATED_DATE = "CREATED_DATE";
// MessageStore methods
@Override
public Message<?> getMessage(UUID id) {
Message<?> message = getRawMessage(id);
if (message != null) {
return normalizeMessage(message);
public Message<?> getMessage(UUID messageId) {
Assert.notNull(messageId, "'messageId' must not be null");
Object object = doRetrieve(MESSAGE_KEY_PREFIX + messageId);
if (object != null) {
return extractMessage(object);
}
else {
return null;
}
}
private Message<?> extractMessage(Object object) {
if (object instanceof MessageHolder) {
return ((MessageHolder) object).getMessage();
}
else if (object instanceof Message) {
return (Message<?>) object;
}
else {
throw new IllegalArgumentException(
"Object of class [" + object.getClass().getName() +
"] must be an instance of [org.springframework.integration.store.MessageHolder].");
}
}
@Override
public MessageMetadata getMessageMetadata(UUID messageId) {
Assert.notNull(messageId, "'messageId' must not be null");
Object object = doRetrieve(MESSAGE_KEY_PREFIX + messageId);
if (object != null) {
extractMessage(object);
if (object instanceof MessageHolder) {
return ((MessageHolder) object).getMessageMetadata();
}
}
return null;
}
@@ -62,21 +94,20 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
public <T> Message<T> addMessage(Message<T> message) {
Assert.notNull(message, "'message' must not be null");
UUID messageId = message.getHeaders().getId();
doStore(MESSAGE_KEY_PREFIX + messageId, message);
return (Message<T>) getRawMessage(messageId);
doStoreIfAbsent(MESSAGE_KEY_PREFIX + messageId, new MessageHolder(message));
return (Message<T>) getMessage(messageId);
}
@Override
public Message<?> removeMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Object message = doRemove(MESSAGE_KEY_PREFIX + id);
if (message != null) {
Assert.isInstanceOf(Message.class, message);
Object object = doRemove(MESSAGE_KEY_PREFIX + id);
if (object != null) {
return extractMessage(object);
}
if (message != null) {
return normalizeMessage((Message<?>) message);
else {
return null;
}
return null;
}
@Override
@@ -131,14 +162,12 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
}
for (Message<?> message : messages) {
// enrich Message with additional headers and add it to MS
Message<?> enrichedMessage = enrichMessage(message);
addMessage(enrichedMessage);
addMessage(message);
if (metadata != null) {
metadata.add(enrichedMessage.getHeaders().getId());
metadata.add(message.getHeaders().getId());
}
else {
group.add(enrichedMessage);
group.add(message);
}
}
@@ -296,41 +325,12 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
protected abstract void doStore(Object id, Object objectToStore);
protected abstract void doStoreIfAbsent(Object id, Object objectToStore);
protected abstract Object doRemove(Object id);
protected abstract Collection<?> doListKeys(String keyPattern);
@SuppressWarnings({ "rawtypes", "unchecked" })
private Message<?> normalizeMessage(Message<?> message) {
Message<?> normalizedMessage = getMessageBuilderFactory().fromMessage(message)
.removeHeader("CREATED_DATE")
.build();
Map innerMap = (Map) new DirectFieldAccessor(normalizedMessage.getHeaders()).getPropertyValue("headers");
innerMap.put(MessageHeaders.ID, message.getHeaders().getId());
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
return normalizedMessage;
}
/**
* Will enrich Message with additional meta headers
*/
@SuppressWarnings({ "rawtypes", "unchecked" })
private Message<?> enrichMessage(Message<?> message) {
Message<?> enrichedMessage = getMessageBuilderFactory().fromMessage(message)
.setHeader(CREATED_DATE, System.currentTimeMillis())
.build();
Map innerMap = (Map) new DirectFieldAccessor(enrichedMessage.getHeaders()).getPropertyValue("headers");
innerMap.put(MessageHeaders.ID, message.getHeaders().getId());
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
return enrichedMessage;
}
private Message<?> getRawMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Object message = doRetrieve(MESSAGE_KEY_PREFIX + id);
return (Message<?>) message;
}
private final class MessageGroupIterator implements Iterator<MessageGroup> {
private final Iterator<?> idIterator;

29
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java
View File

@@ -23,11 +23,6 @@ import java.util.LinkedHashSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -43,7 +38,7 @@ import org.springframework.messaging.Message;
*/
@ManagedResource
public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageGroupStore
implements MessageGroupStore, Iterable<MessageGroup>, BeanFactoryAware {
implements MessageGroupStore, Iterable<MessageGroup> {
protected final Log logger = LogFactory.getLog(getClass());
@@ -54,12 +49,6 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG
private volatile boolean timeoutOnIdle;
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private boolean lazyLoadMessageGroups = true;
protected AbstractMessageGroupStore() {
@@ -70,22 +59,6 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG
this.lazyLoadMessageGroups = lazyLoadMessageGroups;
}
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
protected MessageGroupFactory getMessageGroupFactory() {
if (this.lazyLoadMessageGroups) {

57
spring-integration-core/src/main/java/org/springframework/integration/store/MessageHolder.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* Copyright 2016 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.store;
import java.io.Serializable;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* The {@link MessageStore} specific value object to keep the {@link Message} and its metadata.
*
* @author Artem Bilan
* @since 5.0
*/
public class MessageHolder implements Serializable {
private static final long serialVersionUID = 1L;
private final Message<?> message;
private final MessageMetadata messageMetadata;
public MessageHolder(Message<?> message) {
Assert.notNull(message, "'message' must not be null.");
this.message = message;
this.messageMetadata = new MessageMetadata(message.getHeaders().getId());
this.messageMetadata.setTimestamp(System.currentTimeMillis());
}
public void setTimestamp(long timestamp) {
this.messageMetadata.setTimestamp(timestamp);
}
public Message<?> getMessage() {
return this.message;
}
public MessageMetadata getMessageMetadata() {
return this.messageMetadata;
}
}

52
spring-integration-core/src/main/java/org/springframework/integration/store/MessageMetadata.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2016 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.store;
import java.io.Serializable;
import java.util.UUID;
/**
* Value Object holding metadata about a Message in the MessageStore.
*
* @author Artem Bilan
* @since 5.0
*/
public class MessageMetadata implements Serializable {
private static final long serialVersionUID = 1L;
private final UUID messageId;
private long timestamp;
public MessageMetadata(UUID messageId) {
this.messageId = messageId;
}
public void setTimestamp(long timestamp) {
this.timestamp = timestamp;
}
public UUID getMessageId() {
return this.messageId;
}
public long getTimestamp() {
return this.timestamp;
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -27,6 +27,7 @@ import org.springframework.messaging.Message;
* @author Mark Fisher
* @author Iwein Fuld
* @author Dave Syer
* @author Artem Bilan
*
* @since 2.0
*/
@@ -38,6 +39,16 @@ public interface MessageStore {
*/
Message<?> getMessage(UUID id);
/**
* Return a {@link MessageMetadata} for the {@link Message} by provided {@code id}.
* @param id The message identifier.
* @return The MessageMetadata with the given id, or <i>null</i>
* if no Message with that id exists in the MessageStore
* or the message has no metadata (legacy message from an earlier version).
* @since 5.0
*/
MessageMetadata getMessageMetadata(UUID id);
/**
* Put the provided Message into the MessageStore. The store may need to mutate the message internally, and if it
* does then the return value can be different than the input. The id of the return value will be used as an index

13
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -191,6 +191,19 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
return (key != null) ? this.idToMessage.get(key) : null;
}
@Override
public MessageMetadata getMessageMetadata(UUID id) {
Message<?> message = getMessage(id);
if (message != null) {
MessageMetadata messageMetadata = new MessageMetadata(id);
messageMetadata.setTimestamp(message.getHeaders().getTimestamp());
return messageMetadata;
}
else {
return null;
}
}
@Override
public Message<?> removeMessage(UUID key) {
if (key != null) {

14
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -60,7 +60,8 @@ public class ConcurrentAggregatorTests {
@Before
public void configureAggregator() {
this.taskExecutor = new SimpleAsyncTaskExecutor();
this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), store);
this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), this.store);
this.aggregator.setReleaseStrategy(new SimpleSequenceSizeReleaseStrategy());
}
@@ -267,6 +268,9 @@ public class ConcurrentAggregatorTests {
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, null);
Message<?> message4 = createMessage(7, "ABC", 3, 3, replyChannel, null);
CountDownLatch latch = new CountDownLatch(4);
this.aggregator.setReleaseStrategy(new SequenceSizeReleaseStrategy());
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator,
@@ -278,9 +282,9 @@ public class ConcurrentAggregatorTests {
assertTrue(latch.await(10, TimeUnit.SECONDS));
Message<?> reply = replyChannel.receive(1000);
Message<?> reply = replyChannel.receive(10000);
assertNotNull("A message should be aggregated", reply);
assertThat(((Integer) reply.getPayload()), is(105));
assertThat(reply.getPayload(), is(105));
}
@@ -325,13 +329,13 @@ public class ConcurrentAggregatorTests {
this.aggregator.handleMessage(message);
}
catch (Exception e) {
e.printStackTrace();
this.exception = e;
}
finally {
this.latch.countDown();
}
}
}
@@ -345,6 +349,7 @@ public class ConcurrentAggregatorTests {
}
return product;
}
}
@@ -355,6 +360,7 @@ public class ConcurrentAggregatorTests {
public Object processMessageGroup(MessageGroup group) {
return null;
}
}
}