From a1f554c04d956d029230c53ad0bf2e01c0354491 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 29 Sep 2016 18:57:16 -0400 Subject: [PATCH] INT-2460: Remove Message Modification Logic in MS 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 --- .../store/AbstractKeyValueMessageStore.java | 102 +++++++++--------- .../store/AbstractMessageGroupStore.java | 29 +---- .../integration/store/MessageHolder.java | 57 ++++++++++ .../integration/store/MessageMetadata.java | 52 +++++++++ .../integration/store/MessageStore.java | 13 ++- .../integration/store/SimpleMessageStore.java | 13 +++ .../aggregator/ConcurrentAggregatorTests.java | 14 ++- .../gemfire/store/GemfireMessageStore.java | 11 ++ .../integration/jdbc/JdbcMessageStore.java | 94 ++++++++-------- .../jdbc/store/JdbcChannelMessageStore.java | 58 +++++----- .../jdbc/JdbcMessageStoreTests.java | 8 +- .../mysql/MySqlJdbcMessageStoreTests.java | 5 +- .../AbstractJdbcChannelMessageStoreTests.java | 3 - .../metadata/MongoDbMetadataStore.java | 59 +++------- ...stractConfigurableMongoDbMessageStore.java | 61 ++++++----- .../ConfigurableMongoDbMessageStore.java | 6 +- .../mongodb/store/MessageDocument.java | 18 +++- .../store/MongoDbChannelMessageStore.java | 5 +- .../mongodb/store/MongoDbMessageStore.java | 62 ++++++----- .../metadata/MongoDbMetadataStoreTests.java | 13 +-- .../AbstractMongoDbMessageStoreTests.java | 9 ++ .../redis/store/RedisMessageStore.java | 32 ++++-- src/reference/asciidoc/mongodb.adoc | 5 +- 23 files changed, 436 insertions(+), 293 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/store/MessageHolder.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/store/MessageMetadata.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index 0755d54d87..d959a18fee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -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 Message addMessage(Message message) { Assert.notNull(message, "'message' must not be null"); UUID messageId = message.getHeaders().getId(); - doStore(MESSAGE_KEY_PREFIX + messageId, message); - return (Message) getRawMessage(messageId); + doStoreIfAbsent(MESSAGE_KEY_PREFIX + messageId, new MessageHolder(message)); + return (Message) 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 { private final Iterator idIterator; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index 638decb3e0..75582be2fa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -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, BeanFactoryAware { + implements MessageGroupStore, Iterable { 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageHolder.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageHolder.java new file mode 100644 index 0000000000..7066ba5e66 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageHolder.java @@ -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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageMetadata.java new file mode 100644 index 0000000000..8ab176be3b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageMetadata.java @@ -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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java index 9a0d1c11a8..514540e450 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java @@ -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 null + * 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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 4801444ae0..147bb5c248 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -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) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index 6178d6e881..5a0ad740a5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java @@ -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; } + } } diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java index 92f30e44c2..70f4fa2323 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java @@ -106,6 +106,17 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements this.messageStoreRegion.put(id, objectToStore); } + @Override + protected void doStoreIfAbsent(Object id, Object objectToStore) { + Assert.notNull(id, "'id' must not be null"); + Assert.notNull(objectToStore, "'objectToStore' must not be null"); + Object present = this.messageStoreRegion.putIfAbsent(id, objectToStore); + if (present != null && logger.isDebugEnabled()) { + logger.debug("The message: [" + present + "] is already present in the store. " + + "The [" + objectToStore + "] is ignored."); + } + } + @Override protected Object doRemove(Object id) { Assert.notNull(id, "'id' must not be null"); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 87173a360d..928a11a6e1 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -36,7 +36,6 @@ import javax.sql.DataSource; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -46,6 +45,7 @@ import org.springframework.dao.DuplicateKeyException; import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageMetadata; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.util.UUIDConverter; @@ -61,7 +61,6 @@ import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -178,12 +177,16 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * The name of the message header that stores a flag to indicate that the message has been saved. This is an * optimization for the put method. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String SAVED_KEY = JdbcMessageStore.class.getSimpleName() + ".SAVED"; /** * The name of the message header that stores a timestamp for the time the message was inserted. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String CREATED_DATE_KEY = JdbcMessageStore.class.getSimpleName() + ".CREATED_DATE"; private final MessageMapper mapper = new MessageMapper(); @@ -326,42 +329,49 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } @Override - @SuppressWarnings({ "rawtypes", "unchecked" }) + public MessageMetadata getMessageMetadata(UUID id) { + List list = + this.jdbcTemplate.query(getQuery(Query.GET_MESSAGE), + (rs, rn) -> { + MessageMetadata messageMetadata = + new MessageMetadata(UUID.fromString(rs.getString("MESSAGE_ID"))); + messageMetadata.setTimestamp(rs.getTimestamp("CREATED_DATE").getTime()); + return messageMetadata; + }, getKey(id), this.region); + if (list.isEmpty()) { + return null; + } + return list.get(0); + } + + @Override + @SuppressWarnings("unchecked") public Message addMessage(final Message message) { - if (message.getHeaders().containsKey(SAVED_KEY)) { - Message saved = (Message) getMessage(message.getHeaders().getId()); - if (saved != null) { - if (saved.equals(message)) { - return message; - } // We need to save it under its own id - } + UUID id = message.getHeaders().getId(); + final String messageId = getKey(id); + final byte[] messageBytes = this.serializer.convert(message); + + if (logger.isDebugEnabled()) { + logger.debug("Inserting message with id key=" + messageId); } - final long createdDate = System.currentTimeMillis(); - Message result = this.getMessageBuilderFactory().fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE) - .setHeader(CREATED_DATE_KEY, createdDate).build(); - - Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers"); - // using reflection to set ID since it is immutable through MessageHeaders - innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID)); - - final String messageId = getKey(result.getHeaders().getId()); - final byte[] messageBytes = this.serializer.convert(result); - - this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE), new PreparedStatementSetter() { - - @Override - public void setValues(PreparedStatement ps) throws SQLException { - if (logger.isDebugEnabled()) { - logger.debug("Inserting message with id key=" + messageId); - } + try { + this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE), ps -> { ps.setString(1, messageId); - ps.setString(2, JdbcMessageStore.this.region); - ps.setTimestamp(3, new Timestamp(createdDate)); - JdbcMessageStore.this.lobHandler.getLobCreator().setBlobAsBytes(ps, 4, messageBytes); + ps.setString(2, this.region); + ps.setTimestamp(3, new Timestamp(System.currentTimeMillis())); + + this.lobHandler.getLobCreator().setBlobAsBytes(ps, 4, messageBytes); + }); + } + catch (DuplicateKeyException e) { + if (logger.isDebugEnabled()) { + logger.debug("The Message with id [" + id + "] already exists.\n" + + "Ignoring INSERT and SELECT existing..."); } - }); - return result; + return (Message) getMessage(id); + } + return message; } @Override @@ -442,19 +452,14 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final AtomicReference completeFlag = new AtomicReference(); final AtomicReference lastReleasedSequenceRef = new AtomicReference(); - this.jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new RowCallbackHandler() { + this.jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), rs -> { + updateDate.set(rs.getTimestamp("UPDATED_DATE")); - @Override - public void processRow(ResultSet rs) throws SQLException { - updateDate.set(rs.getTimestamp("UPDATED_DATE")); + createDate.set(rs.getTimestamp("CREATED_DATE")); - createDate.set(rs.getTimestamp("CREATED_DATE")); - - completeFlag.set(rs.getInt("COMPLETE") > 0); - - lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); - } + completeFlag.set(rs.getInt("COMPLETE") > 0); + lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); }, key, this.region); if (createDate.get() == null && updateDate.get() == null) { @@ -751,7 +756,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override public Message mapRow(ResultSet rs, int rowNum) throws SQLException { - return (Message) JdbcMessageStore.this.deserializer.convert(JdbcMessageStore.this.lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES")); + byte[] messageBytes = JdbcMessageStore.this.lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES"); + return (Message) JdbcMessageStore.this.deserializer.convert(messageBytes); } } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java index 41d2c2f9cb..0a97fd0e61 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java @@ -16,8 +16,6 @@ package org.springframework.integration.jdbc.store; -import java.sql.PreparedStatement; -import java.sql.SQLException; import java.sql.Types; import java.util.HashMap; import java.util.HashSet; @@ -35,7 +33,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; @@ -43,6 +40,7 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.core.serializer.support.DeserializingConverter; import org.springframework.core.serializer.support.SerializingConverter; +import org.springframework.dao.DuplicateKeyException; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.jdbc.JdbcMessageStore; import org.springframework.integration.jdbc.store.channel.ChannelMessageStoreQueryProvider; @@ -61,7 +59,6 @@ import org.springframework.integration.transaction.TransactionSynchronizationFac import org.springframework.integration.util.UUIDConverter; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.core.PreparedStatementSetter; import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.jdbc.support.lob.DefaultLobHandler; @@ -70,7 +67,6 @@ import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -127,12 +123,16 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto /** * The name of the message header that stores a flag to indicate that the message has been saved. This is an * optimization for the put method. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String SAVED_KEY = JdbcChannelMessageStore.class.getSimpleName() + ".SAVED"; /** * The name of the message header that stores a timestamp for the time the message was inserted. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String CREATED_DATE_KEY = JdbcChannelMessageStore.class.getSimpleName() + ".CREATED_DATE"; private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); @@ -399,7 +399,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto } if (this.jdbcTemplate.getFetchSize() != 1 && logger.isWarnEnabled()) { - logger.warn("The jdbcTemplate's fetchsize is not 1. This may cause FIFO issues with Oracle databases."); + logger.warn("The jdbcTemplate's fetch size is not 1. This may cause FIFO issues with Oracle databases."); } if (this.beanFactory != null) { @@ -419,36 +419,31 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto * @param message a message */ @Override - @SuppressWarnings({"rawtypes", "unchecked"}) + @SuppressWarnings("unchecked") public MessageGroup addMessageToGroup(Object groupId, final Message message) { - final String groupKey = getKey(groupId); + String groupKey = getKey(groupId); - final long createdDate = System.currentTimeMillis(); - final Message result = this.messageBuilderFactory.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE) - .setHeader(CREATED_DATE_KEY, createdDate).build(); + long createdDate = System.currentTimeMillis(); - final Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers"); - // using reflection to set ID since it is immutable through MessageHeaders - innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID)); + String messageId = getKey(message.getHeaders().getId()); - final String messageId = getKey(result.getHeaders().getId()); - final byte[] messageBytes = this.serializer.convert(result); + byte[] messageBytes = this.serializer.convert(message); - this.jdbcTemplate.update(getQuery(this.channelMessageStoreQueryProvider.getCreateMessageQuery()), - new PreparedStatementSetter() { + if (logger.isDebugEnabled()) { + logger.debug("Inserting message with id key=" + messageId); + } - @Override - public void setValues(PreparedStatement ps) throws SQLException { - if (logger.isDebugEnabled()) { - logger.debug("Inserting message with id key=" + messageId); - } + try { + this.jdbcTemplate.update(getQuery(this.channelMessageStoreQueryProvider.getCreateMessageQuery()), + ps -> { ps.setString(1, messageId); ps.setString(2, groupKey); - ps.setString(3, JdbcChannelMessageStore.this.region); + ps.setString(3, this.region); ps.setLong(4, createdDate); - Integer priority = new IntegrationMessageHeaderAccessor(message).getPriority(); + Integer priority = message.getHeaders() + .get(IntegrationMessageHeaderAccessor.PRIORITY, Integer.class); if (JdbcChannelMessageStore.this.priorityEnabled && priority != null) { ps.setInt(5, priority); @@ -457,10 +452,15 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto ps.setNull(5, Types.NUMERIC); } - JdbcChannelMessageStore.this.lobHandler.getLobCreator().setBlobAsBytes(ps, 6, messageBytes); - } - - }); + this.lobHandler.getLobCreator().setBlobAsBytes(ps, 6, messageBytes); + }); + } + catch (DuplicateKeyException e) { + if (logger.isDebugEnabled()) { + logger.debug("The Message with id [" + messageId + "] already exists.\n" + + "Ignoring INSERT..."); + } + } return getMessageGroup(groupId); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index 8956215f1b..be5ffbdcf6 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -20,7 +20,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders; @@ -101,12 +100,9 @@ public class JdbcMessageStoreTests { public void testAddAndGet() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); Message saved = messageStore.addMessage(message); - assertNotNull(messageStore.getMessage(message.getHeaders().getId())); Message result = messageStore.getMessage(saved.getHeaders().getId()); assertNotNull(result); assertThat(saved, sameExceptIgnorableHeaders(result)); - assertNotNull(result.getHeaders().get(JdbcMessageStore.SAVED_KEY)); - assertNotNull(result.getHeaders().get(JdbcMessageStore.CREATED_DATE_KEY)); } @Test @@ -186,7 +182,7 @@ public class JdbcMessageStoreTests { Message message = MessageBuilder.withPayload("foo").build(); message = messageStore.addMessage(message); Message result = messageStore.addMessage(message); - assertSame(message, result); + assertEquals(message, result); } @Test @@ -207,7 +203,7 @@ public class JdbcMessageStoreTests { Message copy = MessageBuilder.fromMessage(saved).setHeader("newHeader", 1).build(); Message result = messageStore.addMessage(copy); assertNotSame(saved, result); - assertThat(saved, sameExceptIgnorableHeaders(result, JdbcMessageStore.CREATED_DATE_KEY, "newHeader")); + assertThat(saved, sameExceptIgnorableHeaders(result, "newHeader")); assertNotNull(messageStore.getMessage(saved.getHeaders().getId())); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java index 0f2267ff12..f2eec6e9f1 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java @@ -139,12 +139,9 @@ public class MySqlJdbcMessageStoreTests { public void testAddAndGet() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); Message saved = messageStore.addMessage(message); - assertNotNull(messageStore.getMessage(message.getHeaders().getId())); Message result = messageStore.getMessage(saved.getHeaders().getId()); assertNotNull(result); assertThat(saved, sameExceptIgnorableHeaders(result)); - assertNotNull(result.getHeaders().get(JdbcMessageStore.SAVED_KEY)); - assertNotNull(result.getHeaders().get(JdbcMessageStore.CREATED_DATE_KEY)); } @Test @@ -252,7 +249,7 @@ public class MySqlJdbcMessageStoreTests { Message copy = MessageBuilder.fromMessage(saved).setHeader("newHeader", 1).build(); Message result = messageStore.addMessage(copy); assertNotSame(saved, result); - assertThat(saved, sameExceptIgnorableHeaders(result, JdbcMessageStore.CREATED_DATE_KEY, "newHeader")); + assertThat(saved, sameExceptIgnorableHeaders(result, "newHeader")); assertNotNull(messageStore.getMessage(saved.getHeaders().getId())); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java index ee2f586fc3..3c237bbb2f 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractJdbcChannelMessageStoreTests.java @@ -98,9 +98,6 @@ public abstract class AbstractJdbcChannelMessageStoreTests { assertNotNull(messageFromDb); assertEquals(message.getHeaders().getId(), messageFromDb.getHeaders().getId()); - - assertNotNull(messageFromDb.getHeaders().get(JdbcChannelMessageStore.SAVED_KEY)); - assertNotNull(messageFromDb.getHeaders().get(JdbcChannelMessageStore.CREATED_DATE_KEY)); } } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStore.java index 26517bd43c..1ba0107a8b 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-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. @@ -19,21 +19,17 @@ package org.springframework.integration.mongodb.metadata; import java.util.HashMap; import java.util.Map; -import org.springframework.dao.DataAccessException; import org.springframework.data.mongodb.MongoDbFactory; -import org.springframework.data.mongodb.core.CollectionCallback; +import org.springframework.data.mongodb.core.FindAndModifyOptions; import org.springframework.data.mongodb.core.MongoTemplate; -import org.springframework.data.mongodb.core.ScriptOperations; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; -import org.springframework.data.mongodb.core.script.NamedMongoScript; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.util.Assert; import com.mongodb.BasicDBObject; import com.mongodb.DBCollection; -import com.mongodb.MongoException; /** * MongoDbMetadataStore implementation of {@link ConcurrentMetadataStore}. @@ -54,26 +50,10 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore { private static final String VALUE = "value"; - private static final String PUT_IF_ABSENT_FUNCTION = - "function putIfAbsent(collection, key, value){ " + - " var alreadyPresent = db[collection].findOne({\"_id\": key}, {\"_id\": 0}); " + - " if(alreadyPresent == null){" + - " db[collection].insert({\"_id\": key, \"value\": value}); " + - " return null; " + - " }" + - " return alreadyPresent;" + - "}"; - - private static final String PUT_IF_ABSENT_SCRIPT_NAME = "metadataStorePutIfAbsent"; - private final MongoTemplate template; private final String collectionName; - private final ScriptOperations scriptOperations; - - private volatile boolean scriptInitialized; - /** * Configure the MongoDbMetadataStore by provided {@link MongoDbFactory} and * default collection name - {@link #DEFAULT_COLLECTION_NAME}. @@ -112,7 +92,6 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore { Assert.hasText(collectionName, "'collectionName' must not be empty."); this.template = template; this.collectionName = collectionName; - this.scriptOperations = template.scriptOps(); } /** @@ -123,24 +102,17 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore { * If a document exists with the specified {@code key}, the method performs an {@code update}. * @param key the metadata entry key * @param value the metadata entry value - * @see MongoTemplate#execute(String, CollectionCallback) + * @see MongoTemplate#execute(String, org.springframework.data.mongodb.core.CollectionCallback) * @see DBCollection#save */ @Override public void put(String key, String value) { Assert.hasText(key, "'key' must not be empty."); Assert.hasText(value, "'value' must not be empty."); - final Map entry = new HashMap(); + final Map entry = new HashMap<>(); entry.put(ID_FIELD, key); entry.put(VALUE, value); - this.template.execute(this.collectionName, (CollectionCallback) new CollectionCallback() { - - @Override - public Object doInCollection(DBCollection collection) throws MongoException, DataAccessException { - return collection.save(new BasicDBObject(entry)); - } - - }); + this.template.execute(this.collectionName, collection -> collection.save(new BasicDBObject(entry))); } /** @@ -188,29 +160,22 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore { * } * except that the action is performed atomically. *

- * Performs the {@code stored} JavaScript function. * @param key the metadata entry key * @param value the metadata entry value to store * @return null if successful, the old value otherwise. * @see java.util.concurrent.ConcurrentMap#putIfAbsent(Object, Object) - * @see ScriptOperations#call(String, Object...) */ @Override public String putIfAbsent(String key, String value) { Assert.hasText(key, "'key' must not be empty."); Assert.hasText(value, "'value' must not be empty."); - if (!this.scriptInitialized) { - synchronized (this) { - if (!this.scriptInitialized) { - this.scriptOperations.register( - new NamedMongoScript(PUT_IF_ABSENT_SCRIPT_NAME, PUT_IF_ABSENT_FUNCTION)); - this.scriptInitialized = true; - } - } - } - BasicDBObject result = - (BasicDBObject) this.scriptOperations.call(PUT_IF_ABSENT_SCRIPT_NAME, this.collectionName, key, value); - return (result == null) ? null : (String) result.get(VALUE); + + Query query = new Query(Criteria.where(ID_FIELD).is(key)); + query.fields().exclude(ID_FIELD); + @SuppressWarnings("unchecked") + Map result = this.template.findAndModify(query, new Update().setOnInsert(VALUE, value), + new FindAndModifyOptions().upsert(true), Map.class, this.collectionName); + return result == null ? null : result.get(VALUE); } /** diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java index 4c7c04e9db..ce45cce81f 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java @@ -27,10 +27,10 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; +import org.springframework.dao.DuplicateKeyException; import org.springframework.data.domain.Sort; import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.core.FindAndModifyOptions; @@ -48,11 +48,11 @@ import org.springframework.integration.mongodb.support.MongoDbMessageBytesConver import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.BasicMessageGroupStore; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageMetadata; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; /** @@ -71,12 +71,16 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe /** * The name of the message header that stores a flag to indicate that the message has been saved. This is an * optimization for the put method. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String SAVED_KEY = "MongoDbMessageStore.SAVED"; /** * The name of the message header that stores a timestamp for the time the message was inserted. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String CREATED_DATE_KEY = "MongoDbMessageStore.CREATED_DATE"; protected final Log logger = LogFactory.getLog(this.getClass()); @@ -140,6 +144,10 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe indexOperations.ensureIndex(new Index(MessageDocumentFields.MESSAGE_ID, Sort.Direction.ASC)); + indexOperations.ensureIndex(new Index(MessageDocumentFields.GROUP_ID, Sort.Direction.ASC) + .on(MessageDocumentFields.MESSAGE_ID, Sort.Direction.ASC) + .unique()); + indexOperations.ensureIndex(new Index(MessageDocumentFields.GROUP_ID, Sort.Direction.ASC) .on(MessageDocumentFields.LAST_MODIFIED_TIME, Sort.Direction.DESC) .on(MessageDocumentFields.SEQUENCE, Sort.Direction.DESC)); @@ -152,6 +160,20 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe return document != null ? document.getMessage() : null; } + public MessageMetadata getMessageMetadata(UUID id) { + Assert.notNull(id, "'id' must not be null"); + Query query = Query.query(Criteria.where(MessageDocumentFields.MESSAGE_ID).is(id)); + MessageDocument document = this.mongoTemplate.findOne(query, MessageDocument.class, this.collectionName); + if (document != null) { + MessageMetadata messageMetadata = new MessageMetadata(id); + messageMetadata.setTimestamp(document.getCreatedTime()); + return messageMetadata; + } + else { + return null; + } + } + @Override public void removeMessageGroup(Object groupId) { this.mongoTemplate.remove(groupIdQuery(groupId), this.collectionName); @@ -181,32 +203,19 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } protected void addMessageDocument(final MessageDocument document) { - Message message = document.getMessage(); - if (message.getHeaders().containsKey(SAVED_KEY)) { - Message saved = getMessage(message.getHeaders().getId()); - if (saved != null) { - if (saved.equals(message)) { - return; - } // We need to save it under its own id + if (document.getGroupCreatedTime() == 0) { + document.setGroupCreatedTime(System.currentTimeMillis()); + } + document.setCreatedTime(System.currentTimeMillis()); + try { + this.mongoTemplate.insert(document, this.collectionName); + } + catch (DuplicateKeyException e) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("The Message with id [" + document.getMessageId() + "] already exists.\n" + + "Ignoring INSERT and SELECT existing..."); } } - - final long createdDate = document.getCreatedTime() == 0 - ? System.currentTimeMillis() - : document.getCreatedTime(); - - Message result = this.messageBuilderFactory.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE) - .setHeader(CREATED_DATE_KEY, createdDate).build(); - - @SuppressWarnings("unchecked") - Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()) - .getPropertyValue("headers"); - // using reflection to set ID since it is immutable through MessageHeaders - innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID)); - innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().get(MessageHeaders.TIMESTAMP)); - - document.setCreatedTime(createdDate); - this.mongoTemplate.insert(document, this.collectionName); } protected static Query groupIdQuery(Object groupId) { diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java index 4a267d8f05..58ce4adf15 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java @@ -145,7 +145,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb MessageDocument messageDocument = this.mongoTemplate.findOne(query, MessageDocument.class, this.collectionName); if (messageDocument != null) { - long createdTime = messageDocument.getCreatedTime(); + long createdTime = messageDocument.getGroupCreatedTime(); long lastModifiedTime = messageDocument.getLastModifiedTime(); boolean complete = messageDocument.isComplete(); int lastReleasedSequence = messageDocument.getLastReleasedSequence(); @@ -181,7 +181,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb boolean complete = false; if (messageDocument != null) { - createdTime = messageDocument.getCreatedTime(); + createdTime = messageDocument.getGroupCreatedTime(); lastReleasedSequence = messageDocument.getLastReleasedSequence(); complete = messageDocument.isComplete(); } @@ -191,7 +191,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb document.setGroupId(groupId); document.setComplete(complete); document.setLastReleasedSequence(lastReleasedSequence); - document.setCreatedTime(createdTime); + document.setGroupCreatedTime(createdTime); document.setLastModifiedTime(messageDocument == null ? createdTime : System.currentTimeMillis()); document.setSequence(getNextId()); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java index 3e8b635e51..f177f27850 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java @@ -50,6 +50,8 @@ public class MessageDocument { private Long createdTime = 0L; + private Long groupCreatedTime = 0L; + private Object groupId; private Long lastModifiedTime = 0L; @@ -70,6 +72,10 @@ public class MessageDocument { return this.message; } + public UUID getMessageId() { + return this.messageId; + } + public void setGroupId(Object groupId) { this.groupId = groupId; } @@ -94,6 +100,14 @@ public class MessageDocument { this.createdTime = createdTime; } + public Long getGroupCreatedTime() { + return this.groupCreatedTime; + } + + public void setGroupCreatedTime(long groupCreatedTime) { + this.groupCreatedTime = groupCreatedTime; + } + public Boolean isComplete() { return this.complete; } @@ -114,10 +128,6 @@ public class MessageDocument { this.sequence = sequence; } - public UUID getMessageId() { - return this.messageId; - } - public Integer getPriority() { return this.priority; } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java index 64815010a1..7f157bf9b6 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java @@ -72,7 +72,8 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa this(mongoDbFactory, null, collectionName); } - public MongoDbChannelMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, String collectionName) { + public MongoDbChannelMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, + String collectionName) { super(mongoDbFactory, mappingMongoConverter, collectionName); } @@ -105,7 +106,7 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa document.setCreatedTime(System.currentTimeMillis()); document.setLastModifiedTime(System.currentTimeMillis()); if (this.priorityEnabled) { - document.setPriority(new IntegrationMessageHeaderAccessor(message).getPriority()); + document.setPriority(message.getHeaders().get(IntegrationMessageHeaderAccessor.PRIORITY, Integer.class)); } document.setSequence(this.getNextId()); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index 528833d4bc..5c34b373c0 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -59,6 +59,7 @@ import org.springframework.integration.message.AdviceMessage; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.MessageMetadata; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MutableMessage; @@ -100,12 +101,16 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore /** * The name of the message header that stores a flag to indicate that the message has been saved. This is an * optimization for the put method. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String SAVED_KEY = ConfigurableMongoDbMessageStore.class.getSimpleName() + ".SAVED"; /** * The name of the message header that stores a timestamp for the time the message was inserted. + * @deprecated since 5.0. This constant isn't used any more. */ + @Deprecated public static final String CREATED_DATE_KEY = ConfigurableMongoDbMessageStore.class.getSimpleName() + ".CREATED_DATE"; private final static String GROUP_ID_KEY = "_groupId"; @@ -187,33 +192,16 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore return message; } - private void addMessageDocument(final MessageWrapper document) { - Message message = document.getMessage(); - if (message.getHeaders().containsKey(SAVED_KEY)) { - Message saved = getMessage(message.getHeaders().getId()); - if (saved != null) { - if (saved.equals(message)) { - return; - } // We need to save it under its own id + private void addMessageDocument(MessageWrapper document) { + UUID messageId = (UUID) document.headers.get(MessageHeaders.ID); + Query query = whereMessageIdIsAndGroupIdIs(messageId, document.get_GroupId()); + if (!this.template.exists(query, MessageWrapper.class, this.collectionName)) { + if (document.get_Group_timestamp() == 0) { + document.set_Group_timestamp(System.currentTimeMillis()); } + document.set_message_timestamp(System.currentTimeMillis()); + this.template.insert(document, this.collectionName); } - - final long createdDate = document.get_Group_timestamp() == 0 - ? System.currentTimeMillis() - : document.get_Group_timestamp(); - - Message result = getMessageBuilderFactory().fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE) - .setHeader(CREATED_DATE_KEY, createdDate).build(); - - @SuppressWarnings("unchecked") - Map innerMap = - (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers"); - // using reflection to set ID since it is immutable through MessageHeaders - innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID)); - innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().get(MessageHeaders.TIMESTAMP)); - - document.set_Group_timestamp(createdDate); - this.template.insert(document, this.collectionName); } @Override @@ -224,6 +212,20 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore return (messageWrapper != null) ? messageWrapper.getMessage() : null; } + public MessageMetadata getMessageMetadata(UUID id) { + Assert.notNull(id, "'id' must not be null"); + MessageWrapper messageWrapper = + this.template.findOne(whereMessageIdIs(id), MessageWrapper.class, this.collectionName); + if (messageWrapper != null) { + MessageMetadata messageMetadata = new MessageMetadata(id); + messageMetadata.setTimestamp(messageWrapper.get_message_timestamp()); + return messageMetadata; + } + else { + return null; + } + } + @Override @ManagedAttribute public long getMessageCount() { @@ -768,6 +770,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore @SuppressWarnings("unused") private final Message inputMessage; + private long _message_timestamp; + private volatile long _group_timestamp; private volatile long _group_update_timestamp; @@ -822,6 +826,14 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore this._group_timestamp = groupTimestamp; } + public long get_message_timestamp() { + return this._message_timestamp; + } + + public void set_message_timestamp(long _message_timestamp) { + this._message_timestamp = _message_timestamp; + } + public long get_Group_update_timestamp() { return this._group_update_timestamp; } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStoreTests.java index 9295e03bf4..8ad91ad23b 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-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. @@ -32,6 +32,7 @@ import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; /** * @author Senthil Arumugam, Samiraj Panneer Selvam + * @author Artem Bilan * @since 4.2 * */ @@ -99,16 +100,16 @@ public class MongoDbMetadataStoreTests extends MongoDbAvailableTests { @MongoDbAvailable public void testPutIfAbsent() throws Exception { String fileID = store.get(file1); - assertNull("Get First time, Key doesnt exists", fileID); + assertNull("Get First time, Value must not exist", fileID); fileID = store.putIfAbsent(file1, file1Id); - assertNull("Insert First time, Key insertion successful", fileID); + assertNull("Insert First time, Value must return null", fileID); fileID = store.putIfAbsent(file1, "56789"); - assertNotNull("Key Already Exists - Insertion Failed, for different value", fileID); - assertEquals("Retrieving the Old Value", file1Id, fileID); + assertNotNull("Key Already Exists - Insertion Failed, ol value must be returned", fileID); + assertEquals("The Old Value must be equal to returned", file1Id, fileID); - assertEquals("Retrieving the Old Value", file1Id, store.get(file1)); + assertEquals("The Old Value must return", file1Id, store.get(file1)); } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java index 6cef43a77c..2a08271fa3 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java @@ -264,6 +264,15 @@ public abstract class AbstractMongoDbMessageStoreTests extends MongoDbAvailableT assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); } + @Test + @MongoDbAvailable + public void testAddAndUpdateAlreadySaved() throws Exception { + MessageStore messageStore = getMessageStore(); + Message message = MessageBuilder.withPayload("foo").build(); + message = messageStore.addMessage(message); + Message result = messageStore.addMessage(message); + assertEquals(message, result); + } public static class Foo implements Serializable { /** diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java index 94e7a203d1..75c151d472 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java @@ -17,7 +17,6 @@ package org.springframework.integration.redis.store; import java.util.Collection; -import java.util.Set; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundValueOperations; @@ -72,12 +71,34 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore { ops.set(objectToStore); } catch (SerializationException e) { - throw new IllegalArgumentException("If relying on the default RedisSerializer (JdkSerializationRedisSerializer) " + - "the Object must be Serializable. Either make it Serializable or provide your own implementation of " + - "RedisSerializer via 'setValueSerializer(..)'", e); + rethrowAsIllegalArgumentException(e); + } } + @Override + protected void doStoreIfAbsent(Object id, Object objectToStore) { + Assert.notNull(id, "'id' must not be null"); + Assert.notNull(objectToStore, "'objectToStore' must not be null"); + BoundValueOperations ops = this.redisTemplate.boundValueOps(id); + try { + Boolean present = ops.setIfAbsent(objectToStore); + if (present != null && logger.isDebugEnabled()) { + logger.debug("The message: [" + present + "] is already present in the store. " + + "The [" + objectToStore + "] is ignored."); + } + } + catch (SerializationException e) { + rethrowAsIllegalArgumentException(e); + } + } + + private void rethrowAsIllegalArgumentException(SerializationException e) { + throw new IllegalArgumentException("If relying on the default RedisSerializer " + + "(JdkSerializationRedisSerializer) the Object must be Serializable. " + + "Either make it Serializable or provide your own implementation of " + + "RedisSerializer via 'setValueSerializer(..)'", e); + } @Override protected Object doRemove(Object id) { @@ -93,7 +114,6 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore { @Override protected Collection doListKeys(String keyPattern) { Assert.hasText(keyPattern, "'keyPattern' must not be empty"); - Set keys = this.redisTemplate.keys(keyPattern); - return keys; + return this.redisTemplate.keys(keyPattern); } } diff --git a/src/reference/asciidoc/mongodb.adoc b/src/reference/asciidoc/mongodb.adoc index 671e4c5fca..7fb884f833 100644 --- a/src/reference/asciidoc/mongodb.adoc +++ b/src/reference/asciidoc/mongodb.adoc @@ -167,10 +167,7 @@ public MetadataStore metadataStore(MongoDbFactory factory) { The `MongoDbMetadataStore` also implements `ConcurrentMetadataStore`, allowing it to be reliably shared across multiple application instances where only one instance will be allowed to store or modify a key's value. -All these operations are _atomic_ via MongoDB guarantees. For this purpose the `putIfAbsent` operation is implemented -as a _stored_ JavaScript function. Fom more information see -http://docs.spring.io/spring-data/data-mongo/docs/current/reference/html/#mongo.server-side-scripts[Script Operations] -and `MongoDbMetadataStore` JavaDocs. +All these operations are _atomic_ via MongoDB guarantees. [[mongodb-inbound-channel-adapter]] === MongoDB Inbound Channel Adapter