diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java index 55b0717..09e155b 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java @@ -19,6 +19,7 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.w3c.dom.Element; @@ -41,6 +42,7 @@ public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapte MqttParserUtils.parseCommon(element, builder); builder.addConstructorArgValue(element.getAttribute("topics")); builder.addPropertyReference("outputChannel", channelName); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); return builder.getBeanDefinition(); } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java~ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java~ new file mode 100644 index 0000000..b86b774 --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java~ @@ -0,0 +1,49 @@ +/* + * Copyright 2002-2013 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt.config.xml; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelAdapterParser; +import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; +import org.w3c.dom.Element; + +/** + * The MqttAdapter Message Driven Channel adapter parser + * + * @author Gary Russell + * @since 1.0 + * + */ +public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapterParser { + + + @Override + protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { + + BeanDefinitionBuilder builder = BeanDefinitionBuilder + .genericBeanDefinition(MqttPahoMessageDrivenChannelAdapter.class); + + MqttParserUtils.parseCommon(element, builder); + builder.addConstructorArgValue(element.getAttribute("topics")); + builder.addPropertyReference("outputChannel", channelName); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + + return builder.getBeanDefinition(); + } + +} 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 b66286f..d4305c7 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 @@ -15,6 +15,7 @@ */ package org.springframework.integration.mqtt.inbound; +import java.util.Arrays; import java.util.concurrent.ScheduledFuture; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; @@ -111,7 +112,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv this.cancelReconnect(); } if (logger.isDebugEnabled()) { - logger.debug("Connected and subscribed to " + this.getTopic()); + logger.debug("Connected and subscribed to " + Arrays.asList(this.getTopic())); } } } @@ -158,7 +159,13 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { Message message = this.getConverter().toMessage(topic, mqttMessage); - this.sendMessage(message); + try { + this.sendMessage(message); + } + catch (RuntimeException e) { + logger.error("Unhandled exception for " + message.toString(), e); + throw e; + } } @Override 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~ new file mode 100644 index 0000000..55b760a --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java~ @@ -0,0 +1,174 @@ +/* + * Copyright 2002-2013 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt.inbound; + +import java.util.concurrent.ScheduledFuture; + +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttCallback; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttException; +import org.eclipse.paho.client.mqttv3.MqttMessage; + +import org.springframework.integration.Message; +import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; +import org.springframework.integration.mqtt.core.MqttPahoClientFactory; + +/** + * Eclipse Paho Implementation. + * + * @author Gary Russell + * @since 1.0 + * + */ +public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDrivenChannelAdapter + implements MqttCallback { + + private final MqttPahoClientFactory clientFactory; + + private volatile MqttClient client; + + private volatile ScheduledFuture reconnectFuture; + + private volatile boolean connected; + + + public MqttPahoMessageDrivenChannelAdapter(String url, String clientId, MqttPahoClientFactory clientFactory, + String... topic) { + super(url, clientId, topic); + this.clientFactory = clientFactory; + } + + public MqttPahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) { + this(url, clientId, new DefaultMqttPahoClientFactory(), topic); + } + + @Override + protected void doStart() { + super.doStart(); + try { + this.connectAndSubscribe(); + } + catch (Exception e) { + logger.error("Exception while connecting and subscribing, retrying", e); + this.scheduleReconnect(); + } + } + + @Override + protected void doStop() { + this.cancelReconnect(); + super.doStop(); + try { + this.client.unsubscribe(this.getTopic()); + } + catch (MqttException e) { + logger.error("Exception while unsubscribing", e); + } + try { + this.client.disconnect(); + } + catch (MqttException e) { + logger.error("Exception while disconnecting", e); + } + try { + this.client.close(); + } + catch (MqttException e) { + logger.error("Exception while closing", e); + } + this.connected = false; + this.client = null; + } + + private void connectAndSubscribe() throws MqttException { + this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId()); + this.client.connect(this.clientFactory.getConnectionOptions()); + try { + this.client.subscribe(this.getTopic()); + } + catch (MqttException e) { + this.client.disconnect(); + throw e; + } + if (this.client.isConnected()) { + this.client.setCallback(this); + this.connected = true; + if (this.reconnectFuture != null) { + this.cancelReconnect(); + } + if (logger.isDebugEnabled()) { + logger.debug("Connected and subscribed to " + Arrays.asList(this.getTopic())); + } + } + } + + private synchronized void cancelReconnect() { + if (this.reconnectFuture != null) { + this.reconnectFuture.cancel(false); + this.reconnectFuture = null; + } + } + + private void scheduleReconnect() { + try { + this.reconnectFuture = this.getTaskScheduler().scheduleWithFixedDelay(new Runnable() { + + @Override + public void run() { + try { + if (logger.isDebugEnabled()) { + logger.debug("Attempting reconnect"); + } + if (!connected) { + connectAndSubscribe(); + } + } + catch (MqttException e) { + logger.error("Exception while connecting and subscribing", e); + } + } + }, 10000); + } + catch (Exception e) { + logger.error("Failed to schedule reconnect", e); + } + } + + @Override + public void connectionLost(Throwable cause) { + this.logger.error("Lost connection:" + cause.getMessage() + "; retrying..."); + this.connected = false; + this.scheduleReconnect(); + } + + @Override + public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { + Message message = this.getConverter().toMessage(topic, mqttMessage); + try { + this.sendMessage(message); + } + catch (RuntimeException e) { + logger.error("Unhandled exception for " + message.toString(), e); + throw e; + } + } + + @Override + public void deliveryComplete(IMqttDeliveryToken token) { + } + +} diff --git a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd index ecf59a9..3f11fe6 100644 --- a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd +++ b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd @@ -47,19 +47,29 @@ + + + + + + + + + If a downstream exception is thrown and an error-channel is specified, + the MessagingException will be sent to this channel. Otherwise, any such exception + will be logged. + + + diff --git a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd~ b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd~ new file mode 100644 index 0000000..ecf59a9 --- /dev/null +++ b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-1.0.xsd~ @@ -0,0 +1,200 @@ + + + + + + + + + + + + + + + The definition for the Spring Integration MqttAdapter + Inbound Channel Adapter. + + + + + + + + + + + + + + + + + Specifies one or more (comma-delimited) topics on which to listen for messages. + + + + + + + + + + + + + + + + Defines an outbound Channel Adapter. + + + + + + + + + + + + Channel from which messages will be output. + When a message is sent to this channel it will + cause the query + to be executed. + + + + + + + + + + + Specifies the order for invocation when this endpoint is connected as a + subscriber to a SubscribableChannel. + + + + + + + Specifies the default topic to which messages will be sent. Required if an + outbound message does not have an 'mqtt_topic' header. + + + + + + + Specifies the default quality of service. Default 0. + + + + + + + Specifies the default value of the 'retained' flag. Default false. + + + + + + + + + + + Identifies the underlying Spring bean definition, which is an + instance of either 'EventDrivenConsumer' or 'PollingConsumer', + depending on whether the component's input channel is a + 'SubscribableChannel' or 'PollableChannel'. + + + + + + + Flag to indicate that the component should start automatically + on startup (default true). + + + + + + + + + + Flag to indicate the phase in which the component should start automatically + on startup. See SmartLifecycle. + + + + + + + + + + MQTT broker URL. + + + + + + + MQTT client ID. + + + + + + + to/from + a paho MqttMessage. Default is DefaultMqttMessageConverter. + ]]> + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java index 57587d8..6c15e53 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java @@ -65,6 +65,7 @@ public class BrokerRunning extends TestWatcher { finally { if (client != null) { try { + client.disconnect(); client.close(); } catch (MqttException e) { diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java~ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java~ new file mode 100644 index 0000000..57587d8 --- /dev/null +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BrokerRunning.java~ @@ -0,0 +1,82 @@ +/* + * Copyright 2002-2013 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt; + +import static org.junit.Assume.assumeNoException; +import static org.junit.Assume.assumeTrue; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttException; +import org.junit.rules.TestWatcher; +import org.junit.runner.Description; +import org.junit.runners.model.Statement; +import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; + +/** + * @author Gary Russell + * @since 1.0 + * + */ +public class BrokerRunning extends TestWatcher { + + private static Log logger = LogFactory.getLog(BrokerRunning.class); + + // Static so that we only test once on failure: speeds up test suite + private static Map brokerOnline = new HashMap(); + + private final int port; + + private BrokerRunning(int port) { + this.port = port; + brokerOnline.put(port, true); + } + + @Override + public Statement apply(Statement base, Description description) { + assumeTrue(brokerOnline.get(port)); + String url = "tcp://localhost:" + port; + MqttClient client = null; + try { + client = new DefaultMqttPahoClientFactory().getClientInstance(url, "junit-" + System.currentTimeMillis()); + client.connect(); + } + catch (MqttException e) { + logger.warn("Tests not running because no broker on " + url + ":", e); + assumeNoException(e); + } + finally { + if (client != null) { + try { + client.close(); + } + catch (MqttException e) { + } + } + } + return super.apply(base, description); + } + + + + public static BrokerRunning isRunning(int port) { + return new BrokerRunning(port); + } +} diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests-context.xml b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests-context.xml new file mode 100644 index 0000000..8576476 --- /dev/null +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests-context.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java new file mode 100644 index 0000000..440c2da --- /dev/null +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java @@ -0,0 +1,139 @@ +/* + * Copyright 2014 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.contains; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; +import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class DownstreamExceptionTests { + + @ClassRule + public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883); + + @Autowired + private Service service; + + @Autowired + private MqttPahoMessageDrivenChannelAdapter noErrorChannel; + + @Autowired + private MqttPahoMessageDrivenChannelAdapter withErrorChannel; + + @Autowired + private PollableChannel errors; + + @Test + public void testNoErrorChannel() throws Exception { + service.n = 0; + Log logger = spy(TestUtils.getPropertyValue(noErrorChannel, "logger", Log.class)); + final CountDownLatch latch = new CountDownLatch(1); + doAnswer(new Answer() { + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + if (((String) invocation.getArguments()[0]).contains("Unhandled")) { + latch.countDown(); + } + return null; + } + }).when(logger).error(anyString(), any(Throwable.class)); + new DirectFieldAccessor(noErrorChannel).setPropertyValue("logger", logger); + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-fooEx1"); + adapter.afterPropertiesSet(); + adapter.start(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + service.barrier.reset(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + verify(logger).error(contains("Unhandled exception for"), any(Throwable.class)); + service.barrier.reset(); + adapter.stop(); + } + + @Test + public void testWithErrorChannel() throws Exception { + assertSame(this.errors, TestUtils.getPropertyValue(this.withErrorChannel, "errorChannel")); + service.n = 0; + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-fooEx2"); + adapter.afterPropertiesSet(); + adapter.start(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + service.barrier.reset(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + assertNotNull(errors.receive(10000)); + service.barrier.reset(); + adapter.stop(); + } + + public static class Service { + + public CyclicBarrier barrier = new CyclicBarrier(2); + + public int n; + + public void foo(String foo) throws Exception { + barrier.await(10, TimeUnit.SECONDS); + if (n++ > 0) { + throw new RuntimeException("bar"); + } + } + + } + +} diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java~ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java~ new file mode 100644 index 0000000..4effd3f --- /dev/null +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java~ @@ -0,0 +1,139 @@ +/* + * Copyright 2014 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.contains; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; +import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class DownstreamExceptionTests { + + @ClassRule + public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883); + + @Autowired + private Service service; + + @Autowired + private MqttPahoMessageDrivenChannelAdapter noErrorChannel; + + @Autowired + private MqttPahoMessageDrivenChannelAdapter withErrorChannel; + + @Autowired + private PollableChannel errors; + + @Test + public void testNoErrorChannel() throws Exception { + service.n = 0; + Log logger = spy(TestUtils.getPropertyValue(noErrorChannel, "logger", Log.class)); + final CountDownLatch latch = new CountDownLatch(1); + doAnswer(new Answer() { + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + if (((String) invocation.getArguments()[0]).contains("Unhandled")) { + latch.countDown(); + } + return null; + } + }).when(logger).error(anyString(), any(Throwable.class)); + new DirectFieldAccessor(noErrorChannel).setPropertyValue("logger", logger); + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-fooEx1"); + adapter.afterPropertiesSet(); + adapter.start(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + service.barrier.reset(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + verify(logger).error(contains("Unhandled exception for"), any(Throwable.class)); + service.barrier.reset(); + adapter.stop(); + } + + @Test + public void testWithErrorChannel() throws Exception { + assertSame(this.errors, TestUtils.getPropertyValue(this.withErrorChannel, "errorChannel")); + service.n = 0; + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-fooEx2"); + adapter.afterPropertiesSet(); + adapter.start(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + service.barrier.reset(); + adapter.handleMessage(new GenericMessage("foo")); + service.barrier.await(10, TimeUnit.SECONDS); + assertNotNull(errors.receive(10000)); + service.barrier.reset(); + adapter.stop(); + } + + public static class Service { + + public CyclicBarrier barrier = new CyclicBarrier(2); + + public int n; + + public void foo(String foo) throws Exception { + barrier.await(10, TimeUnit.SECONDS); + if (n++ > 0) { + throw new RuntimeException("bar"); + } + } + + } + +}