From f0f2c41ae3eaabd82ce1a02cff097901cfa45ae3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 19 Jan 2021 18:40:10 -0500 Subject: [PATCH] GH-3446: Stream support in the MessageGroupStore Fixes https://github.com/spring-projects/spring-integration/issues/3446 * For better resources utilization provide a `Stream>` API on the `MessageGroupStore`, `MessageGroup` and `MessageGroupQueue` * Use this API in the `DelayHandler` when it reschedules persisted messages --- .../integration/handler/DelayHandler.java | 16 +++++----- .../store/AbstractKeyValueMessageStore.java | 15 ++++++++-- .../integration/store/MessageGroup.java | 20 +++++++++---- .../integration/store/MessageGroupQueue.java | 30 ++++++++++--------- .../integration/store/MessageGroupStore.java | 13 +++++++- .../store/PersistentMessageGroup.java | 25 +++++++++++++++- .../config/xml/DelayerParserTests.java | 19 ++++++------ .../jdbc/store/JdbcMessageStore.java | 26 ++++++++-------- .../ConfigurableMongoDbMessageStore.java | 23 +++++++++++--- .../mongodb/store/MongoDbMessageStore.java | 14 ++++++++- src/reference/asciidoc/message-store.adoc | 25 ++++++++-------- src/reference/asciidoc/whats-new.adoc | 3 ++ 12 files changed, 157 insertions(+), 72 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 1f14f860e0..0a5273a823 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; import org.aopalliance.aop.Advice; @@ -321,7 +322,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement } /** - * Checks if 'requestMessage' wasn't delayed before ({@link #releaseMessageAfterDelay} + * Check if 'requestMessage' wasn't delayed before ({@link #releaseMessageAfterDelay} * and {@link DelayHandler.DelayedMessageWrapper}). Than determine 'delay' for * 'requestMessage' ({@link #determineDelayForMessage}) and if {@code delay > 0} * schedules 'releaseMessage' task after 'delay'. @@ -562,9 +563,10 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement @Override public synchronized void reschedulePersistedMessages() { MessageGroup messageGroup = this.messageStore.getMessageGroup(this.messageGroupId); - for (final Message message : messageGroup.getMessages()) { - getTaskScheduler() - .schedule(() -> { + try (Stream> messageStream = messageGroup.streamMessages()) { + TaskScheduler taskScheduler = getTaskScheduler(); + messageStream.forEach((message) -> + taskScheduler.schedule(() -> { // This is fine to keep the reference to the message, // because the scheduled task is performed immediately. long delay = determineDelayForMessage(message); @@ -574,12 +576,12 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement else { releaseMessage(message); } - }, new Date()); + }, new Date())); } } /** - * Handles {@link ContextRefreshedEvent} to invoke + * Handle {@link ContextRefreshedEvent} to invoke * {@link #reschedulePersistedMessages} as late as possible after application context * startup. Also it checks {@link #initialized} to ignore other * {@link ContextRefreshedEvent}s which may be published in the 'parent-child' 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 93f931a415..c43c105fa1 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -24,6 +24,7 @@ import java.util.List; import java.util.Set; import java.util.UUID; import java.util.stream.Collectors; +import java.util.stream.Stream; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; @@ -160,7 +161,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override @ManagedAttribute public long getMessageCount() { - Collection messageIds = doListKeys(this.messagePrefix + "*"); + Collection messageIds = doListKeys(this.messagePrefix + '*'); return (messageIds != null) ? messageIds.size() : 0; } @@ -346,11 +347,19 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS return messages; } + @Override + public Stream> streamMessagesForGroup(Object groupId) { + return getGroupMetadata(groupId) + .getMessageIds() + .stream() + .map(this::getMessage); + } + @Override @SuppressWarnings("unchecked") public Iterator iterator() { final Iterator idIterator = normalizeKeys( - (Collection) doListKeys(this.groupPrefix + "*")) + (Collection) doListKeys(this.groupPrefix + '*')) .iterator(); return new MessageGroupIterator(idIterator); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java index a6d0cf8773..c8fdd35f39 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -17,14 +17,15 @@ package org.springframework.integration.store; import java.util.Collection; +import java.util.stream.Stream; import org.springframework.messaging.Message; /** * A group of messages that are correlated with each other and should be processed in the same context. *

