From 89cf737ad45457975c68dc1e4ca5bcfe31a90ee9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 8 Nov 2013 10:28:28 -0500 Subject: [PATCH] MQTT - Update to Spring Integration 4.0.0.M1 --- .../mqtt/outbound/AbstractMqttMessageHandler.java | 2 +- .../mqtt/support/DefaultPahoMessageConverter.java | 12 ++++++------ .../mqtt/support/MqttMessageConverter.java | 2 +- .../integration/mqtt/BackTobackAdapterTests.java | 2 +- .../integration/mqtt/MqttAdapterTests.java | 2 +- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java index 177a0ac8bf..3dc72403ef 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java @@ -140,7 +140,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler protected void handleMessageInternal(Message message) throws Exception { this.connectIfNeeded(); String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC); - MqttMessage mqttMessage = this.converter.fromMessage(message, MqttMessage.class); + MqttMessage mqttMessage = (MqttMessage) this.converter.fromMessage(message, MqttMessage.class); if (topic == null && this.defaultTopic == null) { throw new MessageHandlingException(message, "No '" + MqttHeaders.TOPIC + "' header and no default topic defined"); 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 1d8753491b..63f61c61b7 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 @@ -15,12 +15,11 @@ */ package org.springframework.integration.mqtt.support; -import java.lang.reflect.Type; - import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.converter.MessageConversionException; import org.springframework.util.Assert; @@ -54,11 +53,12 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter { } @Override - @SuppressWarnings("unchecked") - public Message toMessage(MqttMessage mqttMessage) { - return toMessage(null, mqttMessage); + public Message toMessage(Object mqttMessage, MessageHeaders headers) { + Assert.isInstanceOf(MqttMessage.class, mqttMessage); + return toMessage(null, (MqttMessage) mqttMessage); } + @Override public Message toMessage(String topic, MqttMessage mqttMessage) { try { MessageBuilder messageBuilder = MessageBuilder.withPayload(new String(mqttMessage.getPayload(), this.charset)) @@ -76,7 +76,7 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter { } @Override - public MqttMessage fromMessage(Message message, Type targetClass) { + public MqttMessage fromMessage(Message message, Class targetClass) { Object payload = message.getPayload(); Assert.isTrue(payload instanceof byte[] || payload instanceof String); byte[] payloadBytes; diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java index 61c586a98e..146a7a6379 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java @@ -27,7 +27,7 @@ import org.springframework.messaging.support.converter.MessageConverter; * @since 1.0 * */ -public interface MqttMessageConverter extends MessageConverter { +public interface MqttMessageConverter extends MessageConverter { Message toMessage(String topic, MqttMessage mqttMessage); } diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java index 0c4f31e830..45a17f8793 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java @@ -22,12 +22,12 @@ import org.junit.Rule; import org.junit.Test; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.message.GenericMessage; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.integration.mqtt.support.MqttHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index c3cd9125ce..e55800eda2 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -44,12 +44,12 @@ import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.message.GenericMessage; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory.Will; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /**