From 71d41757cd9655b41e334926968d247cf9e267e1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 8 Jan 2021 11:29:59 -0500 Subject: [PATCH] GH-3454: From MQTT conversion error - to error ch (#3456) * GH-3454: From MQTT conversion error - to error ch Fixes https://github.com/spring-projects/spring-integration/issues/3454 The message converter may return null when we try to covert from the MQTT message. The thrown exception may also reset the client connect. * Fix `MqttPahoMessageDrivenChannelAdapter` to catch any conversion errors (including `null` result) and try to send an `ErrorMessage` with that info into the provided `errorChannel`. Otherwise re-throw it as as **Cherry-pick to `5.4.x` & `5.3.x`** * * Apply review language-specific changes # Conflicts: # spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java # spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java # src/reference/asciidoc/mqtt.adoc --- .../MqttPahoMessageDrivenChannelAdapter.java | 53 ++++++++++++--- .../integration/mqtt/MqttAdapterTests.java | 67 ++++++++++++++----- src/reference/asciidoc/mqtt.adoc | 3 + 3 files changed, 96 insertions(+), 27 deletions(-) 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 7b86d1bddd..4846b474ad 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. @@ -37,6 +37,8 @@ import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent; import org.springframework.integration.mqtt.event.MqttSubscribedEvent; 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; /** @@ -169,7 +171,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } catch (Exception e) { logger.error("Exception while connecting and subscribing, retrying", e); - this.scheduleReconnect(); + scheduleReconnect(); } } @@ -181,7 +183,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv try { if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS) || (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN) - && this.cleanSession)) { + && this.cleanSession)) { this.client.unsubscribe(getTopic()); } @@ -237,7 +239,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(); @@ -365,14 +367,47 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override public void messageArrived(String topic, MqttMessage mqttMessage) { - Message message = this.getConverter().toMessage(topic, mqttMessage); + Message message = toMessage(topic, mqttMessage); + if (message != null) { + try { + sendMessage(message); + } + catch (RuntimeException e) { + logger.error("Unhandled exception for " + message.toString(), e); + throw e; + } + } + } + + private Message toMessage(String topic, MqttMessage mqttMessage) { + Message message = null; + RuntimeException conversionError = null; try { - sendMessage(message); + message = getConverter().toMessage(topic, mqttMessage); } - catch (RuntimeException e) { - logger.error("Unhandled exception for " + message.toString(), e); - throw e; + catch (RuntimeException ex) { + conversionError = ex; } + + if (message == null && conversionError == null) { + conversionError = new IllegalStateException("'MqttMessageConverter' returned 'null'"); + } + + if (conversionError != null) { + GenericMessage failedMessage = new GenericMessage<>(mqttMessage); + if (!sendErrorMessageIfNecessary(failedMessage, conversionError)) { + MessageConversionException conversionException; + if (conversionError instanceof MessageConversionException) { + conversionException = (MessageConversionException) conversionError; + } + else { + conversionException = new MessageConversionException(failedMessage, + "Failed to convert from MQTT Message", conversionError); + } + throw conversionException; + } + } + return message; } @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 a9077e742f..ad0c56065e 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. @@ -85,9 +85,12 @@ import org.springframework.integration.mqtt.event.MqttSubscribedEvent; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; +import org.springframework.integration.mqtt.support.MqttMessageConverter; 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; @@ -102,7 +105,7 @@ import org.springframework.util.ReflectionUtils; */ public class MqttAdapterTests { - private IMqttToken alwaysComplete; + private final IMqttToken alwaysComplete; { ProxyFactoryBean pfb = new ProxyFactoryBean(); @@ -194,12 +197,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(); @@ -254,7 +257,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; @@ -266,12 +269,14 @@ public class MqttAdapterTests { "baz", "fix"); 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; @@ -294,6 +299,33 @@ 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 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(ErrorMessage::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"); @@ -416,7 +448,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(); @@ -448,12 +480,11 @@ 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); - given(token.getGrantedQos()).willReturn(new int[] { 0x80 }); + given(token.getGrantedQos()).willReturn(new int[]{ 0x80 }); willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any()); MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, @@ -465,11 +496,11 @@ public class MqttAdapterTests { }, m -> m.getName().equals("connectAndSubscribe")); assertThat(method.get()).isNotNull(); Condition subscribeFailed = new Condition<>(ex -> - ((MqttException) ex.getCause()).getReasonCode() == MqttException.REASON_CODE_SUBSCRIBE_FAILED, - "expected the reason code to be REASON_CODE_SUBSCRIBE_FAILED"); + ((MqttException) ex.getCause()).getReasonCode() == MqttException.REASON_CODE_SUBSCRIBE_FAILED, + "expected the reason code to be REASON_CODE_SUBSCRIBE_FAILED"); assertThatExceptionOfType(InvocationTargetException.class).isThrownBy(() -> method.get().invoke(adapter)) - .withCauseInstanceOf(MqttException.class) - .is(subscribeFailed); + .withCauseInstanceOf(MqttException.class) + .is(subscribeFailed); } @Test @@ -502,7 +533,7 @@ public class MqttAdapterTests { willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any()); IMqttToken token = mock(IMqttToken.class); - given(token.getGrantedQos()).willReturn(new int[] { 2, 0 }); + given(token.getGrantedQos()).willReturn(new int[]{ 2, 0 }); willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any()); MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, @@ -537,7 +568,7 @@ public class MqttAdapterTests { }; MqttConnectOptions connectOptions = new MqttConnectOptions(); - connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" }); + connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" }); if (cleanSession != null) { connectOptions.setCleanSession(cleanSession); } @@ -554,7 +585,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 @@ -564,7 +595,7 @@ public class MqttAdapterTests { }; MqttConnectOptions connectOptions = new MqttConnectOptions(); - connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" }); + connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" }); factory.setConnectionOptions(connectOptions); MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("client", factory); adapter.setDefaultTopic("foo"); diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index 591967512a..1c01778ddc 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -145,6 +145,9 @@ A new application context reverts to the configured settings. Changing the topics while the adapter is stopped (or disconnected from the broker) takes effect the next time a connection is established. +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: