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.

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");

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);
+	}
+
+}

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 + "]";
+	}
+
+}

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);

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<Object>() {
 
 			@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<MqttCallback> callback = new AtomicReference<MqttCallback>();
 		doAnswer(new Answer<Object>() {
@@ -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<MqttIntegrationEvent> events = new LinkedBlockingQueue<MqttIntegrationEvent>();
+		doAnswer(new Answer<Void>() {
+
+			@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());
 	}
 
 }

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