diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index 8b54218b5f..5fec854ef0 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -42,6 +42,8 @@ import org.springframework.integration.mqtt.support.MqttUtils; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.messaging.support.GenericMessage; import org.springframework.util.Assert; /** @@ -205,7 +207,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } catch (Exception ex) { logger.error(ex, "Exception while connecting and subscribing, retrying"); - this.scheduleReconnect(); + scheduleReconnect(); } } @@ -272,7 +274,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv super.removeTopic(topic); } catch (MqttException e) { - throw new MessagingException("Failed to unsubscribe from topic " + Arrays.asList(topic), e); + throw new MessagingException("Failed to unsubscribe from topic(s) " + Arrays.toString(topic), e); } finally { this.topicLock.unlock(); @@ -399,19 +401,52 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override public void messageArrived(String topic, MqttMessage mqttMessage) { - AbstractIntegrationMessageBuilder builder = getConverter().toMessageBuilder(topic, mqttMessage); - if (this.manualAcks) { - builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, - new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client)); + AbstractIntegrationMessageBuilder builder = toMessageBuilder(topic, mqttMessage); + if (builder != null) { + if (this.manualAcks) { + builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, + new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client)); + } + Message message = builder.build(); + try { + sendMessage(message); + } + catch (RuntimeException ex) { + logger.error(ex, () -> "Unhandled exception for " + message.toString()); + throw ex; + } } - Message message = builder.build(); + } + + private AbstractIntegrationMessageBuilder toMessageBuilder(String topic, MqttMessage mqttMessage) { + AbstractIntegrationMessageBuilder builder = null; + Exception conversionError = null; try { - sendMessage(message); + builder = getConverter().toMessageBuilder(topic, mqttMessage); } - catch (RuntimeException ex) { - logger.error(ex, () -> "Unhandled exception for " + message.toString()); - throw ex; + catch (Exception ex) { + conversionError = ex; } + + if (builder == null && conversionError == null) { + conversionError = new IllegalStateException("'MqttMessageConverter' returned 'null'"); + } + + if (conversionError != null) { + GenericMessage message = new GenericMessage<>(mqttMessage); + if (!sendErrorMessageIfNecessary(message, conversionError)) { + MessageConversionException conversionException; + if (conversionError instanceof MessageConversionException) { + conversionException = (MessageConversionException) conversionError; + } + else { + conversionException = new MessageConversionException(message, "Failed to convert from MQTT Message", + conversionError); + } + throw conversionException; + } + } + return builder; } @Override 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 cc8ff67186..266a28155d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -89,9 +89,13 @@ import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannel import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttHeaderAccessor; +import org.springframework.integration.mqtt.support.MqttMessageConverter; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -106,7 +110,7 @@ import org.springframework.util.ReflectionUtils; */ public class MqttAdapterTests { - private IMqttToken alwaysComplete; + private final IMqttToken alwaysComplete; { ProxyFactoryBean pfb = new ProxyFactoryBean(); @@ -198,12 +202,12 @@ public class MqttAdapterTests { return deliveryToken; }).given(client).publish(anyString(), any(MqttMessage.class)); - handler.handleMessage(new GenericMessage("Hello, world!")); + handler.handleMessage(new GenericMessage<>("Hello, world!")); verify(client, times(1)).connect(any(MqttConnectOptions.class)); assertThat(connectCalled.get()).isTrue(); AtomicReference failed = new AtomicReference<>(); - handler.setApplicationEventPublisher(event -> failed.set(event)); + handler.setApplicationEventPublisher(failed::set); handler.connectionLost(new IllegalStateException()); assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class); handler.stop(); @@ -258,7 +262,7 @@ public class MqttAdapterTests { return null; }).given(client).connect(any(MqttConnectOptions.class)); - final AtomicReference callback = new AtomicReference(); + final AtomicReference callback = new AtomicReference<>(); willAnswer(invocation -> { callback.set(invocation.getArgument(0)); return null; @@ -271,12 +275,14 @@ public class MqttAdapterTests { adapter.setManualAcks(true); QueueChannel outputChannel = new QueueChannel(); adapter.setOutputChannel(outputChannel); + QueueChannel errorChannel = new QueueChannel(); + adapter.setErrorChannel(errorChannel); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.setBeanFactory(mock(BeanFactory.class)); ApplicationEventPublisher applicationEventPublisher = mock(ApplicationEventPublisher.class); - final BlockingQueue events = new LinkedBlockingQueue(); + final BlockingQueue events = new LinkedBlockingQueue<>(); willAnswer(invocation -> { events.add(invocation.getArgument(0)); return null; @@ -304,6 +310,39 @@ public class MqttAdapterTests { assertThat(event).isInstanceOf(MqttSubscribedEvent.class); assertThat(((MqttSubscribedEvent) event).getMessage()).isEqualTo("Connected and subscribed to [baz, fix]"); + adapter.setConverter(new MqttMessageConverter() { + + @Override public Message toMessage(String topic, MqttMessage mqttMessage) { + return null; + } + + @Override public AbstractIntegrationMessageBuilder toMessageBuilder(String topic, + MqttMessage mqttMessage) { + + return null; + } + + @Override public Object fromMessage(Message message, Class targetClass) { + return null; + } + + @Override public Message toMessage(Object payload, MessageHeaders headers) { + return null; + } + + + }); + + callback.get().messageArrived("baz", message); + + ErrorMessage errorMessage = (ErrorMessage) errorChannel.receive(0); + assertThat(errorMessage).isNotNull() + .extracting(Message::getPayload) + .isInstanceOf(IllegalStateException.class); + IllegalStateException exception = (IllegalStateException) errorMessage.getPayload(); + assertThat(exception).hasMessage("'MqttMessageConverter' returned 'null'"); + assertThat(errorMessage.getOriginalMessage().getPayload()).isSameAs(message); + // lose connection and make first reconnect fail failConnection.set(true); RuntimeException e = new RuntimeException("foo"); @@ -426,7 +465,7 @@ public class MqttAdapterTests { // the following assertion should be equalTo, but leq to protect against a slow CI server assertThat(attemptingReconnectCount.get()).isLessThanOrEqualTo(2); AtomicReference failed = new AtomicReference<>(); - adapter.setApplicationEventPublisher(event -> failed.set(event)); + adapter.setApplicationEventPublisher(failed::set); adapter.connectionLost(new IllegalStateException()); assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class); adapter.stop(); @@ -458,8 +497,7 @@ public class MqttAdapterTests { new DirectFieldAccessor(client).setPropertyValue("aClient", aClient); willAnswer(new CallsRealMethods()).given(client).connect(any(MqttConnectOptions.class)); willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class)); - willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class), - (IMqttMessageListener[]) isNull()); + willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class), isNull()); willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any()); IMqttToken token = mock(IMqttToken.class); @@ -567,7 +605,7 @@ public class MqttAdapterTests { return adapter; } - private MqttPahoMessageHandler buildAdapterOut(final IMqttAsyncClient client) throws MqttException { + private MqttPahoMessageHandler buildAdapterOut(final IMqttAsyncClient client) { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() { @Override diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index e52042bc32..0ccde980cc 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -154,7 +154,7 @@ Starting with version 5.3, you can set the `manualAcks` property to true. Often used to asynchronously acknowledge delivery. When set to `true`, header (`IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK`) is added to the message with the value being a `SimpleAcknowledgment`. You must invoke the `acknowledge()` method to complete the delivery. -See the Javadocs for `IMqppClient` `setManualAcks()` and `messageArrivedComplete()` for more information. +See the Javadocs for `IMqttClient` `setManualAcks()` and `messageArrivedComplete()` for more information. For convenience a header accessor is provided: ==== @@ -164,6 +164,9 @@ StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge(); ---- ==== +Starting with version `5.2.11`, when the message converter throws an exception or returns `null` from the `MqttMessage` conversion, the `MqttPahoMessageDrivenChannelAdapter` sends an `ErrorMessage` into the `errorChannel`, if provided. +Re-throws this conversion error otherwise into an MQTT client callback. + ==== Configuring with Java Configuration The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration: