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

Miscellaneous fixes

Browse Source 
* Fix `MiscellaneousTests` to extend `ActiveMQMultiContextTests`
to let it to close ActiveMQ connection factory in the end of test
* Make `RequestReplyScenariosWithTempReplyQueuesTests` more robust
closing `DefaultMessageListenerContainer` and `ExecutorService`
in the end of tests
* Fix new Sonar smells
* Rework some SQL calls in the `JdbcMessageStore` to note expose a
`PreparedStatement` API
* Fix JavaDoc in the `CorrelationHandlerSpec`
This commit is contained in:
Artem Bilan
2021-03-26 12:39:00 -04:00
parent 7abbe30c81
commit f2009271dc
8 changed files with 90 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java
View File

@@ -403,6 +403,7 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
* 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.
* @return the endpoint spec.
* @since 5.5
*/
public S groupConditionSupplier(BiFunction<Message<?>, String, String> conditionSupplier) {

11
spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileAggregator.java
View File

@@ -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.
* <p>
* 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.
* <p>
* If default behavior of this component does not satisfy the target logic, it is recommended to
* configure an aggregator with individual strategies.

3
spring-integration-file/src/main/java/org/springframework/integration/file/aggregator/FileMarkerReleaseStrategy.java
View File

@@ -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<Message<?>, String, String> GROUP_CONDITION =

124
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java
View File

@@ -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) {

32
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java
View File

@@ -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();
}

15
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java
View File

@@ -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>) (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();
}
}

0
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml → spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/multi-producer-and-consumers-temp-reply.xml
View File

12
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java
View File

@@ -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;
}