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 45c3b73760..ab17df8ee1 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 @@ -403,6 +403,7 @@ public abstract class CorrelationHandlerSpec, String, String> conditionSupplier) { 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 db8262c5d0..01bc6d660b 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 @@ -27,21 +27,22 @@ import org.springframework.integration.aggregator.HeaderAttributeCorrelationStra import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.ReleaseStrategy; import org.springframework.integration.file.FileHeaders; -import org.springframework.integration.file.splitter.FileSplitter; import org.springframework.integration.store.MessageGroup; import org.springframework.messaging.Message; /** - * A convenient component to utilize a {@link FileSplitter.FileMarker}-based aggregation logic. + * A convenient component to utilize + * a {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker}-based aggregation logic. * Implements all three {@link CorrelationStrategy}, {@link ReleaseStrategy} and {@link MessageGroupProcessor} * for runtime optimization. * Delegates to {@link HeaderAttributeCorrelationStrategy} with {@link FileHeaders#FILENAME} attribute, * {@link FileMarkerReleaseStrategy} and {@link FileAggregatingMessageGroupProcessor}, respectively. *

- * The default {@link FileSplitter} behavior with markers enabled is about do not provide a sequence details + * The default {@link org.springframework.integration.file.splitter.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. + * header which is still populated by the {@link org.springframework.integration.file.splitter.FileSplitter} + * for each line emitted, including {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker} messages. *

* If default behavior of this component does not satisfy the target logic, it is recommended to * configure an aggregator with individual strategies. 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 559c2b4b1c..b9b22c1600 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 @@ -17,7 +17,6 @@ package org.springframework.integration.file.aggregator; import java.util.function.BiFunction; -import java.util.function.Function; import org.springframework.integration.aggregator.GroupConditionProvider; import org.springframework.integration.aggregator.ReleaseStrategy; @@ -42,7 +41,7 @@ import org.springframework.messaging.MessageHeaders; public class FileMarkerReleaseStrategy implements ReleaseStrategy, GroupConditionProvider { /** - * The {@link Function} for + * The {@link BiFunction} for * {@link org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler#setGroupConditionSupplier(BiFunction)}. */ public static final BiFunction, String, String> GROUP_CONDITION = 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 dd9e741d70..f383e53ddc 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 @@ -314,11 +314,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa try { this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE), ps -> { - ps.setString(1, messageId); - ps.setString(2, this.region); - ps.setTimestamp(3, new Timestamp(System.currentTimeMillis())); + ps.setString(1, messageId); // NOSONAR - magic number + ps.setString(2, this.region); // NOSONAR - magic number + ps.setTimestamp(3, new Timestamp(System.currentTimeMillis())); // NOSONAR - magic number - this.lobHandler.getLobCreator().setBlobAsBytes(ps, 4, messageBytes); + this.lobHandler.getLobCreator().setBlobAsBytes(ps, 4, messageBytes); // NOSONAR - magic number }); } catch (DuplicateKeyException e) { @@ -352,9 +352,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa logger.debug("Inserting message with id key=" + messageId + " and created date=" + createdDate); } - ps.setString(1, groupKey); - ps.setString(2, messageId); - ps.setString(3, JdbcMessageStore.this.region); + ps.setString(1, groupKey); // NOSONAR - magic number + ps.setString(2, messageId); // NOSONAR - magic number + ps.setString(3, JdbcMessageStore.this.region); // NOSONAR - magic number }); if (groupNotExist) { @@ -436,17 +436,17 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa messages, getRemoveBatchSize(), (ps, messageToRemove) -> { - ps.setString(1, groupKey); - ps.setString(2, getKey(messageToRemove.getHeaders().getId())); - ps.setString(3, JdbcMessageStore.this.region); + ps.setString(1, groupKey); // NOSONAR - magic number + ps.setString(2, getKey(messageToRemove.getHeaders().getId())); // NOSONAR - magic number + ps.setString(3, JdbcMessageStore.this.region); // NOSONAR - magic number }); this.jdbcTemplate.batchUpdate(getQuery(Query.DELETE_MESSAGE), messages, getRemoveBatchSize(), (ps, messageToRemove) -> { - ps.setString(1, getKey(messageToRemove.getHeaders().getId())); - ps.setString(2, JdbcMessageStore.this.region); + ps.setString(1, getKey(messageToRemove.getHeaders().getId())); // NOSONAR - magic number + ps.setString(2, JdbcMessageStore.this.region); // NOSONAR - magic number }); updateMessageGroup(groupKey); @@ -456,41 +456,29 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public void removeMessageGroup(Object groupId) { String groupKey = getKey(groupId); - this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGES_FROM_GROUP), ps -> { - ps.setString(1, groupKey); - ps.setString(2, JdbcMessageStore.this.region); - ps.setString(3, JdbcMessageStore.this.region); - }); + this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGES_FROM_GROUP), groupKey, this.region, this.region); - this.jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), ps -> { - if (logger.isDebugEnabled()) { - logger.debug("Removing relationships for the group with group key=" + groupKey); - } - ps.setString(1, groupKey); - ps.setString(2, JdbcMessageStore.this.region); - }); + if (logger.isDebugEnabled()) { + logger.debug("Removing relationships for the group with group key=" + groupKey); + } + this.jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), groupKey, this.region); - this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), ps -> { - if (logger.isDebugEnabled()) { - logger.debug("Deleting messages with group key=" + groupKey); - } - ps.setString(1, groupKey); - ps.setString(2, JdbcMessageStore.this.region); - }); + if (logger.isDebugEnabled()) { + logger.debug("Deleting messages with group key=" + groupKey); + } + + this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), groupKey, this.region); } @Override public void completeGroup(Object groupId) { 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(System.currentTimeMillis())); - ps.setString(2, groupKey); - ps.setString(3, JdbcMessageStore.this.region); - }); + if (logger.isDebugEnabled()) { + logger.debug("Completing MessageGroup: " + groupKey); + } + this.jdbcTemplate.update(getQuery(Query.COMPLETE_GROUP), + new Timestamp(System.currentTimeMillis()), groupKey, this.region); } @Override @@ -498,32 +486,22 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa 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); - }); + if (logger.isDebugEnabled()) { + logger.debug("Updating message group with id key=" + groupKey + " and updated date=" + updatedDate); + } + this.jdbcTemplate.update(getQuery(Query.UPDATE_MESSAGE_GROUP), updatedDate, condition, groupKey, this.region); } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, final int sequenceNumber) { + public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, "'groupId' must not be null"); 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(System.currentTimeMillis())); - ps.setInt(2, sequenceNumber); - ps.setString(3, groupKey); - ps.setString(4, JdbcMessageStore.this.region); - }); + if (logger.isDebugEnabled()) { + logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " + groupKey); + } + this.jdbcTemplate.update(getQuery(Query.UPDATE_LAST_RELEASED_SEQUENCE), + new Timestamp(System.currentTimeMillis()), sequenceNumber, groupKey, this.region); } @Override @@ -532,7 +510,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa Message polledMessage = doPollForMessage(key); if (polledMessage != null) { - this.removeMessagesFromGroup(groupId, polledMessage); + removeMessagesFromGroup(groupId, polledMessage); } return polledMessage; } @@ -626,26 +604,18 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } 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, this.region); - ps.setTimestamp(3, createdDate); - ps.setTimestamp(4, createdDate); - }); + if (logger.isDebugEnabled()) { + logger.debug("Creating message group with id key=" + groupKey + " and created date=" + createdDate); + } + this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE_GROUP), groupKey, this.region, createdDate, createdDate); } private void updateMessageGroup(String groupId) { - this.jdbcTemplate.update(getQuery(Query.UPDATE_GROUP), ps -> { - if (logger.isDebugEnabled()) { - logger.debug("Updating MessageGroup: " + groupId); - } - ps.setTimestamp(1, new Timestamp(System.currentTimeMillis())); - ps.setString(2, groupId); - ps.setString(3, JdbcMessageStore.this.region); - }); + if (logger.isDebugEnabled()) { + logger.debug("Updating MessageGroup: " + groupId); + } + this.jdbcTemplate.update(getQuery(Query.UPDATE_GROUP), + new Timestamp(System.currentTimeMillis()), groupId, this.region); } private String getKey(Object input) { diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java index 2ec8bb7b48..eb3db2ed5e 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java @@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.ActiveMQMultiContextTests; import org.springframework.integration.test.condition.LongRunningTest; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.StopWatch; @@ -32,9 +33,10 @@ import org.springframework.util.StopWatch; /** * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ @LongRunningTest -public class MiscellaneousTests { +public class MiscellaneousTests extends ActiveMQMultiContextTests { /** * Asserts that receive-timeout is honored even if @@ -43,20 +45,22 @@ public class MiscellaneousTests { */ @Test public void testTimeoutHonoringWhenRequestsQueuedUp() throws Exception { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("honor-timeout.xml", this.getClass()); - final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); - final CountDownLatch latch = new CountDownLatch(3); - final AtomicInteger replies = new AtomicInteger(); - StopWatch stopWatch = new StopWatch(); - stopWatch.start(); - for (int i = 0; i < 3; i++) { - this.exchange(latch, gateway, replies); + try (ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("honor-timeout.xml", getClass())) { + + final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); + final CountDownLatch latch = new CountDownLatch(3); + final AtomicInteger replies = new AtomicInteger(); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + for (int i = 0; i < 3; i++) { + this.exchange(latch, gateway, replies); + } + latch.await(); + stopWatch.stop(); + assertThat(stopWatch.getTotalTimeMillis()).isLessThanOrEqualTo(18000); + assertThat(replies.get()).isEqualTo(1); } - latch.await(); - stopWatch.stop(); - assertThat(stopWatch.getTotalTimeMillis() <= 18000).isTrue(); - assertThat(replies.get()).isEqualTo(1); - context.close(); } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java index a13182836a..bf05cef9db 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java @@ -104,15 +104,16 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti @Test public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReply() throws Exception { + DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer(); try (ClassPathXmlApplicationContext context = - new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass())) { + new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", getClass())) { RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); final Destination requestDestination = context.getBean("siOutQueue", Destination.class); - DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer(); + ; dmlc.setConnectionFactory(connectionFactory); dmlc.setDestination(requestDestination); dmlc.setMessageListener((SessionAwareMessageListener) (message, session) -> { @@ -155,6 +156,10 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti Thread.sleep(1000); assertThat(gateway.exchange(new GenericMessage<>("bar")).getPayload()).isEqualTo("bar"); } + finally { + dmlc.stop(); + dmlc.destroy(); + } } /** @@ -202,11 +207,11 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti @Test public void testConcurrently() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(10); try (ClassPathXmlApplicationContext context = - new ClassPathXmlApplicationContext("mult-producer-and-consumers-temp-reply.xml", this.getClass())) { + new ClassPathXmlApplicationContext("multi-producer-and-consumers-temp-reply.xml", this.getClass())) { final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); - ExecutorService executor = Executors.newFixedThreadPool(10); final int testNumbers = 30; final CountDownLatch latch = new CountDownLatch(testNumbers); final AtomicInteger failures = new AtomicInteger(); @@ -239,6 +244,8 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti assertThat(mismatches.get()).isEqualTo(0); assertThat(failures.get()).isEqualTo(0); assertThat(timeouts.get()).isEqualTo(0); + } + finally { executor.shutdownNow(); } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/multi-producer-and-consumers-temp-reply.xml similarity index 100% rename from spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml rename to spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/multi-producer-and-consumers-temp-reply.xml 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 27de90c90e..502024a262 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,7 +286,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore .create(this, groupId, createdTime, complete); messageGroup.setLastModified(lastModifiedTime); messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequence); - messageGroup.setCondition(messageWrapper.get_Condition()); + messageGroup.setCondition(messageWrapper.getCondition()); return messageGroup; } @@ -310,7 +310,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore createdTime = messageDocument.get_Group_timestamp(); lastReleasedSequence = messageDocument.get_LastReleasedSequenceNumber(); complete = messageDocument.get_Group_complete(); - condition = messageDocument.get_Condition(); + condition = messageDocument.getCondition(); } for (Message message : messages) { @@ -322,7 +322,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore wrapper.set_LastReleasedSequenceNumber(lastReleasedSequence); wrapper.setSequence(getNextId()); if (condition != null) { - wrapper.set_Condition(condition); + wrapper.setCondition(condition); } addMessageDocument(wrapper); @@ -628,7 +628,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore if (completeGroup != null) { wrapper.set_Group_complete(completeGroup); } - wrapper.set_Condition((String) sourceMap.get("_condition")); + wrapper.setCondition((String) sourceMap.get("_condition")); return (S) wrapper; } @@ -943,11 +943,11 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore this._group_complete = completedGroup; } - public String get_Condition() { + public String getCondition() { return this._condition; } - public void set_Condition(String condition) { + public void setCondition(String condition) { this._condition = condition; }