- * The message group allows implementations to be mutable, but this behavior is optional. Implementations should take - * care to document their thread safety and mutability. + * The message group allows implementations to be mutable, but this behavior is optional. + * Implementations should take care to document their thread safety and mutability. * * @author Dave Syer * @author Oleg Zhurakousky @@ -35,7 +36,6 @@ public interface MessageGroup { /** * Query if the message can be added. - * * @param message The message. * @return true if the message can be added. */ @@ -57,12 +57,20 @@ public interface MessageGroup { boolean remove(Message messageToRemove); /** - * Returns all available Messages from the group at the time of invocation - * + * Return all available Messages from the group at the time of invocation * @return The messages. */ Collection> getMessages(); + /** + * Return a stream for messages stored in this group. + * @return the {@link Stream} for messages in this group. + * @since 5.5 + */ + default Stream> streamMessages() { + return getMessages().stream(); + } + /** * @return the key that links these messages together */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java index 332bba48c5..9b1a4e5499 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -42,6 +43,7 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gunnar Hillert * @author Gary Russell + * @author Artem Bilan * * @since 2.0 * @@ -118,7 +120,7 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc @Override public Iterator> iterator() { - return getMessages().iterator(); + return stream().iterator(); } /** @@ -164,29 +166,24 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc @Override public Message peek() { - Message message = null; - final Lock lock = this.storeLock; try { - lock.lockInterruptibly(); - try { - Collection> messages = getMessages(); - if (!messages.isEmpty()) { - message = messages.iterator().next(); - } + this.storeLock.lockInterruptibly(); + try (Stream> messageStream = stream()) { + return messageStream.findFirst().orElse(null); } finally { - lock.unlock(); + this.storeLock.unlock(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } - return message; + return null; } @Override public Message poll(long timeout, TimeUnit unit) throws InterruptedException { - Message message = null; + Message message; long timeoutInNanos = unit.toNanos(timeout); final Lock lock = this.storeLock; lock.lockInterruptibly(); @@ -325,7 +322,7 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc @Override public Message take() throws InterruptedException { - Message message = null; + Message message; final Lock lock = this.storeLock; lock.lockInterruptibly(); @@ -346,6 +343,11 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc return this.messageGroupStore.getMessageGroup(this.groupId).getMessages(); } + @Override + public Stream> stream() { + return this.messageGroupStore.getMessageGroup(this.groupId).streamMessages(); + } + /** * It is assumed that the 'storeLock' is being held by the caller, otherwise * IllegalMonitorStateException may be thrown diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index ef8a199de8..4d0af294e3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -18,6 +18,7 @@ package org.springframework.integration.store; import java.util.Collection; import java.util.Iterator; +import java.util.stream.Stream; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; @@ -142,6 +143,16 @@ public interface MessageGroupStore extends BasicMessageGroupStore { */ Collection> getMessagesForGroup(Object groupId); + /** + * Return a stream for messages stored in the provided group. + * @param groupId the group id to retrieve messages. + * @return the {@link Stream} for messages in this group. + * @since 5.5 + */ + default Stream> streamMessagesForGroup(Object groupId) { + return getMessagesForGroup(groupId).stream(); + } + /** * Invoked when a MessageGroupStore expires a group. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java index a9827abf55..c1a76133c8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -20,6 +20,8 @@ import java.util.AbstractCollection; import java.util.Collection; import java.util.Collections; import java.util.Iterator; +import java.util.Spliterator; +import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -29,6 +31,7 @@ import org.springframework.messaging.Message; /** * @author Artem Bilan + * * @since 4.3 */ class PersistentMessageGroup implements MessageGroup { @@ -59,6 +62,16 @@ class PersistentMessageGroup implements MessageGroup { return Collections.unmodifiableCollection(this.messages); } + /** + * The resulting {@link Stream} must be closed after use, + * it can be declared as a resource in a {@code try-with-resources} statement. + * @return the stream of messages in this group. + */ + @Override + public Stream> streamMessages() { + return this.messageGroupStore.streamMessagesForGroup(this.original.getGroupId()); + } + @Override public Message getOne() { if (this.oneMessage == null) { @@ -224,6 +237,16 @@ class PersistentMessageGroup implements MessageGroup { return PersistentMessageGroup.this.size(); } + @Override + public Spliterator> spliterator() { + return streamMessages().spliterator(); + } + + @Override + public Stream> stream() { + return streamMessages(); + } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java index 77ff5be58d..96950cbf29 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -22,8 +22,7 @@ import java.lang.reflect.Method; import java.util.HashMap; import java.util.List; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; @@ -35,8 +34,7 @@ import org.springframework.integration.handler.DelayHandler; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; import org.springframework.transaction.interceptor.NameMatchTransactionAttributeSource; @@ -51,8 +49,7 @@ import org.springframework.transaction.interceptor.TransactionInterceptor; * * @since 1.0.3 */ -@RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration +@SpringJUnitConfig public class DelayerParserTests { @Autowired @@ -70,7 +67,7 @@ public class DelayerParserTests { .isEqualTo("headers.foo"); assertThat(TestUtils.getPropertyValue(delayHandler, "messagingTemplate.sendTimeout", Long.class)) .isEqualTo(987L); - assertThat(TestUtils.getPropertyValue(delayHandler, "taskScheduler")).isNull(); + assertThat(TestUtils.getPropertyValue(delayHandler, "taskScheduler")).isNotNull(); } @Test @@ -110,7 +107,8 @@ public class DelayerParserTests { assertThat(adviceChain.size()).isEqualTo(1); Object advice = adviceChain.get(0); assertThat(advice instanceof TransactionInterceptor).isTrue(); - TransactionAttributeSource transactionAttributeSource = ((TransactionInterceptor) advice).getTransactionAttributeSource(); + TransactionAttributeSource transactionAttributeSource = + ((TransactionInterceptor) advice).getTransactionAttributeSource(); assertThat(transactionAttributeSource instanceof MatchAlwaysTransactionAttributeSource).isTrue(); Method method = MessageHandler.class.getMethod("handleMessage", Message.class); TransactionDefinition definition = transactionAttributeSource.getTransactionAttribute(method, null); @@ -130,7 +128,8 @@ public class DelayerParserTests { Object txAdvice = adviceChain.get(1); assertThat(txAdvice.getClass()).isEqualTo(TransactionInterceptor.class); - TransactionAttributeSource transactionAttributeSource = ((TransactionInterceptor) txAdvice).getTransactionAttributeSource(); + TransactionAttributeSource transactionAttributeSource = + ((TransactionInterceptor) txAdvice).getTransactionAttributeSource(); assertThat(transactionAttributeSource.getClass()).isEqualTo(NameMatchTransactionAttributeSource.class); HashMap nameMap = TestUtils.getPropertyValue(transactionAttributeSource, "nameMap", HashMap.class); assertThat(nameMap.toString()).isEqualTo("{*=PROPAGATION_REQUIRES_NEW,ISOLATION_DEFAULT,readOnly}"); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java index bf63c2b060..2c0193c739 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; import javax.sql.DataSource; @@ -152,7 +153,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa LIST_GROUP_KEYS("SELECT distinct GROUP_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?"); - private String sql; + private final String sql; Query(String sql) { this.sql = sql; @@ -202,7 +203,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * Public setter for the table prefix property. This will be prefixed to all the table names before queries are * executed. Defaults to {@link #DEFAULT_TABLE_PREFIX}. - * * @param tablePrefix the tablePrefix to set */ public void setTablePrefix(String tablePrefix) { @@ -212,7 +212,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * A unique grouping identifier for all messages persisted with this store. Using multiple regions allows the store * to be partitioned (if necessary) for different purposes. Defaults to DEFAULT. - * * @param region the region name to set */ public void setRegion(String region) { @@ -223,7 +222,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * Override the {@link LobHandler} that is used to create and unpack large objects in SQL queries. The default is * fine for almost all platforms, but some Oracle drivers require a native implementation. - * * @param lobHandler a {@link LobHandler} */ public void setLobHandler(LobHandler lobHandler) { @@ -232,7 +230,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * A converter for serializing messages to byte arrays for storage. - * * @param serializer the serializer to set */ @SuppressWarnings("unchecked") @@ -242,7 +239,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * A converter for deserializing byte arrays to messages. - * * @param deserializer the deserializer to set */ @SuppressWarnings({"unchecked", "rawtypes"}) @@ -410,10 +406,10 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override public MessageGroup getMessageGroup(Object groupId) { String key = getKey(groupId); - final AtomicReference createDate = new AtomicReference(); - final AtomicReference updateDate = new AtomicReference(); - final AtomicReference completeFlag = new AtomicReference(); - final AtomicReference lastReleasedSequenceRef = new AtomicReference(); + final AtomicReference createDate = new AtomicReference<>(); + final AtomicReference updateDate = new AtomicReference<>(); + final AtomicReference completeFlag = new AtomicReference<>(); + final AtomicReference lastReleasedSequenceRef = new AtomicReference<>(); this.jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), rs -> { updateDate.set(rs.getTimestamp("UPDATED_DATE")); @@ -547,6 +543,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa this.region, this.region); } + @Override + public Stream> streamMessagesForGroup(Object groupId) { + return this.jdbcTemplate.queryForStream(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), this.mapper, + getKey(groupId), this.region, this.region); + } + @Override public Iterator iterator() { @@ -570,6 +572,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public void remove() { throw new UnsupportedOperationException("Cannot remove MessageGroup from this iterator."); } + }; } @@ -577,7 +580,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * Replace patterns in the input to produce a valid SQL query. This implementation lazily initializes a * simple map-based cache, only replacing the table prefix on the first access to a named query. Further * accesses will be resolved from the cache. - * * @param base the SQL query to be transformed * @return a transformed query with replacements */ 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 db44866921..de3bcbc7c2 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2020 the original author or authors. + * Copyright 2013-2021 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. @@ -23,6 +23,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.stream.Collectors; +import java.util.stream.Stream; import java.util.stream.StreamSupport; import org.springframework.data.domain.Sort; @@ -74,7 +75,9 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb this(mongoDbFactory, null, DEFAULT_COLLECTION_NAME); } - public ConfigurableMongoDbMessageStore(MongoDatabaseFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter) { + public ConfigurableMongoDbMessageStore(MongoDatabaseFactory mongoDbFactory, + MappingMongoConverter mappingMongoConverter) { + this(mongoDbFactory, mappingMongoConverter, DEFAULT_COLLECTION_NAME); } @@ -82,8 +85,8 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb this(mongoDbFactory, null, collectionName); } - public ConfigurableMongoDbMessageStore(MongoDatabaseFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, - String collectionName) { + public ConfigurableMongoDbMessageStore(MongoDatabaseFactory mongoDbFactory, + MappingMongoConverter mappingMongoConverter, String collectionName) { super(mongoDbFactory, mappingMongoConverter, collectionName); } @@ -278,6 +281,18 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb .collect(Collectors.toList()); } + @Override + public Stream> streamMessagesForGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Query query = groupOrderQuery(groupId); + Stream documents = + getMongoTemplate() + .stream(query, MessageDocument.class, this.collectionName) + .stream(); + + return documents.map(MessageDocument::getMessage); + } + private void updateGroup(Object groupId, Update update) { getMongoTemplate() .findAndModify(groupOrderQuery(groupId), update, FindAndModifyOptions.none(), Map.class, 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 2a9277a206..ec150a4b12 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -28,6 +28,7 @@ import java.util.Map.Entry; import java.util.Properties; import java.util.UUID; import java.util.stream.Collectors; +import java.util.stream.Stream; import org.bson.Document; import org.bson.conversions.Bson; @@ -426,6 +427,17 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore .collect(Collectors.toList()); } + @Override + public Stream> streamMessagesForGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + Query query = whereGroupIdOrder(groupId); + Stream messageWrappers = + this.template.stream(query, MessageWrapper.class, this.collectionName) + .stream(); + + return messageWrappers.map(MessageWrapper::getMessage); + } + @Override @ManagedAttribute public int getMessageCountForAllMessageGroups() { diff --git a/src/reference/asciidoc/message-store.adoc b/src/reference/asciidoc/message-store.adoc index b830201482..1081de2f43 100644 --- a/src/reference/asciidoc/message-store.adoc +++ b/src/reference/asciidoc/message-store.adoc @@ -100,12 +100,9 @@ For this reason, you should either not perform such manipulation or set the `cop [[message-group-factory]] ==== Using `MessageGroupFactory` -Starting with version 4.3, some `MessageGroupStore` implementations can be injected with a custom -`MessageGroupFactory` strategy to create and customize the `MessageGroup` instances used by the `MessageGroupStore`. -This defaults to a `SimpleMessageGroupFactory`, which produces `SimpleMessageGroup` instances based on the `GroupType.HASH_SET` -(`LinkedHashSet`) internal collection. -Other possible options are `SYNCHRONISED_SET` and `BLOCKING_QUEUE`, where the last one can be used to reinstate the -previous `SimpleMessageGroup` behavior. +Starting with version 4.3, some `MessageGroupStore` implementations can be injected with a custom `MessageGroupFactory` strategy to create and customize the `MessageGroup` instances used by the `MessageGroupStore`. +This defaults to a `SimpleMessageGroupFactory`, which produces `SimpleMessageGroup` instances based on the `GroupType.HASH_SET` (`LinkedHashSet`) internal collection. +Other possible options are `SYNCHRONISED_SET` and `BLOCKING_QUEUE`, where the last one can be used to reinstate the previous `SimpleMessageGroup` behavior. Also the `PERSISTENT` option is available. See the next section for more information. Starting with version 5.0.1, the `LIST` option is also available for when the order and uniqueness of messages in the group does not matter. @@ -113,16 +110,12 @@ Starting with version 5.0.1, the `LIST` option is also available for when the or [[lazy-load-message-group]] ==== Persistent `MessageGroupStore` and Lazy-load -Starting with version 4.3, all persistent `MessageGroupStore` instances retrieve `MessageGroup` instances and their `messages` -from the store in the lazy-load manner. -In most cases, it is useful for the correlation `MessageHandler` instances (see <<./aggregator.adoc#aggregator,Aggregator>> and <<./resequencer.adoc#resequencer,Resequencer>>), -when it would add overhead to load entire the `MessageGroup` from the store on each correlation operation. +Starting with version 4.3, all persistent `MessageGroupStore` instances retrieve `MessageGroup` instances and their `messages` from the store in the lazy-load manner. +In most cases, it is useful for the correlation `MessageHandler` instances (see <<./aggregator.adoc#aggregator,Aggregator>> and <<./resequencer.adoc#resequencer,Resequencer>>), when it would add overhead to load entire the `MessageGroup` from the store on each correlation operation. You can use the `AbstractMessageGroupStore.setLazyLoadMessageGroups(false)` option to switch off the lazy-load behavior from the configuration. -Our performance tests for lazy-load on MongoDB `MessageStore` (<<./mongodb.adoc#mongodb-message-store,MongoDB Message Store>>) and -`` (<<./aggregator.adoc#aggregator,Aggregator>>) -use a custom `release-strategy` similar to the following: +Our performance tests for lazy-load on MongoDB `MessageStore` (<<./mongodb.adoc#mongodb-message-store,MongoDB Message Store>>) and `` (<<./aggregator.adoc#aggregator,Aggregator>>) use a custom `release-strategy` similar to the following: ==== [source,xml] @@ -149,3 +142,9 @@ ms % Task name ... ---- ==== + +However starting with version 5.5, all the persistent `MessageGroupStore` implementations provide a `streamMessagesForGroup(Object groupId)` contract based on the target database streaming API. +This improves resources utilization when groups are very big in the store. +Internally in the framework this new API is used in the <<./delayer.adoc#delayer,Delayer>> (for example) when it reschedules persisted messages on startup. +A returned `Stream>` must be closed in the end of processing, e.g. via auto-close by the `try-with-resources`. +Whenever a `PersistentMessageGroup` is used, its `streamMessages()` delegates to the `MessageGroupStore.streamMessagesForGroup()`. \ No newline at end of file diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d930a0ac91..a5608e706a 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -18,6 +18,9 @@ If you are interested in more details, see the Issue Tracker tickets that were r [[x5.5-general]] === General Changes + +All the persistent `MessageGroupStore` implementation provide a `streamMessagesForGroup(Object groupId)` contract based on the target database streaming API. +See <<./message-store.adoc#message-store,Message Store>> for more information. [[x5.5-amqp]] ==== AMQP Changes