diff --git a/components/camel-paho-mqtt5/pom.xml b/components/camel-paho-mqtt5/pom.xml index 95213382a046e..5ce845becdf67 100644 --- a/components/camel-paho-mqtt5/pom.xml +++ b/components/camel-paho-mqtt5/pom.xml @@ -52,6 +52,11 @@ camel-test-junit6 test + + org.assertj + assertj-core + test + diff --git a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java index ef6413031742f..e15df423a9be6 100644 --- a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java +++ b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java @@ -58,91 +58,131 @@ public void setClient(MqttClient client) { protected void doStart() throws Exception { super.doStart(); - connectionOptions = getEndpoint().createMqttConnectionOptions(); - - if (client == null) { - clientId = getEndpoint().getConfiguration().getClientId(); - if (clientId == null) { - clientId = PahoMqtt5Endpoint.generateClientId(); - } - stopClient = true; - client = new MqttClient( - getEndpoint().getConfiguration().getBrokerUrl(), - clientId, - PahoMqtt5Endpoint.createMqttClientPersistence(getEndpoint().getConfiguration())); - LOG.debug("Connecting client: {} to broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl()); - if (getEndpoint().getConfiguration().isManualAcksEnabled()) { - client.setManualAcks(true); - + stopClient = client == null; + try { + connectionOptions = getEndpoint().createMqttConnectionOptions(); + + if (stopClient) { + clientId = getEndpoint().getConfiguration().getClientId(); + if (clientId == null) { + clientId = PahoMqtt5Endpoint.generateClientId(); + } + client = createClient(); + LOG.debug("Connecting client: {} to broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl()); + if (getEndpoint().getConfiguration().isManualAcksEnabled()) { + client.setManualAcks(true); + } + client.connect(connectionOptions); } - client.connect(connectionOptions); - } - client.setCallback(new MqttCallback() { + client.setCallback(new MqttCallback() { - @Override - public void connectComplete(boolean reconnect, String serverURI) { - if (reconnect) { - try { - client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); - } catch (MqttException e) { - LOG.error("MQTT resubscribe failed {}", e.getMessage(), e); + @Override + public void connectComplete(boolean reconnect, String serverURI) { + if (reconnect) { + try { + client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } catch (MqttException e) { + LOG.error("MQTT resubscribe failed {}", e.getMessage(), e); + } } } - } - @Override - public void authPacketArrived(int reasonCode, MqttProperties properties) { - LOG.debug("Auth packet arrived {} {}", reasonCode, properties); - } + @Override + public void authPacketArrived(int reasonCode, MqttProperties properties) { + LOG.debug("Auth packet arrived {} {}", reasonCode, properties); + } - @Override - public void disconnected(MqttDisconnectResponse response) { - LOG.debug("MQTT broker disconnected due {}", response.getReasonString(), response.getException()); - } + @Override + public void disconnected(MqttDisconnectResponse response) { + LOG.debug("MQTT broker disconnected due {}", response.getReasonString(), response.getException()); + } - @Override - public void mqttErrorOccurred(MqttException exception) { - LOG.debug("Error occurred {}", exception.getMessage(), exception); - } + @Override + public void mqttErrorOccurred(MqttException exception) { + LOG.debug("Error occurred {}", exception.getMessage(), exception); + } - @Override - public void messageArrived(String topic, MqttMessage message) throws Exception { - LOG.debug("Message arrived on topic: {} -> {}", topic, message); - Exchange exchange = createExchange(message, topic); + @Override + public void messageArrived(String topic, MqttMessage message) throws Exception { + LOG.debug("Message arrived on topic: {} -> {}", topic, message); + Exchange exchange = createExchange(message, topic); - // use default consumer callback - AsyncCallback cb = defaultConsumerCallback(exchange, true); - getAsyncProcessor().process(exchange, cb); - } + // use default consumer callback + AsyncCallback cb = defaultConsumerCallback(exchange, true); + getAsyncProcessor().process(exchange, cb); + } - @Override - public void deliveryComplete(IMqttToken token) { - LOG.debug("Delivery complete. Token: {}", token); - } - }); + @Override + public void deliveryComplete(IMqttToken token) { + LOG.debug("Delivery complete. Token: {}", token); + } + }); - LOG.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic()); - client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + LOG.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic()); + client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } catch (Exception startException) { + MqttClient ownedClient = stopClient ? client : null; + if (ownedClient != null) { + client = null; + stopClient = false; + closeOwnedClient(ownedClient, startException); + } + throw startException; + } } @Override protected void doStop() throws Exception { - super.doStop(); - - if (stopClient && client != null && client.isConnected()) { - String topic = getEndpoint().getTopic(); - // only unsubscribe if we are not durable - if (getEndpoint().getConfiguration().isCleanStart()) { - LOG.debug("Unsubscribing client: {} from topic: {}", clientId, topic); - client.unsubscribe(topic); - } else { - LOG.debug("Client: {} is durable so will not unsubscribe from topic: {}", clientId, topic); + MqttClient ownedClient = stopClient ? client : null; + Exception stopException = null; + try { + super.doStop(); + + if (ownedClient != null && ownedClient.isConnected()) { + String topic = getEndpoint().getTopic(); + // only unsubscribe if we are not durable + if (getEndpoint().getConfiguration().isCleanStart()) { + LOG.debug("Unsubscribing client: {} from topic: {}", clientId, topic); + ownedClient.unsubscribe(topic); + } else { + LOG.debug("Client: {} is durable so will not unsubscribe from topic: {}", clientId, topic); + } + LOG.debug("Disconnecting client: {} from broker: {}", clientId, + getEndpoint().getConfiguration().getBrokerUrl()); + ownedClient.disconnect(); + } + } catch (Exception e) { + stopException = e; + } finally { + client = null; + stopClient = false; + if (ownedClient != null) { + stopException = closeOwnedClient(ownedClient, stopException); + } + } + if (stopException != null) { + throw stopException; + } + } + + MqttClient createClient() throws MqttException { + return new MqttClient( + getEndpoint().getConfiguration().getBrokerUrl(), + clientId, + PahoMqtt5Endpoint.createMqttClientPersistence(getEndpoint().getConfiguration())); + } + + private Exception closeOwnedClient(MqttClient ownedClient, Exception primaryException) { + try { + ownedClient.close(true); + } catch (Exception closeException) { + if (primaryException == null) { + return closeException; } - LOG.debug("Disconnecting client: {} from broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl()); - client.disconnect(); + primaryException.addSuppressed(closeException); } - client = null; + return primaryException; } @Override diff --git a/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java new file mode 100644 index 0000000000000..182bc06d72e9e --- /dev/null +++ b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerLifecycleTest.java @@ -0,0 +1,142 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.camel.component.paho.mqtt5; + +import org.apache.camel.CamelContext; +import org.apache.camel.ExtendedCamelContext; +import org.apache.camel.Processor; +import org.apache.camel.spi.ExchangeFactory; +import org.eclipse.paho.mqttv5.client.MqttClient; +import org.eclipse.paho.mqttv5.client.MqttConnectionOptions; +import org.eclipse.paho.mqttv5.common.MqttException; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.catchThrowableOfType; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class PahoMqtt5ConsumerLifecycleTest { + + @Test + void failedStartForceClosesOwnedClient() throws Exception { + MqttClient client = mock(MqttClient.class); + MqttException connectException = new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION); + doThrow(connectException).when(client).connect(any(MqttConnectionOptions.class)); + PahoMqtt5Consumer consumer = createConsumer(new PahoMqtt5Configuration(), client); + + MqttException thrown = catchThrowableOfType(consumer::doStart, MqttException.class); + + assertThat(thrown).isSameAs(connectException); + verify(client).close(true); + } + + @Test + void failedStartSuppressesCloseFailure() throws Exception { + MqttClient client = mock(MqttClient.class); + MqttException connectException = new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION); + MqttException closeException = new MqttException(1); + doThrow(connectException).when(client).connect(any(MqttConnectionOptions.class)); + doThrow(closeException).when(client).close(true); + PahoMqtt5Consumer consumer = createConsumer(new PahoMqtt5Configuration(), client); + + MqttException thrown = catchThrowableOfType(consumer::doStart, MqttException.class); + + assertThat(thrown).isSameAs(connectException); + assertThat(thrown.getSuppressed()).containsExactly(closeException); + } + + @Test + void stopForceClosesOwnedClientWhenDisconnected() throws Exception { + MqttClient client = mock(MqttClient.class); + PahoMqtt5Consumer consumer = createConsumer(new PahoMqtt5Configuration(), client); + consumer.doStart(); + + consumer.doStop(); + + verify(client, never()).disconnect(); + verify(client).close(true); + } + + @Test + void failedStopSuppressesCloseFailure() throws Exception { + MqttClient client = mock(MqttClient.class); + MqttException disconnectException = new MqttException(1); + MqttException closeException = new MqttException(2); + when(client.isConnected()).thenReturn(true); + doThrow(disconnectException).when(client).disconnect(); + doThrow(closeException).when(client).close(true); + PahoMqtt5Configuration configuration = new PahoMqtt5Configuration(); + configuration.setCleanStart(false); + PahoMqtt5Consumer consumer = createConsumer(configuration, client); + consumer.doStart(); + + MqttException thrown = catchThrowableOfType(consumer::doStop, MqttException.class); + + assertThat(thrown).isSameAs(disconnectException); + assertThat(thrown.getSuppressed()).containsExactly(closeException); + } + + @Test + void durableConnectedClientDisconnectsAndClosesWithoutUnsubscribe() throws Exception { + MqttClient client = mock(MqttClient.class); + when(client.isConnected()).thenReturn(true); + PahoMqtt5Configuration configuration = new PahoMqtt5Configuration(); + configuration.setCleanStart(false); + PahoMqtt5Consumer consumer = createConsumer(configuration, client); + consumer.doStart(); + + consumer.doStop(); + + verify(client, never()).unsubscribe("test"); + verify(client).disconnect(); + verify(client).close(true); + } + + @Test + void sharedClientIsNotClosed() throws Exception { + MqttClient client = mock(MqttClient.class); + PahoMqtt5Consumer consumer = createConsumer(new PahoMqtt5Configuration(), mock(MqttClient.class)); + consumer.setClient(client); + + consumer.doStart(); + consumer.doStop(); + + verify(client, never()).close(true); + } + + private static PahoMqtt5Consumer createConsumer(PahoMqtt5Configuration configuration, MqttClient createdClient) { + CamelContext context = mock(CamelContext.class); + ExtendedCamelContext extension = mock(ExtendedCamelContext.class); + ExchangeFactory exchangeFactory = mock(ExchangeFactory.class); + when(context.getCamelContextExtension()).thenReturn(extension); + when(extension.getExchangeFactory()).thenReturn(exchangeFactory); + when(exchangeFactory.newExchangeFactory(any())).thenReturn(exchangeFactory); + PahoMqtt5Endpoint endpoint = new PahoMqtt5Endpoint( + "paho-mqtt5:test", "test", new PahoMqtt5Component(context), configuration); + return new PahoMqtt5Consumer(endpoint, mock(Processor.class)) { + @Override + MqttClient createClient() { + return createdClient; + } + }; + } +} diff --git a/components/camel-paho/pom.xml b/components/camel-paho/pom.xml index 3c4c7895b7172..d5804f694efab 100644 --- a/components/camel-paho/pom.xml +++ b/components/camel-paho/pom.xml @@ -54,6 +54,11 @@ camel-test-junit6 test + + org.assertj + assertj-core + test + org.apache.camel diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java index 4bde73ec28e85..e2e1658f981c7 100644 --- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java +++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java @@ -57,81 +57,121 @@ public void setClient(MqttClient client) { protected void doStart() throws Exception { super.doStart(); - connectOptions = PahoEndpoint.createMqttConnectOptions(getEndpoint().getConfiguration()); - - if (client == null) { - clientId = getEndpoint().getConfiguration().getClientId(); - if (clientId == null) { - clientId = "camel-" + MqttClient.generateClientId(); - } - stopClient = true; - client = new MqttClient( - getEndpoint().getConfiguration().getBrokerUrl(), - clientId, - PahoEndpoint.createMqttClientPersistence(getEndpoint().getConfiguration())); - LOG.debug("Connecting client: {} to broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl()); - if (getEndpoint().getConfiguration().isManualAcksEnabled()) { - client.setManualAcks(true); - + stopClient = client == null; + try { + connectOptions = PahoEndpoint.createMqttConnectOptions(getEndpoint().getConfiguration()); + + if (stopClient) { + clientId = getEndpoint().getConfiguration().getClientId(); + if (clientId == null) { + clientId = "camel-" + MqttClient.generateClientId(); + } + client = createClient(); + LOG.debug("Connecting client: {} to broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl()); + if (getEndpoint().getConfiguration().isManualAcksEnabled()) { + client.setManualAcks(true); + } + client.connect(connectOptions); } - client.connect(connectOptions); - } - client.setCallback(new MqttCallbackExtended() { + client.setCallback(new MqttCallbackExtended() { - @Override - public void connectComplete(boolean reconnect, String serverURI) { - if (reconnect) { - try { - client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); - } catch (MqttException e) { - LOG.error("MQTT resubscribe failed {}", e.getMessage(), e); + @Override + public void connectComplete(boolean reconnect, String serverURI) { + if (reconnect) { + try { + client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } catch (MqttException e) { + LOG.error("MQTT resubscribe failed {}", e.getMessage(), e); + } } } - } - @Override - public void connectionLost(Throwable cause) { - LOG.debug("MQTT broker connection lost due {}", cause.getMessage(), cause); - } + @Override + public void connectionLost(Throwable cause) { + LOG.debug("MQTT broker connection lost due {}", cause.getMessage(), cause); + } - @Override - public void messageArrived(String topic, MqttMessage message) throws Exception { - LOG.debug("Message arrived on topic: {} -> {}", topic, message); - Exchange exchange = createExchange(message, topic); + @Override + public void messageArrived(String topic, MqttMessage message) throws Exception { + LOG.debug("Message arrived on topic: {} -> {}", topic, message); + Exchange exchange = createExchange(message, topic); - // use default consumer callback - AsyncCallback cb = defaultConsumerCallback(exchange, true); - getAsyncProcessor().process(exchange, cb); - } + // use default consumer callback + AsyncCallback cb = defaultConsumerCallback(exchange, true); + getAsyncProcessor().process(exchange, cb); + } - @Override - public void deliveryComplete(IMqttDeliveryToken token) { - LOG.debug("Delivery complete. Token: {}", token); - } - }); + @Override + public void deliveryComplete(IMqttDeliveryToken token) { + LOG.debug("Delivery complete. Token: {}", token); + } + }); - LOG.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic()); - client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + LOG.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic()); + client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos()); + } catch (Exception startException) { + MqttClient ownedClient = stopClient ? client : null; + if (ownedClient != null) { + client = null; + stopClient = false; + closeOwnedClient(ownedClient, startException); + } + throw startException; + } } @Override protected void doStop() throws Exception { - super.doStop(); - - if (stopClient && client != null && client.isConnected()) { - String topic = getEndpoint().getTopic(); - // only unsubscribe if we are not durable - if (getEndpoint().getConfiguration().isCleanSession()) { - LOG.debug("Unsubscribing client: {} from topic: {}", clientId, topic); - client.unsubscribe(topic); - } else { - LOG.debug("Client: {} is durable so will not unsubscribe from topic: {}", clientId, topic); + MqttClient ownedClient = stopClient ? client : null; + Exception stopException = null; + try { + super.doStop(); + + if (ownedClient != null && ownedClient.isConnected()) { + String topic = getEndpoint().getTopic(); + // only unsubscribe if we are not durable + if (getEndpoint().getConfiguration().isCleanSession()) { + LOG.debug("Unsubscribing client: {} from topic: {}", clientId, topic); + ownedClient.unsubscribe(topic); + } else { + LOG.debug("Client: {} is durable so will not unsubscribe from topic: {}", clientId, topic); + } + LOG.debug("Disconnecting client: {} from broker: {}", clientId, + getEndpoint().getConfiguration().getBrokerUrl()); + ownedClient.disconnect(); + } + } catch (Exception e) { + stopException = e; + } finally { + client = null; + stopClient = false; + if (ownedClient != null) { + stopException = closeOwnedClient(ownedClient, stopException); + } + } + if (stopException != null) { + throw stopException; + } + } + + MqttClient createClient() throws MqttException { + return new MqttClient( + getEndpoint().getConfiguration().getBrokerUrl(), + clientId, + PahoEndpoint.createMqttClientPersistence(getEndpoint().getConfiguration())); + } + + private Exception closeOwnedClient(MqttClient ownedClient, Exception primaryException) { + try { + ownedClient.close(true); + } catch (Exception closeException) { + if (primaryException == null) { + return closeException; } - LOG.debug("Disconnecting client: {} from broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl()); - client.disconnect(); + primaryException.addSuppressed(closeException); } - client = null; + return primaryException; } @Override diff --git a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerLifecycleTest.java b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerLifecycleTest.java new file mode 100644 index 0000000000000..81d8750dd0a23 --- /dev/null +++ b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerLifecycleTest.java @@ -0,0 +1,142 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.camel.component.paho; + +import org.apache.camel.CamelContext; +import org.apache.camel.ExtendedCamelContext; +import org.apache.camel.Processor; +import org.apache.camel.spi.ExchangeFactory; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttConnectOptions; +import org.eclipse.paho.client.mqttv3.MqttException; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.catchThrowableOfType; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class PahoConsumerLifecycleTest { + + @Test + void failedStartForceClosesOwnedClient() throws Exception { + MqttClient client = mock(MqttClient.class); + MqttException connectException = new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION); + doThrow(connectException).when(client).connect(any(MqttConnectOptions.class)); + PahoConsumer consumer = createConsumer(new PahoConfiguration(), client); + + MqttException thrown = catchThrowableOfType(consumer::doStart, MqttException.class); + + assertThat(thrown).isSameAs(connectException); + verify(client).close(true); + } + + @Test + void failedStartSuppressesCloseFailure() throws Exception { + MqttClient client = mock(MqttClient.class); + MqttException connectException = new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION); + MqttException closeException = new MqttException(MqttException.REASON_CODE_CLIENT_DISCONNECTING); + doThrow(connectException).when(client).connect(any(MqttConnectOptions.class)); + doThrow(closeException).when(client).close(true); + PahoConsumer consumer = createConsumer(new PahoConfiguration(), client); + + MqttException thrown = catchThrowableOfType(consumer::doStart, MqttException.class); + + assertThat(thrown).isSameAs(connectException); + assertThat(thrown.getSuppressed()).containsExactly(closeException); + } + + @Test + void stopForceClosesOwnedClientWhenDisconnected() throws Exception { + MqttClient client = mock(MqttClient.class); + PahoConsumer consumer = createConsumer(new PahoConfiguration(), client); + consumer.doStart(); + + consumer.doStop(); + + verify(client, never()).disconnect(); + verify(client).close(true); + } + + @Test + void failedStopSuppressesCloseFailure() throws Exception { + MqttClient client = mock(MqttClient.class); + MqttException disconnectException = new MqttException(MqttException.REASON_CODE_CLIENT_DISCONNECTING); + MqttException closeException = new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION); + when(client.isConnected()).thenReturn(true); + doThrow(disconnectException).when(client).disconnect(); + doThrow(closeException).when(client).close(true); + PahoConfiguration configuration = new PahoConfiguration(); + configuration.setCleanSession(false); + PahoConsumer consumer = createConsumer(configuration, client); + consumer.doStart(); + + MqttException thrown = catchThrowableOfType(consumer::doStop, MqttException.class); + + assertThat(thrown).isSameAs(disconnectException); + assertThat(thrown.getSuppressed()).containsExactly(closeException); + } + + @Test + void durableConnectedClientDisconnectsAndClosesWithoutUnsubscribe() throws Exception { + MqttClient client = mock(MqttClient.class); + when(client.isConnected()).thenReturn(true); + PahoConfiguration configuration = new PahoConfiguration(); + configuration.setCleanSession(false); + PahoConsumer consumer = createConsumer(configuration, client); + consumer.doStart(); + + consumer.doStop(); + + verify(client, never()).unsubscribe("test"); + verify(client).disconnect(); + verify(client).close(true); + } + + @Test + void sharedClientIsNotClosed() throws Exception { + MqttClient client = mock(MqttClient.class); + PahoConsumer consumer = createConsumer(new PahoConfiguration(), mock(MqttClient.class)); + consumer.setClient(client); + + consumer.doStart(); + consumer.doStop(); + + verify(client, never()).close(true); + } + + private static PahoConsumer createConsumer(PahoConfiguration configuration, MqttClient createdClient) { + CamelContext context = mock(CamelContext.class); + ExtendedCamelContext extension = mock(ExtendedCamelContext.class); + ExchangeFactory exchangeFactory = mock(ExchangeFactory.class); + when(context.getCamelContextExtension()).thenReturn(extension); + when(extension.getExchangeFactory()).thenReturn(exchangeFactory); + when(exchangeFactory.newExchangeFactory(any())).thenReturn(exchangeFactory); + PahoEndpoint endpoint = new PahoEndpoint( + "paho:test", "test", new PahoComponent(context), configuration); + return new PahoConsumer(endpoint, mock(Processor.class)) { + @Override + MqttClient createClient() { + return createdClient; + } + }; + } +}