From 3ce1fcb110065d33bf495d2ac72a38c8c071281e Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Tue, 3 Jan 2017 12:28:17 -0500
Subject: [PATCH] INT-4199: Fix Asserts with no Message

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

MQTT and JDBC.

Conflicts:
	spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
Resolved.
---
 .../integration/jdbc/JdbcMessageStore.java         |  6 ++++--
 .../jdbc/store/JdbcChannelMessageStore.java        |  6 ++++--
 .../mqtt/support/DefaultPahoMessageConverter.java  | 14 +++++++++++---
 3 files changed, 19 insertions(+), 7 deletions(-)

diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
index fe079d3e6c..25acc8cc93 100644
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
@@ -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.
@@ -691,7 +691,9 @@ 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);
+		if (messages.size() > 1) {
+			throw new IllegalStateException("The query must return zero or 1 row; got " + messages.size() + " rows");
+		}
 		if (messages.size() > 0) {
 			return messages.get(0);
 		}
diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
index 41d2c2f9cb..205801db35 100644
--- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
+++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
@@ -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.
@@ -600,7 +600,9 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 		}
 
 
-		Assert.isTrue(messages.size() == 0 || messages.size() == 1);
+		if (messages.size() > 1) {
+			throw new IllegalStateException("The query must return zero or 1 row; got " + messages.size() + " rows");
+		}
 		if (messages.size() > 0) {
 
 			final Message<?> message = messages.get(0);
diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java
index 7dbf05f7a3..5dd2f857ea 100644
--- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java
+++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java
@@ -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.
@@ -132,7 +132,10 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
 
 	@Override
 	public Message<?> toMessage(Object mqttMessage, MessageHeaders headers) {
-		Assert.isInstanceOf(MqttMessage.class, mqttMessage);
+		if (!(mqttMessage instanceof MqttMessage)) {
+			throw new IllegalArgumentException("This converter can only convert an 'MqttMessage'; received: "
+						+ mqttMessage.getClass().getName());
+		}
 		return toMessage(null, (MqttMessage) mqttMessage);
 	}
 
@@ -193,7 +196,12 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
 	 */
 	protected byte[] messageToMqttBytes(Message<?> message) {
 		Object payload = message.getPayload();
-		Assert.isTrue(payload instanceof byte[] || payload instanceof String);
+		if (!(payload instanceof byte[] || payload instanceof String)) {
+			throw new IllegalArgumentException(
+					"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 {