From b486247105c05284c9187a22697262e47a1c453d Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 5 Nov 2015 14:43:59 -0500 Subject: [PATCH] INT-3878: MQTT Application Events (inbound) JIRA: https://jira.spring.io/browse/INT-3878 Publish `ApplicationEvent`s to report inbound channel adapter connection status. Fixing typos and simple polishing. --- .../expression/ParentContextTests.java | 2 +- .../mqtt/event/MqttConnectionFailedEvent.java | 30 ++++++++ .../mqtt/event/MqttSubscribedEvent.java | 42 +++++++++++ .../MqttPahoMessageDrivenChannelAdapter.java | 64 +++++++++++++---- .../integration/mqtt/MqttAdapterTests.java | 69 ++++++++++++++++++- src/reference/asciidoc/mqtt.adoc | 7 ++ 6 files changed, 198 insertions(+), 16 deletions(-) create mode 100644 spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttConnectionFailedEvent.java create mode 100644 spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttSubscribedEvent.java diff --git a/spring-integration-core/src/test/java/org/springframework/integration/expression/ParentContextTests.java b/spring-integration-core/src/test/java/org/springframework/integration/expression/ParentContextTests.java index 122b1a11ff..ece3c51ce1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/expression/ParentContextTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/expression/ParentContextTests.java @@ -90,7 +90,7 @@ public class ParentContextTests { assertEquals(4, parentFunctions.size()); Object jsonPath = parentFunctions.get("jsonPath"); assertNotNull(jsonPath); - assertThat(jsonPath, Matchers.isOneOf(JsonPathUtils.class.getMethods())); + assertThat((Method) jsonPath, Matchers.isOneOf(JsonPathUtils.class.getMethods())); assertEquals(2, evalContexts.size()); ClassPathXmlApplicationContext child = new ClassPathXmlApplicationContext(parent); child.setConfigLocation("org/springframework/integration/expression/ChildContext-context.xml"); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttConnectionFailedEvent.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttConnectionFailedEvent.java new file mode 100644 index 0000000000..fc043742e6 --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttConnectionFailedEvent.java @@ -0,0 +1,30 @@ +/* + * Copyright 2015 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.event; + +/** + * @author Gary Russell + * @since 4.2.2 + * + */ +@SuppressWarnings("serial") +public class MqttConnectionFailedEvent extends MqttIntegrationEvent { + + public MqttConnectionFailedEvent(Object source, Throwable cause) { + super(source, cause); + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttSubscribedEvent.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttSubscribedEvent.java new file mode 100644 index 0000000000..ba0f3a2e47 --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttSubscribedEvent.java @@ -0,0 +1,42 @@ +/* + * Copyright 2015 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.event; + +/** + * @author Gary Russell + * @since 4.2.2 + * + */ +@SuppressWarnings("serial") +public class MqttSubscribedEvent extends MqttIntegrationEvent { + + private final String message; + + public MqttSubscribedEvent(Object source, String message) { + super(source); + this.message = message; + } + + public String getMessage() { + return message; + } + + @Override + public String toString() { + return "MqttSubscribedEvent [message=" + message + ", source=" + source + "]"; + } + +} 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 b802bf6aef..ab25f4201f 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 @@ -25,8 +25,12 @@ import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; +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.util.Assert; @@ -39,10 +43,12 @@ import org.springframework.util.Assert; * */ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDrivenChannelAdapter - implements MqttCallback { + implements MqttCallback, ApplicationEventPublisherAware { private static final int DEFAULT_COMPLETION_TIMEOUT = 30000; + private static final int DEFAULT_RECOVERY_INTERVAL = 10000; + private final MqttPahoClientFactory clientFactory; private volatile MqttAsyncClient client; @@ -53,6 +59,9 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv private volatile int completionTimeout = DEFAULT_COMPLETION_TIMEOUT; + private volatile int recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + + private ApplicationEventPublisher applicationEventPublisher; /** * Use this constructor for a single url (although it may be overridden @@ -103,11 +112,29 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv this.completionTimeout = completionTimeout; } + /** + * The time (ms) to wait between reconnection attempts. + * Default {@value #DEFAULT_RECOVERY_INTERVAL}. + * @param recoveryInterval the interval. + * @since 4.2.2 + */ + public void setRecoveryInterval(int recoveryInterval) { + this.recoveryInterval = recoveryInterval; + } + + /** + * @since 4.2.2 + */ + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + @Override protected void doStart() { super.doStart(); try { - this.connectAndSubscribe(); + connectAndSubscribe(); } catch (Exception e) { logger.error("Exception while connecting and subscribing, retrying", e); @@ -117,11 +144,11 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override protected void doStop() { - this.cancelReconnect(); + cancelReconnect(); super.doStop(); if (this.client != null) { try { - this.client.unsubscribe(this.getTopic()) + this.client.unsubscribe(getTopic()) .waitForCompletion(this.completionTimeout); } catch (MqttException e) { @@ -184,20 +211,23 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv private void connectAndSubscribe() throws MqttException { MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions(); - Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null, + Assert.state(getUrl() != null || connectionOptions.getServerURIs() != null, "If no 'url' provided, connectionOptions.getServerURIs() must not be null"); - this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId()); + this.client = this.clientFactory.getAsyncClientInstance(getUrl(), getClientId()); this.client.setCallback(this); this.topicLock.lock(); try { this.client.connect(connectionOptions) .waitForCompletion(this.completionTimeout); - this.client.subscribe(this.getTopic(), this.getQos()) + this.client.subscribe(getTopic(), getQos()) .waitForCompletion(this.completionTimeout); } catch (MqttException e) { - logger.error("Error connecting or subscribing to " + Arrays.asList(this.getTopic()), e); + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, e)); + } + logger.error("Error connecting or subscribing to " + Arrays.asList(getTopic()), e); this.client.disconnect() .waitForCompletion(this.completionTimeout); throw e; @@ -208,10 +238,14 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv if (this.client.isConnected()) { this.connected = true; if (this.reconnectFuture != null) { - this.cancelReconnect(); + cancelReconnect(); } + String message = "Connected and subscribed to " + Arrays.asList(getTopic()); if (logger.isDebugEnabled()) { - logger.debug("Connected and subscribed to " + Arrays.asList(this.getTopic())); + logger.debug(message); + } + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, message)); } } } @@ -241,7 +275,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv logger.error("Exception while connecting and subscribing", e); } } - }, 10000); + + }, this.recoveryInterval); } catch (Exception e) { logger.error("Failed to schedule reconnect", e); @@ -252,14 +287,17 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv public void connectionLost(Throwable cause) { this.logger.error("Lost connection:" + cause.getMessage() + "; retrying..."); this.connected = false; - this.scheduleReconnect(); + scheduleReconnect(); + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, cause)); + } } @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { Message message = this.getConverter().toMessage(topic, mqttMessage); try { - this.sendMessage(message); + sendMessage(message); } catch (RuntimeException e) { logger.error("Unhandled exception for " + message.toString(), e); 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 0b83b4c25f..7b58baf3f6 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-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -15,9 +15,11 @@ */ package org.springframework.integration.mqtt; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; @@ -30,6 +32,10 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.util.Properties; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -39,6 +45,7 @@ import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttToken; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; @@ -47,9 +54,13 @@ import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory.Will; +import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent; +import org.springframework.integration.mqtt.event.MqttIntegrationEvent; +import org.springframework.integration.mqtt.event.MqttSubscribedEvent; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.messaging.Message; @@ -196,10 +207,20 @@ public class MqttAdapterTests { final MqttToken token = mock(MqttToken.class); final AtomicBoolean connectCalled = new AtomicBoolean(); + final AtomicBoolean failConnection = new AtomicBoolean(); + final CountDownLatch waitToFail = new CountDownLatch(1); + final CountDownLatch failInProcess = new CountDownLatch(1); + final CountDownLatch goodConnection = new CountDownLatch(2); + final MqttException reconnectException = new MqttException(MqttException.REASON_CODE_SERVER_CONNECT_ERROR); doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { + if (failConnection.get()) { + failInProcess.countDown(); + waitToFail.await(10, TimeUnit.SECONDS); + throw reconnectException; + } MqttConnectOptions options = (MqttConnectOptions) invocation.getArguments()[0]; assertEquals(23, options.getConnectionTimeout()); assertEquals(45, options.getKeepAliveInterval()); @@ -211,10 +232,12 @@ public class MqttAdapterTests { assertEquals("bar", new String(options.getWillMessage().getPayload())); assertEquals(2, options.getWillMessage().getQos()); connectCalled.set(true); + goodConnection.countDown(); return token; } }).when(client).connect(any(MqttConnectOptions.class)); doReturn(token).when(client).subscribe(any(String[].class), any(int[].class)); + doReturn(token).when(client).disconnect(); final AtomicReference callback = new AtomicReference(); doAnswer(new Answer() { @@ -228,13 +251,26 @@ public class MqttAdapterTests { when(client.isConnected()).thenReturn(true); - MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, "baz"); + MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, + "baz", "fix"); QueueChannel outputChannel = new QueueChannel(); adapter.setOutputChannel(outputChannel); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.setBeanFactory(mock(BeanFactory.class)); + ApplicationEventPublisher applicationEventPublisher = mock(ApplicationEventPublisher.class); + final BlockingQueue events = new LinkedBlockingQueue(); + doAnswer(new Answer() { + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + events.add((MqttIntegrationEvent) invocation.getArguments()[0]); + return null; + } + }).when(applicationEventPublisher).publishEvent(any(MqttIntegrationEvent.class)); + adapter.setApplicationEventPublisher(applicationEventPublisher); + adapter.setRecoveryInterval(500); adapter.afterPropertiesSet(); adapter.start(); @@ -246,6 +282,35 @@ public class MqttAdapterTests { Message outMessage = outputChannel.receive(0); assertNotNull(outMessage); assertEquals("qux", outMessage.getPayload()); + + MqttIntegrationEvent event = events.poll(10, TimeUnit.SECONDS); + assertThat(event, instanceOf(MqttSubscribedEvent.class)); + assertEquals("Connected and subscribed to [baz, fix]", ((MqttSubscribedEvent) event).getMessage()); + + // lose connection and make first reconnect fail + failConnection.set(true); + RuntimeException e = new RuntimeException("foo"); + adapter.connectionLost(e); + + event = events.poll(10, TimeUnit.SECONDS); + assertThat(event, instanceOf(MqttConnectionFailedEvent.class)); + assertSame(event.getCause(), e); + + assertTrue(failInProcess.await(10, TimeUnit.SECONDS)); + waitToFail.countDown(); + failConnection.set(false); + event = events.poll(10, TimeUnit.SECONDS); + assertThat(event, instanceOf(MqttConnectionFailedEvent.class)); + assertSame(event.getCause(), reconnectException); + + // reconnect can now succeed; however, we might have other failures on a slow server (500ms retry). + assertTrue(goodConnection.await(10, TimeUnit.SECONDS)); + int n = 0; + while (!(event instanceof MqttSubscribedEvent) && n++ < 20) { + event = events.poll(10, TimeUnit.SECONDS); + } + assertThat(event, instanceOf(MqttSubscribedEvent.class)); + assertEquals("Connected and subscribed to [baz, fix]", ((MqttSubscribedEvent) event).getMessage()); } } diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index 33a9cf74d7..ef8c843176 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -80,7 +80,14 @@ The `DefaultPahoMessageConverter` can be configured to return the raw `byte[]` i NOTE: Starting with _version 4.1_ the url can be omitted and, instead, the server URIs can be provided in the `serverURIs` property of the `DefaultMqttPahoClientFactory`. This enables, for example, connection to a highly available (HA) cluster. +Starting with _version 4.2.2_, an `MqttSubscribedEvent` is published when the adapter successfully subscribes to the +topic(s). +`MqttConnectionFailedEvent` s are published when the connection/subscription fails. +These events can be received by a bean that implements `ApplicationListener`. +Also, a new property `recoveryInterval` controls the interval at which the adapter will attempt to reconnect after +a failure; it defaults to `10000ms` (ten seconds). +This is not currently available using XML configuration. ==== Adding/Removing Topics at Runtime