From d8dc9d427e85e56612278e1249fdf186c83805c1 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 6 Mar 2017 14:34:10 -0500 Subject: [PATCH] AMQP-4238: Detect Subscription Failures and QOS JIRA: https://jira.spring.io/browse/INT-4238 Revert to using the sync client in the message-driven adapter so we can detect subscription failures (the sync client throws an exception). The only reason to use the async client was to timeout disconnects; this can be achieved with the sync client and `disconnectForcibly`. Also, the subscribe method updates the qos argument with the granted QOS values. Detect and log if any QOS does not match the request. Polishing Polishing - PR Comments --- .../MqttPahoMessageDrivenChannelAdapter.java | 50 +++-- .../integration/mqtt/MqttAdapterTests.java | 175 ++++++++++++++---- 2 files changed, 167 insertions(+), 58 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 07f3beedbb..fa1791646c 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 @@ -20,9 +20,10 @@ import java.util.Arrays; import java.util.Date; import java.util.concurrent.ScheduledFuture; -import org.eclipse.paho.client.mqttv3.IMqttAsyncClient; +import org.eclipse.paho.client.mqttv3.IMqttClient; 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.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; @@ -54,7 +55,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv private final MqttPahoClientFactory clientFactory; - private volatile IMqttAsyncClient client; + private volatile IMqttClient client; private volatile ScheduledFuture reconnectFuture; @@ -110,7 +111,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } /** - * Set the completion timeout for async operations. Not settable using the namespace. + * Set the completion timeout for operations. Not settable using the namespace. * Default 30000 milliseconds. * @param completionTimeout The timeout. * @since 4.1 @@ -159,16 +160,14 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS) || (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN) && this.cleanSession)) { - this.client.unsubscribe(getTopic()) - .waitForCompletion(this.completionTimeout); + this.client.unsubscribe(getTopic()); } } catch (MqttException e) { logger.error("Exception while unsubscribing", e); } try { - this.client.disconnect() - .waitForCompletion(this.completionTimeout); + this.client.disconnectForcibly(this.completionTimeout); } catch (MqttException e) { logger.error("Exception while disconnecting", e); @@ -190,8 +189,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv try { super.addTopic(topic, qos); if (this.client != null && this.client.isConnected()) { - this.client.subscribe(topic, qos) - .waitForCompletion(this.completionTimeout); + this.client.subscribe(topic, qos); } } catch (MqttException e) { @@ -208,8 +206,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv this.topicLock.lock(); try { if (this.client != null && this.client.isConnected()) { - this.client.unsubscribe(topic) - .waitForCompletion(this.completionTimeout); + this.client.unsubscribe(topic); } super.removeTopic(topic); } @@ -230,23 +227,36 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } Assert.state(getUrl() != null || connectionOptions.getServerURIs() != null, "If no 'url' provided, connectionOptions.getServerURIs() must not be null"); - this.client = this.clientFactory.getAsyncClientInstance(getUrl(), getClientId()); + this.client = this.clientFactory.getClientInstance(getUrl(), getClientId()); this.client.setCallback(this); + if (this.client instanceof MqttClient) { + ((MqttClient) this.client).setTimeToWait(this.completionTimeout); + } this.topicLock.lock(); + String[] topics = getTopic(); try { - this.client.connect(connectionOptions) - .waitForCompletion(this.completionTimeout); - this.client.subscribe(getTopic(), getQos()) - .waitForCompletion(this.completionTimeout); + this.client.connect(connectionOptions); + int[] requestedQos = getQos(); + int[] grantedQos = Arrays.copyOf(requestedQos, requestedQos.length); + this.client.subscribe(topics, grantedQos); + for (int i = 0; i < requestedQos.length; i++) { + if (grantedQos[i] != requestedQos[i]) { + if (logger.isWarnEnabled()) { + logger.warn("Granted QOS different to Requested QOS; topics: " + Arrays.toString(topics) + + " requested: " + Arrays.toString(requestedQos) + + " granted: " + Arrays.toString(grantedQos)); + } + break; + } + } } catch (MqttException 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); + logger.error("Error connecting or subscribing to " + Arrays.toString(topics), e); + this.client.disconnectForcibly(this.completionTimeout); throw e; } finally { @@ -254,7 +264,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } if (this.client.isConnected()) { this.connected = true; - String message = "Connected and subscribed to " + Arrays.asList(getTopic()); + String message = "Connected and subscribed to " + Arrays.toString(topics); if (logger.isDebugEnabled()) { logger.debug(message); } 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 c4af79a4ee..d4590f1c08 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 @@ -16,6 +16,7 @@ package org.springframework.integration.mqtt; +import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.junit.Assert.assertEquals; @@ -23,19 +24,22 @@ 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.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.doReturn; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.util.Properties; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -49,9 +53,11 @@ import javax.net.SocketFactory; import org.aopalliance.intercept.MethodInterceptor; import org.apache.commons.logging.Log; +import org.eclipse.paho.client.mqttv3.IMqttClient; import org.eclipse.paho.client.mqttv3.IMqttToken; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; +import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttException; @@ -60,6 +66,7 @@ import org.eclipse.paho.client.mqttv3.MqttSecurityException; import org.eclipse.paho.client.mqttv3.MqttToken; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.junit.Test; +import org.mockito.internal.stubbing.answers.CallsRealMethods; import org.springframework.aop.framework.ProxyFactoryBean; import org.springframework.beans.DirectFieldAccessor; @@ -85,6 +92,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.ReflectionUtils; /** * @author Gary Russell @@ -151,7 +159,7 @@ public class MqttAdapterTests { factory = spy(factory); final MqttAsyncClient client = mock(MqttAsyncClient.class); - doAnswer(invocation -> client).when(factory).getAsyncClientInstance(anyString(), anyString()); + willAnswer(invocation -> client).given(factory).getAsyncClientInstance(anyString(), anyString()); MqttPahoMessageHandler handler = new MqttPahoMessageHandler("foo", "bar", factory); handler.setDefaultTopic("mqtt-foo"); @@ -161,7 +169,7 @@ public class MqttAdapterTests { final MqttToken token = mock(MqttToken.class); final AtomicBoolean connectCalled = new AtomicBoolean(); - doAnswer(invocation -> { + willAnswer(invocation -> { MqttConnectOptions options = invocation.getArgument(0); assertEquals(23, options.getConnectionTimeout()); assertEquals(45, options.getKeepAliveInterval()); @@ -174,18 +182,18 @@ public class MqttAdapterTests { assertEquals(2, options.getWillMessage().getQos()); connectCalled.set(true); return token; - }).when(client).connect(any(MqttConnectOptions.class)); - doReturn(token).when(client).subscribe(any(String[].class), any(int[].class)); + }).given(client).connect(any(MqttConnectOptions.class)); + willReturn(token).given(client).subscribe(any(String[].class), any(int[].class)); final MqttDeliveryToken deliveryToken = mock(MqttDeliveryToken.class); final AtomicBoolean publishCalled = new AtomicBoolean(); - doAnswer(invocation -> { + willAnswer(invocation -> { assertEquals("mqtt-foo", invocation.getArguments()[0]); MqttMessage message = invocation.getArgument(1); assertEquals("Hello, world!", new String(message.getPayload())); publishCalled.set(true); return deliveryToken; - }).when(client).publish(anyString(), any(MqttMessage.class)); + }).given(client).publish(anyString(), any(MqttMessage.class)); handler.handleMessage(new GenericMessage("Hello, world!")); @@ -211,17 +219,16 @@ public class MqttAdapterTests { factory.setWill(will); factory = spy(factory); - final MqttAsyncClient client = mock(MqttAsyncClient.class); - doAnswer(invocation -> client).when(factory).getAsyncClientInstance(anyString(), anyString()); + final IMqttClient client = mock(IMqttClient.class); + willAnswer(invocation -> client).given(factory).getClientInstance(anyString(), anyString()); - 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(invocation -> { + willAnswer(invocation -> { if (failConnection.get()) { failInProcess.countDown(); waitToFail.await(10, TimeUnit.SECONDS); @@ -239,18 +246,16 @@ public class MqttAdapterTests { 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(); + return null; + }).given(client).connect(any(MqttConnectOptions.class)); final AtomicReference callback = new AtomicReference(); - doAnswer(invocation -> { + willAnswer(invocation -> { callback.set(invocation.getArgument(0)); return null; - }).when(client).setCallback(any(MqttCallback.class)); + }).given(client).setCallback(any(MqttCallback.class)); - when(client.isConnected()).thenReturn(true); + given(client.isConnected()).willReturn(true); MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, "baz", "fix"); @@ -262,10 +267,10 @@ public class MqttAdapterTests { adapter.setBeanFactory(mock(BeanFactory.class)); ApplicationEventPublisher applicationEventPublisher = mock(ApplicationEventPublisher.class); final BlockingQueue events = new LinkedBlockingQueue(); - doAnswer(invocation -> { + willAnswer(invocation -> { events.add(invocation.getArgument(0)); return null; - }).when(applicationEventPublisher).publishEvent(any(MqttIntegrationEvent.class)); + }).given(applicationEventPublisher).publishEvent(any(MqttIntegrationEvent.class)); adapter.setApplicationEventPublisher(applicationEventPublisher); adapter.setRecoveryInterval(500); adapter.afterPropertiesSet(); @@ -313,7 +318,7 @@ public class MqttAdapterTests { @Test public void testStopActionDefault() throws Exception { - final MqttAsyncClient client = mock(MqttAsyncClient.class); + final IMqttClient client = mock(IMqttClient.class); MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, null); adapter.start(); @@ -323,7 +328,7 @@ public class MqttAdapterTests { @Test public void testStopActionDefaultNotClean() throws Exception { - final MqttAsyncClient client = mock(MqttAsyncClient.class); + final IMqttClient client = mock(IMqttClient.class); MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, false, null); adapter.start(); @@ -333,7 +338,7 @@ public class MqttAdapterTests { @Test public void testStopActionAlways() throws Exception { - final MqttAsyncClient client = mock(MqttAsyncClient.class); + final IMqttClient client = mock(IMqttClient.class); MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, false, ConsumerStopAction.UNSUBSCRIBE_ALWAYS); @@ -344,7 +349,7 @@ public class MqttAdapterTests { @Test public void testStopActionNever() throws Exception { - final MqttAsyncClient client = mock(MqttAsyncClient.class); + final IMqttClient client = mock(IMqttClient.class); MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER); adapter.start(); @@ -379,7 +384,7 @@ public class MqttAdapterTests { @Test public void testReconnect() throws Exception { - final MqttAsyncClient client = mock(MqttAsyncClient.class); + final IMqttClient client = mock(IMqttClient.class); MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER); adapter.setRecoveryInterval(10); Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class)); @@ -405,12 +410,110 @@ public class MqttAdapterTests { taskScheduler.destroy(); } - private MqttPahoMessageDrivenChannelAdapter buildAdapter(final MqttAsyncClient client, Boolean cleanSession, + @Test + public void testSubscribeFailure() throws Exception { + DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); + factory.setCleanSession(false); + factory.setConnectionTimeout(23); + factory.setKeepAliveInterval(45); + factory.setPassword("pass"); + MemoryPersistence persistence = new MemoryPersistence(); + factory.setPersistence(persistence); + final SocketFactory socketFactory = mock(SocketFactory.class); + factory.setSocketFactory(socketFactory); + final Properties props = new Properties(); + factory.setSslProperties(props); + factory.setUserName("user"); + Will will = new Will("foo", "bar".getBytes(), 2, true); + factory.setWill(will); + + factory = spy(factory); + MqttAsyncClient aClient = mock(MqttAsyncClient.class); + final MqttClient client = mock(MqttClient.class); + willAnswer(invocation -> client).given(factory).getClientInstance(anyString(), anyString()); + given(client.isConnected()).willReturn(true); + 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)); + willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any()); + + IMqttToken token = mock(IMqttToken.class); + given(token.getGrantedQos()).willReturn(new int[] { 0x80 }); + willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), any(), any()); + + MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, + "baz", "fix"); + AtomicReference method = new AtomicReference<>(); + ReflectionUtils.doWithMethods(MqttPahoMessageDrivenChannelAdapter.class, m -> { + m.setAccessible(true); + method.set(m); + }, m -> m.getName().equals("connectAndSubscribe")); + assertNotNull(method.get()); + try { + method.get().invoke(adapter); + fail("Expected InvocationTargetException"); + } + catch (InvocationTargetException e) { + assertThat(e.getCause(), instanceOf(MqttException.class)); + assertThat(((MqttException) e.getCause()).getReasonCode(), + equalTo((int) MqttException.REASON_CODE_SUBSCRIBE_FAILED)); + } + } + + @Test + public void testDifferentQos() throws Exception { + DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); + factory.setCleanSession(false); + factory.setConnectionTimeout(23); + factory.setKeepAliveInterval(45); + factory.setPassword("pass"); + MemoryPersistence persistence = new MemoryPersistence(); + factory.setPersistence(persistence); + final SocketFactory socketFactory = mock(SocketFactory.class); + factory.setSocketFactory(socketFactory); + final Properties props = new Properties(); + factory.setSslProperties(props); + factory.setUserName("user"); + Will will = new Will("foo", "bar".getBytes(), 2, true); + factory.setWill(will); + + factory = spy(factory); + MqttAsyncClient aClient = mock(MqttAsyncClient.class); + final MqttClient client = mock(MqttClient.class); + willAnswer(invocation -> client).given(factory).getClientInstance(anyString(), anyString()); + given(client.isConnected()).willReturn(true); + 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)); + willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any()); + + IMqttToken token = mock(IMqttToken.class); + given(token.getGrantedQos()).willReturn(new int[] { 2, 0 }); + willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), any(), any()); + + MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, + "baz", "fix"); + AtomicReference method = new AtomicReference<>(); + ReflectionUtils.doWithMethods(MqttPahoMessageDrivenChannelAdapter.class, m -> { + m.setAccessible(true); + method.set(m); + }, m -> m.getName().equals("connectAndSubscribe")); + assertNotNull(method.get()); + Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class)); + new DirectFieldAccessor(adapter).setPropertyValue("logger", logger); + given(logger.isWarnEnabled()).willReturn(true); + method.get().invoke(adapter); + verify(logger, atLeastOnce()) + .warn("Granted QOS different to Requested QOS; topics: [baz, fix] requested: [1, 1] granted: [2, 0]"); + verify(client).setTimeToWait(30_000L); + } + + private MqttPahoMessageDrivenChannelAdapter buildAdapter(final IMqttClient client, Boolean cleanSession, ConsumerStopAction action) throws MqttException, MqttSecurityException { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() { @Override - public MqttAsyncClient getAsyncClientInstance(String uri, String clientId) throws MqttException { + public IMqttClient getClientInstance(String uri, String clientId) throws MqttException { return client; } @@ -422,11 +525,7 @@ public class MqttAdapterTests { if (action != null) { factory.setConsumerStopAction(action); } - when(client.connect(any(MqttConnectOptions.class))).thenReturn(this.alwaysComplete); - when(client.subscribe(any(String[].class), any(int[].class))).thenReturn(this.alwaysComplete); - when(client.disconnect()).thenReturn(this.alwaysComplete); - when(client.unsubscribe(any(String[].class))).thenReturn(this.alwaysComplete); - when(client.isConnected()).thenReturn(true); + given(client.isConnected()).willReturn(true); MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client", factory, "foo"); adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class)); adapter.setOutputChannel(new NullChannel()); @@ -435,18 +534,18 @@ public class MqttAdapterTests { return adapter; } - private void verifyUnsubscribe(MqttAsyncClient client) throws Exception { + private void verifyUnsubscribe(IMqttClient client) throws Exception { verify(client).connect(any(MqttConnectOptions.class)); verify(client).subscribe(any(String[].class), any(int[].class)); verify(client).unsubscribe(any(String[].class)); - verify(client).disconnect(); + verify(client).disconnectForcibly(anyLong()); } - private void verifyNotUnsubscribe(MqttAsyncClient client) throws Exception { + private void verifyNotUnsubscribe(IMqttClient client) throws Exception { verify(client).connect(any(MqttConnectOptions.class)); verify(client).subscribe(any(String[].class), any(int[].class)); verify(client, never()).unsubscribe(any(String[].class)); - verify(client).disconnect(); + verify(client).disconnectForcibly(anyLong()); } @Configuration