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

INT-4199: Fix Asserts with no Message

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4199

MQTT and JDBC.

Also remove unused message builder from the `JdbcMessageStore`.
This commit is contained in:
Gary Russell
2017-01-03 12:28:17 -05:00
committed by Artem Bilan
parent 0cbfd6e3e0
commit 395f08b50d
3 changed files with 14 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -639,7 +639,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
protected Message<?> doPollForMessage(String groupIdKey) {
List<Message<?>> messages = this.jdbcTemplate.query(getQuery(Query.POLL_FROM_GROUP), this.mapper,
groupIdKey, this.region, groupIdKey, this.region);
Assert.isTrue(messages.size() == 0 || messages.size() == 1);
Assert.state(messages.size() < 2,
() -> "The query must return zero or 1 row; got " + messages.size() + " rows");
if (messages.size() > 0) {
return messages.get(0);
}

14
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -52,9 +52,6 @@ import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.PriorityCapableChannelMessageStore;
import org.springframework.integration.store.SimpleMessageGroupFactory;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
@@ -135,8 +132,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
@Deprecated
public static final String CREATED_DATE_KEY = JdbcChannelMessageStore.class.getSimpleName() + ".CREATED_DATE";
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile String region = DEFAULT_REGION;
private volatile String tablePrefix = DEFAULT_TABLE_PREFIX;
@@ -402,9 +397,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
logger.warn("The jdbcTemplate's fetch size is not 1. This may cause FIFO issues with Oracle databases.");
}
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.jdbcTemplate.afterPropertiesSet();
}
@@ -419,7 +411,6 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
* @param message a message
*/
@Override
@SuppressWarnings("unchecked")
public MessageGroup addMessageToGroup(Object groupId, final Message<?> message) {
String groupKey = getKey(groupId);
@@ -600,7 +591,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
}
Assert.isTrue(messages.size() == 0 || messages.size() == 1);
Assert.state(messages.size() < 2,
() -> "The query must return zero or 1 row; got " + messages.size() + " rows");
if (messages.size() > 0) {
final Message<?> message = messages.get(0);

11
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -177,7 +177,9 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
@Override
public Message<?> toMessage(Object mqttMessage, MessageHeaders headers) {
Assert.isInstanceOf(MqttMessage.class, mqttMessage);
Assert.isInstanceOf(MqttMessage.class, mqttMessage,
() -> "This converter can only convert an 'MqttMessage'; received: "
+ mqttMessage.getClass().getName());
return toMessage(null, (MqttMessage) mqttMessage);
}
@@ -236,7 +238,10 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
*/
protected byte[] messageToMqttBytes(Message<?> message) {
Object payload = message.getPayload();
Assert.isTrue(payload instanceof byte[] || payload instanceof String);
Assert.isTrue(payload instanceof byte[] || payload instanceof String,
() -> "This default converter can only handle 'byte[]' or 'String' payloads; consider adding a "
+ "transformer to your flow definition, or subclass this converter for "
+ payload.getClass().getName() + " payloads");
byte[] payloadBytes;
if (payload instanceof String) {
try {