From 7e9552974cd4e2533e0f00fdc7889eac43415235 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 23 Mar 2021 14:23:46 -0400 Subject: [PATCH] Introduce a `MessageGroup.condition` (#3517) * Introduce a `groupConditionSupplier` for MGS * Add a `MessageGroup.condition` option * Add a `MessageGroupStore.conditionSupplier` option * Use it from the `SimpleMessageStore.addMessagesToGroup()` API to populate `condition` (if any) into a `MessageGroup` * Introduce a `GroupConditionProvider` contract to be implemented on those `ReleaseStrategy` contracts which could be aware of group condition * Populated a `GroupConditionProvider.getGroupConditionSupplier()` into a `MessageGroupStore` from the `AbstractCorrelatingMessageHandler` for end-user convenience * Rework a `FileMarkerReleaseStrategy` to implement a `GroupConditionProvider` to provide a function which produces a condition from `file_lineCount` header of the `END` marker message * Make the `FileMarkerReleaseStrategy` logic already based on the condition from a group * Delegate `GroupConditionProvider` from the `FileAggregator` * Add test for empty file aggregation * * Implement `condition` in the `AbstractKeyValueMessageStore` and `MongoDbMessageStore` * Test `condition` for `mongo-aggregator-config.xml` and `FileAggregatorTests` against GemFire * * Implement `condition` in the `ConfigurableMongoDbMessageStore` * * Implement `condition` in the `JdbcMessageStore` * `FileAggregatorTests` against `JdbcMessageStore` * Refactor `JdbcMessageStore` for better handling of message group metadata * Remove unused `MARKED` column in the DDL in favor of newly introduced `CONDITION` * * Add docs for message group condition * * Move `conditionSupplier` option from MGS to AbstractCorrelatingMH * Make it as a `BiFunction` to propagate existing condition alongside with the message to consult * Expose `groupConditionSupplier` in Java & XML DSLs * * Fix language in docs --- .../AbstractCorrelatingMessageHandler.java | 33 ++++ .../aggregator/GroupConditionProvider.java | 39 +++++ .../config/AggregatorFactoryBean.java | 9 ++ ...stractCorrelatingMessageHandlerParser.java | 4 +- .../dsl/CorrelationHandlerSpec.java | 13 ++ .../store/AbstractKeyValueMessageStore.java | 10 ++ .../store/AbstractMessageGroupStore.java | 8 +- .../integration/store/MessageGroup.java | 16 ++ .../store/MessageGroupMetadata.java | 10 ++ .../integration/store/MessageGroupStore.java | 11 ++ .../store/PersistentMessageGroup.java | 12 ++ .../integration/store/SimpleMessageGroup.java | 21 ++- .../integration/store/SimpleMessageStore.java | 52 +++--- .../integration/config/spring-integration.xsd | 13 ++ .../file/aggregator/FileAggregator.java | 13 +- .../aggregator/FileMarkerReleaseStrategy.java | 43 +++-- .../file/aggregator/FileAggregatorTests.java | 59 ++++++- .../file/aggregator/FileAggregatorTests.xml | 14 +- .../jdbc/store/JdbcMessageStore.java | 151 +++++++++--------- .../integration/jdbc/schema-db2.sql | 2 +- .../integration/jdbc/schema-derby.sql | 2 +- .../integration/jdbc/schema-h2.sql | 2 +- .../integration/jdbc/schema-hsqldb.sql | 2 +- .../integration/jdbc/schema-mysql.sql | 2 +- .../integration/jdbc/schema-oracle.sql | 2 +- .../integration/jdbc/schema-postgresql.sql | 2 +- .../integration/jdbc/schema-sqlserver.sql | 2 +- .../integration/jdbc/schema-sybase.sql | 2 +- ...stractConfigurableMongoDbMessageStore.java | 7 +- .../ConfigurableMongoDbMessageStore.java | 14 +- .../mongodb/store/MessageDocument.java | 14 +- .../mongodb/store/MongoDbMessageStore.java | 29 +++- ...AbstractMongoDbMessageGroupStoreTests.java | 9 +- .../mongodb/store/mongo-aggregator-config.xml | 15 +- .../mongo-aggregator-configurable-config.xml | 13 +- src/reference/asciidoc/aggregator.adoc | 8 + src/reference/asciidoc/file.adoc | 2 + src/reference/asciidoc/message-store.adoc | 20 ++- src/reference/asciidoc/whats-new.adoc | 4 + 39 files changed, 533 insertions(+), 151 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aggregator/GroupConditionProvider.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 1bdd4c8a6b..cc70c079cb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -27,6 +27,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.locks.Lock; +import java.util.function.BiFunction; import org.aopalliance.aop.Advice; @@ -158,6 +159,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private volatile boolean running; + private BiFunction, String, String> groupConditionSupplier; + public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { @@ -361,6 +364,17 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.expireDuration = expireDuration; } + /** + * Configure a {@link BiFunction} to supply a group condition from a message to be added to the group. + * The {@code null} result from the function will reset a condition set before. + * @param conditionSupplier the function to supply a group condition from a message to be added to the group. + * @since 5.5 + * @see GroupConditionProvider + */ + public void setGroupConditionSupplier(BiFunction, String, String> conditionSupplier) { + this.groupConditionSupplier = conditionSupplier; + } + @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; @@ -407,6 +421,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP */ this.lockRegistrySet = true; this.forceReleaseProcessor = createGroupTimeoutProcessor(); + + if (this.releaseStrategy instanceof GroupConditionProvider) { + this.groupConditionSupplier = ((GroupConditionProvider) this.releaseStrategy).getGroupConditionSupplier(); + } } private MessageGroupProcessor createGroupTimeoutProcessor() { @@ -441,6 +459,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP return this.releaseStrategy; } + @Nullable + protected BiFunction, String, String> getGroupConditionSupplier() { + return this.groupConditionSupplier; + } + @Override public MessageChannel getDiscardChannel() { String channelName = this.discardChannelName; @@ -542,6 +565,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.logger.trace(() -> "Adding message to group [ " + messageGroupToLog + "]"); messageGroup = store(correlationKey, message); + setGroupConditionIfAny(message, messageGroup); + if (this.releaseStrategy.canRelease(messageGroup)) { Collection> completedMessages = null; try { @@ -579,6 +604,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } + private void setGroupConditionIfAny(Message message, MessageGroup messageGroup) { + if (this.groupConditionSupplier != null) { + String condition = this.groupConditionSupplier.apply(message, messageGroup.getCondition()); + this.messageStore.setGroupCondition(messageGroup.getGroupId(), condition); + messageGroup.setCondition(condition); + } + } + protected boolean isExpireGroupsUponCompletion() { return false; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/GroupConditionProvider.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/GroupConditionProvider.java new file mode 100644 index 0000000000..848b3a0160 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/GroupConditionProvider.java @@ -0,0 +1,39 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://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.aggregator; + +import java.util.function.BiFunction; + +import org.springframework.messaging.Message; + +/** + * A contract which can be implemented on the {@link ReleaseStrategy} + * and used in the {@link AbstractCorrelatingMessageHandler} to + * populate the provided group condition supplier. + * + * @author Artem Bilan + * + * @since 5.5 + * + * @see AbstractCorrelatingMessageHandler#setGroupConditionSupplier(BiFunction) + */ +@FunctionalInterface +public interface GroupConditionProvider { + + BiFunction, String, String> getGroupConditionSupplier(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index 44fb0e934c..9c04d3c10d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -19,6 +19,7 @@ package org.springframework.integration.config; import java.time.Duration; import java.util.List; import java.util.Map; +import java.util.function.BiFunction; import java.util.function.Function; import org.aopalliance.aop.Advice; @@ -36,6 +37,7 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.util.JavaUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.scheduling.TaskScheduler; @@ -98,6 +100,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe private Function> headersFunction; + private BiFunction, String, String> groupConditionSupplier; + public void setProcessorBean(Object processorBean) { this.processorBean = processorBean; } @@ -187,6 +191,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe this.expireDuration = expireDuration; } + public void setGroupConditionSupplier(BiFunction, String, String> groupConditionSupplier) { + this.groupConditionSupplier = groupConditionSupplier; + } + @Override protected AggregatingMessageHandler createHandler() { MessageGroupProcessor outputProcessor; @@ -233,6 +241,7 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe .acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend) .acceptIfNotNull(this.expireDuration, (duration) -> aggregator.setExpireDuration(Duration.ofMillis(duration))) + .acceptIfNotNull(this.groupConditionSupplier, aggregator::setGroupConditionSupplier) .acceptIfNotNull(this.expireTimeout, aggregator::setExpireTimeout); return aggregator; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index 7d03453310..1f404930a1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.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. @@ -104,6 +104,8 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expire-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expire-duration", "expireDurationMillis"); + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "group-condition-supplier"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index fe7d46b0b7..45c3b73760 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java @@ -20,6 +20,7 @@ import java.time.Duration; import java.util.Arrays; import java.util.LinkedList; import java.util.List; +import java.util.function.BiFunction; import java.util.function.Function; import org.aopalliance.aop.Advice; @@ -36,6 +37,7 @@ import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; @@ -397,4 +399,15 @@ public abstract class CorrelationHandlerSpec, String, String> conditionSupplier) { + this.handler.setGroupConditionSupplier(conditionSupplier); + return _this(); + } + } 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 c43c105fa1..a61f5a9adf 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 @@ -180,6 +180,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS .create(this, groupId, metadata.getTimestamp(), metadata.isComplete()); messageGroup.setLastModified(metadata.getLastModified()); messageGroup.setLastReleasedMessageSequenceNumber(metadata.getLastReleasedMessageSequenceNumber()); + messageGroup.setCondition(metadata.getCondition()); return messageGroup; } else { @@ -294,6 +295,15 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } } + @Override + public void setGroupCondition(Object groupId, String condition) { + MessageGroupMetadata metadata = getGroupMetadata(groupId); + if (metadata != null) { + metadata.setCondition(condition); + doStore(this.groupPrefix + groupId, metadata); + } + } + @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); 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 65451cb5a9..bb9c7cc727 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 @@ -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. @@ -47,10 +47,10 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG private final MessageGroupFactory persistentMessageGroupFactory = new SimpleMessageGroupFactory(SimpleMessageGroupFactory.GroupType.PERSISTENT); - private volatile boolean timeoutOnIdle; - private boolean lazyLoadMessageGroups = true; + private boolean timeoutOnIdle; + protected AbstractMessageGroupStore() { } @@ -111,7 +111,7 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG this.expiryCallbacks.stream() .anyMatch(UniqueExpiryCallback.class::isInstance); - if (uniqueExpiryCallbackPresent && this.logger.isErrorEnabled()) { + if (uniqueExpiryCallbackPresent) { this.logger.error("Only one instance of 'UniqueExpiryCallback' can be registered in the " + "'MessageGroupStore'. Use a separate 'MessageGroupStore' for each aggregator/resequencer."); } 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 c8fdd35f39..d4f1aea565 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 @@ -19,6 +19,7 @@ package org.springframework.integration.store; import java.util.Collection; import java.util.stream.Stream; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -120,6 +121,21 @@ public interface MessageGroup { void setLastModified(long lastModified); + /** + * Add a condition statement to this group which can be consulted later on, e.g. from the release strategy. + * @param condition statement which could be consulted later on, e.g. from the release strategy. + * @since 5.5 + */ + void setCondition(String condition); + + /** + * Return the condition for this group to consult with, e.g. from the release strategy. + * @return the condition for this group to consult with, e.g. from the release strategy. + * @since 5.5 + */ + @Nullable + String getCondition(); + void clear(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java index a8660990dd..b71ce93f8a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java @@ -50,6 +50,8 @@ public class MessageGroupMetadata implements Serializable { private volatile int lastReleasedMessageSequenceNumber; + private volatile String condition; + private MessageGroupMetadata() { //For Jackson deserialization } @@ -129,4 +131,12 @@ public class MessageGroupMetadata implements Serializable { this.lastReleasedMessageSequenceNumber = lastReleasedMessageSequenceNumber; } + public String getCondition() { + return this.condition; + } + + public void setCondition(String condition) { + this.condition = condition; + } + } 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 4d0af294e3..90591e6193 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 @@ -96,6 +96,17 @@ public interface MessageGroupStore extends BasicMessageGroupStore { */ void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber); + /** + * Add a condition sentence into the group. + * Can be used later on for making some decisions for group, e.g. release strategy + * for correlation handler can consult this condition instead of iterating all + * the messages in group. + * @param groupId The group identifier. + * @param condition The condition to store into the group. + * @since 5.5 + */ + void setGroupCondition(Object groupId, String condition); + /** * @return The iterator of currently accumulated {@link MessageGroup}s. */ 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 c1a76133c8..4e1cd00efb 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 @@ -27,6 +27,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -175,6 +176,17 @@ class PersistentMessageGroup implements MessageGroup { this.original.setLastReleasedMessageSequenceNumber(sequenceNumber); } + @Override + public void setCondition(String condition) { + this.original.setCondition(condition); + } + + @Override + @Nullable + public String getCondition() { + return this.original.getCondition(); + } + @Override public void clear() { this.original.clear(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index 282fb99cba..dc48b6e725 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.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. @@ -24,12 +24,13 @@ import java.util.LinkedHashSet; import java.util.Set; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; /** * Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and group id. - * The group will grow during its lifetime, when messages are added to it. + * The group will grow during its lifetime, when messages are {@link #add}ed to it. * This MessageGroup is thread safe. * * @author Iwein Fuld @@ -56,8 +57,11 @@ public class SimpleMessageGroup implements MessageGroup { private volatile boolean complete; + @Nullable + private volatile String condition; + public SimpleMessageGroup(Object groupId) { - this(Collections.>emptyList(), groupId); + this(Collections.emptyList(), groupId); } public SimpleMessageGroup(Collection> messages, Object groupId) { @@ -172,6 +176,17 @@ public class SimpleMessageGroup implements MessageGroup { return this.messages.size(); } + @Override + public void setCondition(String condition) { + this.condition = condition; + } + + @Override + @Nullable + public String getCondition() { + return this.condition; + } + @Override public Message getOne() { synchronized (this.messages) { 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 bbc021173a..9126b5a22f 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 @@ -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. @@ -56,12 +56,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore private static final String INTERRUPTED_WHILE_OBTAINING_LOCK = "Interrupted while obtaining lock"; - private final ConcurrentMap> idToMessage = new ConcurrentHashMap>(); + private final ConcurrentMap> idToMessage = new ConcurrentHashMap<>(); - private final ConcurrentMap groupIdToMessageGroup = - new ConcurrentHashMap(); + private final ConcurrentMap groupIdToMessageGroup = new ConcurrentHashMap<>(); - private final ConcurrentMap groupToUpperBound = new ConcurrentHashMap(); + private final ConcurrentMap groupToUpperBound = new ConcurrentHashMap<>(); private final int groupCapacity; @@ -69,14 +68,14 @@ public class SimpleMessageStore extends AbstractMessageGroupStore private final UpperBound individualUpperBound; - private volatile LockRegistry lockRegistry; + private final long upperBoundTimeout; + + private LockRegistry lockRegistry; + + private boolean copyOnGet = false; private volatile boolean isUsed; - private volatile boolean copyOnGet = false; - - private final long upperBoundTimeout; - /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. The capacities are applied independently to messages stored via @@ -183,7 +182,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore public Message addMessage(Message message) { this.isUsed = true; if (!this.individualUpperBound.tryAcquire(this.upperBoundTimeout)) { - throw new MessagingException(this.getClass().getSimpleName() + throw new MessagingException(getClass().getSimpleName() + " was out of capacity (" + this.individualCapacity + "), try constructing it with a larger capacity."); @@ -254,6 +253,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore .create(group.getMessages(), groupId, group.getTimestamp(), group.isComplete()); simpleMessageGroup.setLastModified(group.getLastModified()); simpleMessageGroup.setLastReleasedMessageSequenceNumber(group.getLastReleasedMessageSequenceNumber()); + simpleMessageGroup.setCondition(group.getCondition()); return simpleMessageGroup; } finally { @@ -353,8 +353,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.lockInterruptibly(); try { MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to remove Message(s) from the MessageGroup"); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to remove Message(s) from the MessageGroup"); UpperBound upperBound = this.groupToUpperBound.get(groupId); Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); boolean modified = false; @@ -380,7 +381,15 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override public Iterator iterator() { - return new HashSet(this.groupIdToMessageGroup.values()).iterator(); + return new HashSet<>(this.groupIdToMessageGroup.values()).iterator(); + } + + @Override + public void setGroupCondition(Object groupId, String condition) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + if (group != null) { + group.setCondition(condition); + } } @Override @@ -390,8 +399,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.lockInterruptibly(); try { MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to set 'lastReleasedSequenceNumber'"); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to set 'lastReleasedSequenceNumber'"); group.setLastReleasedMessageSequenceNumber(sequenceNumber); group.setLastModified(System.currentTimeMillis()); } @@ -412,8 +422,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.lockInterruptibly(); try { MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to complete the MessageGroup"); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to complete the MessageGroup"); group.complete(); group.setLastModified(System.currentTimeMillis()); } @@ -466,8 +477,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.lockInterruptibly(); try { MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to complete the MessageGroup"); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to complete the MessageGroup"); group.clear(); group.setLastModified(System.currentTimeMillis()); UpperBound upperBound = this.groupToUpperBound.get(groupId); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd index e1e67d69c1..9f9f9931d7 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd @@ -4071,6 +4071,19 @@ + + + + + + + + + A reference to a 'java.util.function.BiFunction' bean that may provide a group + condition evaluated against the message to be added to the group. + + + diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java index 34ff8c8d2d..db8262c5d0 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java @@ -16,10 +16,13 @@ package org.springframework.integration.file.aggregator; +import java.util.function.BiFunction; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.GroupConditionProvider; import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy; import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.ReleaseStrategy; @@ -35,7 +38,7 @@ import org.springframework.messaging.Message; * Delegates to {@link HeaderAttributeCorrelationStrategy} with {@link FileHeaders#FILENAME} attribute, * {@link FileMarkerReleaseStrategy} and {@link FileAggregatingMessageGroupProcessor}, respectively. *

- * The default {@link FileSplitter} behavior with markers enabled is do not provide a sequence details + * The default {@link FileSplitter} behavior with markers enabled is about do not provide a sequence details * headers, therefore correlation in this aggregator implementation is done by the {@link FileHeaders#FILENAME} * header which is still populated by the {@link FileSplitter} for each line emitted, including * {@link FileSplitter.FileMarker} messages. @@ -47,7 +50,8 @@ import org.springframework.messaging.Message; * * @since 5.5 */ -public class FileAggregator implements CorrelationStrategy, ReleaseStrategy, MessageGroupProcessor, BeanFactoryAware { +public class FileAggregator implements CorrelationStrategy, ReleaseStrategy, GroupConditionProvider, + MessageGroupProcessor, BeanFactoryAware { private final CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(FileHeaders.FILENAME); @@ -70,6 +74,11 @@ public class FileAggregator implements CorrelationStrategy, ReleaseStrategy, Mes return this.releaseStrategy.canRelease(group); } + @Override + public BiFunction, String, String> getGroupConditionSupplier() { + return this.releaseStrategy.getGroupConditionSupplier(); + } + @Override public Object processMessageGroup(MessageGroup group) { return this.groupProcessor.processMessageGroup(group); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java index f08f77cdcb..559c2b4b1c 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java @@ -16,8 +16,10 @@ package org.springframework.integration.file.aggregator; -import java.util.Collection; +import java.util.function.BiFunction; +import java.util.function.Function; +import org.springframework.integration.aggregator.GroupConditionProvider; import org.springframework.integration.aggregator.ReleaseStrategy; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.splitter.FileSplitter; @@ -29,35 +31,46 @@ import org.springframework.messaging.MessageHeaders; * A {@link ReleaseStrategy} which makes a decision based on the presence of * {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mark#END} * message in the group and its {@link org.springframework.integration.file.FileHeaders#LINE_COUNT} header. + *

+ * The logic of this strategy is based on the {@link FileMarkerReleaseStrategy#GROUP_CONDITION} + * function populated to the {@link org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler}. * * @author Artem Bilan * * @since 5.5 */ -public class FileMarkerReleaseStrategy implements ReleaseStrategy { +public class FileMarkerReleaseStrategy implements ReleaseStrategy, GroupConditionProvider { + + /** + * The {@link Function} for + * {@link org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler#setGroupConditionSupplier(BiFunction)}. + */ + public static final BiFunction, String, String> GROUP_CONDITION = + (message, existingCondition) -> { + MessageHeaders headers = message.getHeaders(); + if (FileSplitter.FileMarker.Mark.END.name().equals(headers.get(FileHeaders.MARKER))) { + Long lineCount = headers.get(FileHeaders.LINE_COUNT, Long.class); + return lineCount != null ? "" + lineCount : existingCondition; + } + return existingCondition; + }; @Override public boolean canRelease(MessageGroup group) { int size = group.size(); if (size > 1) { // Need more than only a START marker - Collection> messages = group.getMessages(); - for (Message message : messages) { - if (checkForEndMarker(size, message.getHeaders())) { - return true; - } + String condition = group.getCondition(); + if (condition != null) { + long lineCount = Long.parseLong(condition); + return lineCount == size - 2; // line count doesn't include START/END markers. } } return false; } - private boolean checkForEndMarker(int groupSize, MessageHeaders headers) { - if (FileSplitter.FileMarker.Mark.END.name().equals(headers.get(FileHeaders.MARKER))) { - Long lineCount = headers.get(FileHeaders.LINE_COUNT, Long.class); - return lineCount != null && lineCount == groupSize - 2; - } - else { - return false; - } + @Override + public BiFunction, String, String> getGroupConditionSupplier() { + return GROUP_CONDITION; } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java index 205b650987..8cdd05254c 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.List; +import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -35,11 +36,16 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.dsl.Files; import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.jdbc.store.JdbcMessageStore; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; @@ -57,6 +63,8 @@ import org.springframework.util.FileCopyUtils; @DirtiesContext public class FileAggregatorTests { + static EmbeddedDatabase dataSource; + @TempDir static File tmpDir; @@ -66,6 +74,10 @@ public class FileAggregatorTests { @Qualifier("fileSplitterAggregatorFlow.input") MessageChannel fileSplitterAggregatorFlow; + @Autowired + @Qualifier("jdbcMessageStoreAggregatorFlow.input") + MessageChannel jdbcMessageStoreAggregatorFlow; + @Autowired PollableChannel resultChannel; @@ -84,6 +96,17 @@ public class FileAggregatorTests { "second line\n" + "last line"; FileCopyUtils.copy(content.getBytes(StandardCharsets.UTF_8), new FileOutputStream(file, false)); + + dataSource = new EmbeddedDatabaseBuilder() + .setType(EmbeddedDatabaseType.H2) + .addScript("classpath:/org/springframework/integration/jdbc/schema-drop-h2.sql") + .addScript("classpath:/org/springframework/integration/jdbc/schema-h2.sql") + .build(); + } + + @AfterAll + public static void destroy() { + dataSource.shutdown(); } @Test @@ -104,6 +127,25 @@ public class FileAggregatorTests { .contains("SECOND LINE", "LAST LINE", "FIRST LINE"); } + @Test + void testEmptyFileAggregator() throws IOException { + File file = new File(tmpDir, "empty.txt"); + file.createNewFile(); + this.jdbcMessageStoreAggregatorFlow.send(new GenericMessage<>(file)); + + Message receive = this.resultChannel.receive(10_000); + assertThat(receive).isNotNull(); + assertThat(receive.getHeaders()) + .containsEntry(FileHeaders.FILENAME, "empty.txt") + .containsEntry(FileHeaders.LINE_COUNT, 0L) + .doesNotContainKey(IntegrationMessageHeaderAccessor.CORRELATION_ID); + + assertThat(receive.getPayload()) + .isInstanceOf(List.class) + .asList() + .isEmpty(); + } + @Test void testFileAggregatorXmlConfig() { this.input.send(new GenericMessage<>(file)); @@ -138,7 +180,22 @@ public class FileAggregatorTests { .transform(String::toUpperCase) .channel("aggregatorChannel") .aggregate(new FileAggregator()) - .channel(c -> c.queue("resultChannel")); + .channel(resultChannel()); + } + + @Bean + public IntegrationFlow jdbcMessageStoreAggregatorFlow() { + return f -> f + .split(Files.splitter().markers()) + .aggregate(aggregator -> + aggregator.processor(new FileAggregator()) + .messageStore(new JdbcMessageStore(dataSource))) + .channel(resultChannel()); + } + + @Bean + PollableChannel resultChannel() { + return new QueueChannel(); } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml index ea331d3747..bb528b2df4 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/aggregator/FileAggregatorTests.xml @@ -3,13 +3,23 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int-file="http://www.springframework.org/schema/integration/file" xmlns:int="http://www.springframework.org/schema/integration" + xmlns:gfe="http://www.springframework.org/schema/geode" xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd - http://www.springframework.org/schema/integration/file https://www.springframework.org/schema/integration/file/spring-integration-file.xsd"> + http://www.springframework.org/schema/integration/file https://www.springframework.org/schema/integration/file/spring-integration-file.xsd + http://www.springframework.org/schema/geode https://www.springframework.org/schema/geode/spring-geode.xsd"> + + + + + + + + - + 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 2c0193c739..dd9e741d70 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 @@ -21,13 +21,12 @@ import java.sql.SQLException; import java.sql.Timestamp; import java.util.Arrays; import java.util.Collection; -import java.util.Date; +import java.util.Collections; import java.util.HashMap; import java.util.Iterator; 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; @@ -36,6 +35,7 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.core.serializer.support.SerializingConverter; import org.springframework.dao.DuplicateKeyException; +import org.springframework.dao.IncorrectResultSizeDataAccessException; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageMetadata; @@ -56,7 +56,7 @@ import org.springframework.util.StringUtils; /** * Implementation of {@link MessageStore} using a relational database via JDBC. SQL scripts to create the necessary - * tables are packaged as org/springframework/integration/jdbc/schema-*.sql, where * is the + * tables are packaged as {@code org/springframework/integration/jdbc/schema-*.sql}, where {@code *} is the * target database type. *

* If you intend backing a {@link org.springframework.messaging.MessageChannel} @@ -83,13 +83,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public static final String DEFAULT_TABLE_PREFIX = "INT_"; private enum Query { - GROUP_EXISTS("SELECT COUNT(GROUP_KEY) FROM %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"), - CREATE_MESSAGE_GROUP("INSERT into %PREFIX%MESSAGE_GROUP" + - "(GROUP_KEY, REGION, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE)" - + " values (?, ?, 0, 0, 0, ?, ?)"), + "(GROUP_KEY, REGION, COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE)" + + " values (?, ?, 0, 0, ?, ?)"), - UPDATE_MESSAGE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"), + UPDATE_MESSAGE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, CONDITION=? " + + "where GROUP_KEY=? and REGION=?"), REMOVE_MESSAGE_FROM_GROUP("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=? and " + "REGION=?"), @@ -118,14 +117,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa "and %PREFIX%GROUP_TO_MESSAGE.GROUP_KEY = ? " + "and m.REGION = ?)"), - GET_GROUP_INFO("SELECT COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE" + - " from %PREFIX%MESSAGE_GROUP where GROUP_KEY = ? and REGION=?"), + GET_GROUP_INFO("SELECT COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE, CONDITION" + + " from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"), GET_MESSAGE("SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and " + "REGION=?"), - GET_GROUP_CREATED_DATE("SELECT CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"), - GET_MESSAGE_COUNT("SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE where REGION=?"), DELETE_MESSAGE("DELETE from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"), @@ -211,7 +208,7 @@ 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. + * to be partitioned (if necessary) for different purposes. Defaults to {@code DEFAULT}. * @param region the region name to set */ public void setRegion(String region) { @@ -241,7 +238,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * A converter for deserializing byte arrays to messages. * @param deserializer the deserializer to set */ - @SuppressWarnings({"unchecked", "rawtypes"}) + @SuppressWarnings({ "unchecked", "rawtypes" }) public void setDeserializer(Deserializer> deserializer) { this.deserializer = new AllowListDeserializingConverter((Deserializer) deserializer); } @@ -336,30 +333,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override public void addMessagesToGroup(Object groupId, Message... messages) { - final String groupKey = getKey(groupId); - boolean groupNotExist = this.jdbcTemplate - .queryForObject(this.getQuery(Query.GROUP_EXISTS), // NOSONAR query never returns null - Integer.class, groupKey, this.region) < 1; + String groupKey = getKey(groupId); + Map groupInfo = getGroupMetadata(groupKey); - final Timestamp updatedDate = new Timestamp(System.currentTimeMillis()); - - final Timestamp createdDate = groupNotExist ? - updatedDate : - this.jdbcTemplate.queryForObject(getQuery(Query.GET_GROUP_CREATED_DATE), Timestamp.class, groupKey, - this.region); - - if (groupNotExist) { - try { - doCreateMessageGroup(groupKey, createdDate); - } - catch (DuplicateKeyException e) { - logger.warn("Lost race to create group; attempting update instead", e); - doUpdateMessageGroup(groupKey, updatedDate); - } - } - else { - doUpdateMessageGroup(groupKey, updatedDate); - } + Timestamp updatedDate = new Timestamp(System.currentTimeMillis()); + boolean groupNotExist = groupInfo.isEmpty(); + Timestamp createdDate = groupNotExist ? updatedDate : (Timestamp) groupInfo.get("CREATED_DATE"); for (Message message : messages) { addMessage(message); @@ -377,6 +356,19 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa ps.setString(2, messageId); ps.setString(3, JdbcMessageStore.this.region); }); + + if (groupNotExist) { + try { + doCreateMessageGroup(groupKey, createdDate); + } + catch (DuplicateKeyException e) { + logger.warn("Lost race to create group; attempting update instead", e); + updateMessageGroup(groupKey); + } + } + else { + updateMessageGroup(groupKey); + } } @Override @@ -406,32 +398,30 @@ 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<>(); + Map groupInfo = getGroupMetadata(key); - this.jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), rs -> { - updateDate.set(rs.getTimestamp("UPDATED_DATE")); - - createDate.set(rs.getTimestamp("CREATED_DATE")); - - completeFlag.set(rs.getInt("COMPLETE") > 0); - - lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); - }, key, this.region); - - if (createDate.get() == null && updateDate.get() == null) { + if (groupInfo.isEmpty()) { return new SimpleMessageGroup(groupId); } MessageGroup messageGroup = getMessageGroupFactory() - .create(this, groupId, createDate.get().getTime(), completeFlag.get()); - messageGroup.setLastModified(updateDate.get().getTime()); - messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceRef.get()); + .create(this, groupId, ((Timestamp) groupInfo.get("CREATED_DATE")).getTime(), + ((Long) groupInfo.get("COMPLETE")) > 0); + messageGroup.setLastModified(((Timestamp) groupInfo.get("UPDATED_DATE")).getTime()); + messageGroup.setLastReleasedMessageSequenceNumber(((Long) groupInfo.get("LAST_RELEASED_SEQUENCE")).intValue()); + messageGroup.setCondition((String) groupInfo.get("CONDITION")); return messageGroup; } + private Map getGroupMetadata(String groupKey) { + try { + return this.jdbcTemplate.queryForMap(getQuery(Query.GET_GROUP_INFO), groupKey, this.region); + } + catch (IncorrectResultSizeDataAccessException ex) { + return Collections.emptyMap(); + } + } + @Override public void removeMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, "'groupId' must not be null"); @@ -450,6 +440,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa ps.setString(2, getKey(messageToRemove.getHeaders().getId())); ps.setString(3, JdbcMessageStore.this.region); }); + this.jdbcTemplate.batchUpdate(getQuery(Query.DELETE_MESSAGE), messages, getRemoveBatchSize(), @@ -457,7 +448,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa ps.setString(1, getKey(messageToRemove.getHeaders().getId())); ps.setString(2, JdbcMessageStore.this.region); }); - this.updateMessageGroup(groupKey); + + updateMessageGroup(groupKey); } @Override @@ -480,7 +472,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), ps -> { if (logger.isDebugEnabled()) { - logger.debug("Marking messages with group key=" + groupKey); + logger.debug("Deleting messages with group key=" + groupKey); } ps.setString(1, groupKey); ps.setString(2, JdbcMessageStore.this.region); @@ -489,43 +481,56 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override public void completeGroup(Object groupId) { - final long updatedDate = System.currentTimeMillis(); final String groupKey = getKey(groupId); this.jdbcTemplate.update(getQuery(Query.COMPLETE_GROUP), ps -> { if (logger.isDebugEnabled()) { logger.debug("Completing MessageGroup: " + groupKey); } - ps.setTimestamp(1, new Timestamp(updatedDate)); + ps.setTimestamp(1, new Timestamp(System.currentTimeMillis())); ps.setString(2, groupKey); ps.setString(3, JdbcMessageStore.this.region); }); } + @Override + public void setGroupCondition(Object groupId, String condition) { + Assert.notNull(groupId, "'groupId' must not be null"); + String groupKey = getKey(groupId); + Timestamp updatedDate = new Timestamp(System.currentTimeMillis()); + this.jdbcTemplate.update(getQuery(Query.UPDATE_MESSAGE_GROUP), ps -> { + if (logger.isDebugEnabled()) { + logger.debug("Updating message group with id key=" + groupKey + " and updated date=" + updatedDate); + } + ps.setTimestamp(1, updatedDate); + ps.setString(2, condition); + ps.setString(3, groupKey); + ps.setString(4, this.region); + }); + } + @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, final int sequenceNumber) { Assert.notNull(groupId, "'groupId' must not be null"); - final long updatedDate = System.currentTimeMillis(); - final String groupKey = getKey(groupId); + String groupKey = getKey(groupId); this.jdbcTemplate.update(getQuery(Query.UPDATE_LAST_RELEASED_SEQUENCE), ps -> { if (logger.isDebugEnabled()) { logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " + groupKey); } - ps.setTimestamp(1, new Timestamp(updatedDate)); + ps.setTimestamp(1, new Timestamp(System.currentTimeMillis())); ps.setInt(2, sequenceNumber); ps.setString(3, groupKey); ps.setString(4, JdbcMessageStore.this.region); }); - this.updateMessageGroup(groupKey); } @Override public Message pollMessageFromGroup(Object groupId) { String key = getKey(groupId); - Message polledMessage = this.doPollForMessage(key); + Message polledMessage = doPollForMessage(key); if (polledMessage != null) { this.removeMessagesFromGroup(groupId, polledMessage); } @@ -597,7 +602,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * To be used to get a reference to JdbcOperations * in case this class is subclassed - * * @return the JdbcOperations implementation */ protected JdbcOperations getJdbcOperations() { @@ -621,30 +625,19 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa return null; } - private void doCreateMessageGroup(final String groupKey, final Timestamp createdDate) { + private void doCreateMessageGroup(String groupKey, Timestamp createdDate) { this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE_GROUP), ps -> { if (logger.isDebugEnabled()) { logger.debug("Creating message group with id key=" + groupKey + " and created date=" + createdDate); } ps.setString(1, groupKey); - ps.setString(2, JdbcMessageStore.this.region); + ps.setString(2, this.region); ps.setTimestamp(3, createdDate); ps.setTimestamp(4, createdDate); }); } - private void doUpdateMessageGroup(final String groupKey, final Timestamp updatedDate) { - this.jdbcTemplate.update(getQuery(Query.UPDATE_MESSAGE_GROUP), ps -> { - if (logger.isDebugEnabled()) { - logger.debug("Updating message group with id key=" + groupKey + " and updated date=" + updatedDate); - } - ps.setTimestamp(1, updatedDate); - ps.setString(2, groupKey); - ps.setString(3, JdbcMessageStore.this.region); - }); - } - - private void updateMessageGroup(final String groupId) { + private void updateMessageGroup(String groupId) { this.jdbcTemplate.update(getQuery(Query.UPDATE_GROUP), ps -> { if (logger.isDebugEnabled()) { logger.debug("Updating MessageGroup: " + groupId); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql index ced3a7373f..c41d4c1b8c 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql index b8e51d5045..eda5a5e374 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql index d0d7c265bb..0bdbd57f39 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql index ce75dea9ef..43460a98e7 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql index a46f6237bd..9b9ca26811 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE DATETIME(6) NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql index 0c6053873e..1cfdd2856d 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY VARCHAR2(36) NOT NULL, REGION VARCHAR2(100) NOT NULL, - MARKED NUMBER(19,0), + CONDITION VARCHAR(255), COMPLETE NUMBER(19,0), LAST_RELEASED_SEQUENCE NUMBER(19,0), CREATED_DATE TIMESTAMP NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql index 5b50bf867e..453d22f6e8 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE TIMESTAMP NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql index a3743df292..8d2b45640e 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE DATETIME NOT NULL, diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql index 4e625f9433..63506a3ede 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql @@ -18,7 +18,7 @@ CREATE TABLE INT_GROUP_TO_MESSAGE ( CREATE TABLE INT_MESSAGE_GROUP ( GROUP_KEY CHAR(36) NOT NULL, REGION VARCHAR(100) NOT NULL, - MARKED BIGINT, + CONDITION VARCHAR(255), COMPLETE BIGINT, LAST_RELEASED_SEQUENCE BIGINT, CREATED_DATE DATETIME NOT NULL, 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 349bdc0182..3916ee831d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-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. @@ -236,6 +236,11 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe throw NOT_IMPLEMENTED; } + @Override + public void setGroupCondition(Object groupId, String condition) { + throw NOT_IMPLEMENTED; + } + @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { throw NOT_IMPLEMENTED; 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 de3bcbc7c2..fd5a94948f 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 @@ -131,8 +131,8 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb .create(this, groupId, createdTime, complete); messageGroup.setLastModified(lastModifiedTime); messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequence); + messageGroup.setCondition(messageDocument.getCondition()); return messageGroup; - } else { return new SimpleMessageGroup(groupId); @@ -157,10 +157,13 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb int lastReleasedSequence = 0; boolean complete = false; + String condition = null; + if (messageDocument != null) { createdTime = messageDocument.getGroupCreatedTime(); lastReleasedSequence = messageDocument.getLastReleasedSequence(); complete = messageDocument.isComplete(); + condition = messageDocument.getCondition(); } for (Message message : messages) { @@ -171,7 +174,9 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb document.setGroupCreatedTime(createdTime); document.setLastModifiedTime(messageDocument == null ? createdTime : System.currentTimeMillis()); document.setSequence(getNextId()); - + if (condition != null) { + document.setCondition(condition); + } addMessageDocument(document); } } @@ -221,6 +226,11 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.LAST_RELEASED_SEQUENCE, sequenceNumber)); } + @Override + public void setGroupCondition(Object groupId, String condition) { + updateGroup(groupId, lastModifiedUpdate().set("condition", condition)); + } + @Override public void completeGroup(Object groupId) { updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.COMPLETE, true)); 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 0507ba4cd5..c98358c7fd 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-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,6 +22,7 @@ import org.springframework.data.annotation.AccessType; import org.springframework.data.annotation.Id; import org.springframework.data.annotation.PersistenceConstructor; import org.springframework.data.mongodb.core.mapping.Document; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -61,6 +62,8 @@ public class MessageDocument { private Integer lastReleasedSequence = 0; + private String condition; + private long sequence; public MessageDocument(Message message) { @@ -151,6 +154,15 @@ public class MessageDocument { return this.groupId; } + @Nullable + public String getCondition() { + return this.condition; + } + + public void setCondition(String condition) { + this.condition = condition; + } + public long getSequence() { return this.sequence; } 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 ec150a4b12..27de90c90e 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 @@ -286,6 +286,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore .create(this, groupId, createdTime, complete); messageGroup.setLastModified(lastModifiedTime); messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequence); + messageGroup.setCondition(messageWrapper.get_Condition()); return messageGroup; } @@ -304,11 +305,12 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore long createdTime = System.currentTimeMillis(); int lastReleasedSequence = 0; boolean complete = false; - + String condition = null; if (messageDocument != null) { createdTime = messageDocument.get_Group_timestamp(); lastReleasedSequence = messageDocument.get_LastReleasedSequenceNumber(); complete = messageDocument.get_Group_complete(); + condition = messageDocument.get_Condition(); } for (Message message : messages) { @@ -318,7 +320,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore wrapper.set_Group_update_timestamp(messageDocument == null ? createdTime : System.currentTimeMillis()); wrapper.set_Group_complete(complete); wrapper.set_LastReleasedSequenceNumber(lastReleasedSequence); - wrapper.set_Sequence(getNextId()); + wrapper.setSequence(getNextId()); + if (condition != null) { + wrapper.set_Condition(condition); + } addMessageDocument(wrapper); } @@ -393,9 +398,14 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore return (int) lCount; } + @Override + public void setGroupCondition(Object groupId, String condition) { + updateGroup(groupId, lastModifiedUpdate().set("_condition", condition)); + } + @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { - this.updateGroup(groupId, lastModifiedUpdate().set(LAST_RELEASED_SEQUENCE_NUMBER, sequenceNumber)); + updateGroup(groupId, lastModifiedUpdate().set(LAST_RELEASED_SEQUENCE_NUMBER, sequenceNumber)); } @Override @@ -618,6 +628,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore if (completeGroup != null) { wrapper.set_Group_complete(completeGroup); } + wrapper.set_Condition((String) sourceMap.get("_condition")); return (S) wrapper; } @@ -860,6 +871,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore private volatile boolean _group_complete; // NOSONAR name + private volatile String _condition; // NOSONAR name + @SuppressWarnings(UNUSED) private long sequence; @@ -930,7 +943,15 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore this._group_complete = completedGroup; } - public void set_Sequence(long sequence) { // NOSONAR name + public String get_Condition() { + return this._condition; + } + + public void set_Condition(String condition) { + this._condition = condition; + } + + public void setSequence(long sequence) { this.sequence = sequence; } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java index 940e578cff..c2fa22792a 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.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.Iterator; import java.util.List; import java.util.Properties; import java.util.UUID; +import java.util.function.BiFunction; import org.junit.After; import org.junit.Before; @@ -31,6 +32,7 @@ import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.integration.aggregator.ReleaseStrategy; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.history.MessageHistory; @@ -54,6 +56,11 @@ import org.springframework.messaging.support.GenericMessage; */ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvailableTests { + public static final BiFunction, String, String> CONDITION_SUPPLIER = (m, c) -> "10"; + + public static final ReleaseStrategy RELEASE_STRATEGY = + group -> group.size() == Integer.parseInt(group.getCondition()); + protected final GenericApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); @Before diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml index 8734a4a2dc..29278e4625 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml @@ -3,12 +3,21 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration" xmlns:mongo="http://www.springframework.org/schema/data/mongo" + xmlns:util="http://www.springframework.org/schema/util" xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd - http://www.springframework.org/schema/data/mongo https://www.springframework.org/schema/data/mongo/spring-mongo.xsd"> + http://www.springframework.org/schema/data/mongo https://www.springframework.org/schema/data/mongo/spring-mongo.xsd + http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd"> + release-strategy="releaseStrategy" + group-condition-supplier="conditionSupplier"/> + + + + @@ -20,4 +29,4 @@ - + \ No newline at end of file diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml index c4592c541e..552b5ec778 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml @@ -3,12 +3,21 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration" xmlns:mongo="http://www.springframework.org/schema/data/mongo" + xmlns:util="http://www.springframework.org/schema/util" xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd - http://www.springframework.org/schema/data/mongo https://www.springframework.org/schema/data/mongo/spring-mongo.xsd"> + http://www.springframework.org/schema/data/mongo https://www.springframework.org/schema/data/mongo/spring-mongo.xsd + http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd"> + release-strategy="releaseStrategy" + group-condition-supplier="conditionSupplier"/> + + + + diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 8a64a236f3..0b94b694eb 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -970,4 +970,12 @@ Flux> window = ---- ==== +[[agg-message-group-condition]] +==== Condition on the Message Group + +Starting with version 5.5, an `AbstractCorrelatingMessageHandler` (including its Java & XML DSLs) exposes a `groupConditionSupplier` option of the `BiFunction, String, String>` implementation. +This function is used on each message added to the group and a result condition sentence is stored into the group for future consideration. +The `ReleaseStrategy` may consult this condition instead of iterating over all the messages in the group. +See `GroupConditionProvider` JavaDocs and <<./message-store.adoc#message-group-condition, Message Group Condition>> for more information. + See also <<./file.adoc#file-aggregator, File Aggregator>>. \ No newline at end of file diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index e3ec04e32f..2b8389bcf6 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -1192,6 +1192,8 @@ When markers are enabled on the `FileSplitter`, it does not populate sequence de The `FileHeaders.FILENAME` is still populated for each line emitted, including START/END marker messages. - The `FileMarkerReleaseStrategy` - checks for `FileSplitter.FileMarker.Mark.END` message in the group and then compare a `FileHeaders.LINE_COUNT` header value with the group size minus `2` - `FileSplitter.FileMarker` instances. +It also implements a convenient `GroupConditionProvider` contact for `conditionSupplier` function to be used in the `AbstractCorrelatingMessageHandler`. +See <<./message-store.adoc#message-group-condition, Message Group Condition>> for more information. - The `FileAggregatingMessageGroupProcessor` just removes `FileSplitter.FileMarker` messages from the group and collect the rest of messages into a list payload to produce. diff --git a/src/reference/asciidoc/message-store.adoc b/src/reference/asciidoc/message-store.adoc index 1081de2f43..d4bcace9ef 100644 --- a/src/reference/asciidoc/message-store.adoc +++ b/src/reference/asciidoc/message-store.adoc @@ -147,4 +147,22 @@ However starting with version 5.5, all the persistent `MessageGroupStore` implem 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 +Whenever a `PersistentMessageGroup` is used, its `streamMessages()` delegates to the `MessageGroupStore.streamMessagesForGroup()`. + +[[message-group-condition]] +==== Message Group Condition + +Starting with version 5.5, the `MessageGroup` abstraction provides a `condition` string option. +The value of this option can be anything that could be parsed later on for any reason to make a decision for the group. +For example a `ReleaseStrategy` from a <<./aggregator.adoc#aggregator-api, correlation message handler>> may consult this property from the group instead of iterating all the messages in the group. +The `MessageGroupStore` exposes a `setGroupCondition(Object groupId, String condition)` API. +For this purpose a `setGroupConditionSupplier(BiFunction, String, String>)` option has been added to the `AbstractCorrelatingMessageHandler`. +This function is evaluated against each message after it has been added to the group as well as the existing condition of the group. +The implementation may decide to return a new value, the existing value, or reset the target condition to `null`. +The value for a `condition` can be a JSON, SpEL expression, number or anything what can be serialized as a string and parsed afterwards. +For example, the `FileMarkerReleaseStrategy` from the <<./file.adoc#file-aggregator, File Aggregator>> component, populates a condition into a group from the `FileHeaders.LINE_COUNT` header of the `FileSplitter.FileMarker.Mark.END` message and consults with it from its `canRelease()` comparing a group size with the value in this condition. +This way it doesn't iterate all the messages in group to find a `FileSplitter.FileMarker.Mark.END` message with the `FileHeaders.LINE_COUNT` header. +It also allows the end marker to arrive at the aggregator before all the other records; for example when processing a file in a multi-threaded environment. + +In addition, for configuration convenience, a `GroupConditionProvider` contract has been introduced. +The `AbstractCorrelatingMessageHandler` checks if the provided `ReleaseStrategy` implements this interface and extracts a `conditionSupplier` for group condition evaluation logic. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 9870817db4..bebcb8061f 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -41,8 +41,12 @@ This is covered as a `ConsumerEndpointSpec.reactive()` option in Java DSL and as See <<./reactive-streams.adoc#reactive-streams,Reactive Streams Support>> for more information. The `groupTimeoutExpression` for a correlation message handler (an `Aggregator` and `Resequencer`) can now be evaluated to a `java.util.Date` for some fine-grained scheduling use-cases. +Also the `BiFunction groupConditionSupplier` option is added to the `AbstractCorrelatingMessageHandler` to supply a `MessageGroup` condition against a message to be added to the group. See <<./aggregator.adoc#aggregator,Aggregator>> for more information. +The `MessageGroup` abstraction can be supplied with a `condition` to evaluate later on to make a decision for the group. +See <<./message-store.adoc#message-group-condition,Message Group Condition>> for more information. + [[x5.5-amqp]] ==== AMQP Changes