GH-3732 adjust PR after code review

Author: mipo256

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

@@ -141,7 +141,7 @@ protected MqttMessageConverter getConverter() {
 	}
 
 	@ManagedAttribute
-	public String[] getTopicNames() {
+	public String[] getTopic() {
 		this.topicLock.lock();
 		try {
 			String[] topicNames = new String[this.topics.size()];

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

@@ -193,7 +193,7 @@ protected synchronized void doStop() {
 						|| (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN)
 						&& this.cleanSession)) {
 
-					this.client.unsubscribe(getTopicNames());
+					this.client.unsubscribe(getTopic());
 				}
 			}
 			catch (MqttException ex) {
@@ -270,7 +270,7 @@ private synchronized void connectAndSubscribe() throws MqttException { // NOSONA
 		}
 
 		this.topicLock.lock();
-		String[] topics = getTopicNames();
+		String[] topics = getTopic();
 		ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
 		try {
 			this.client.connect(connectionOptions);

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -65,6 +65,7 @@
  * See {@link #setPayloadType} for more information about type conversion.
  *
  * @author Artem Bilan
+ * @author Mikhail Polivakha
  *
  * @since 5.5.5
  *
@@ -154,7 +155,8 @@ private void initializeMqttAsyncClient() {
 			this.mqttClient = new MqttAsyncClient(getUrl(), getClientId(), this.persistence);
 			this.mqttClient.setCallback(this);
 			this.mqttClient.setManualAcks(isManualAcks());
-		} catch (MqttException ex) {
+		}
+		catch (MqttException ex) {
 			throw new BeanCreationException("Cannot create 'MqttAsyncClient' for: " + getComponentName(), ex);
 		}
 	}
@@ -163,7 +165,7 @@ private void initializeMqttAsyncClient() {
 	protected void doStart() {
 		ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
 		this.topicLock.lock();
-		String[] topics = getTopicNames();
+		String[] topics = getTopic();
 		try {
 			this.mqttClient.connect(this.connectionOptions).waitForCompletion(getCompletionTimeout());
 			if (topics.length > 0) {
@@ -190,14 +192,19 @@ protected void doStart() {
 	@Override
 	protected void doStop() {
 		this.topicLock.lock();
-		initializeMqttAsyncClientIfRequired();
-		String[] topics = getTopicNames();
+		String[] topics = getTopic();
 		try {
-			this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
-			this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());
-		} catch (MqttException ex) {
+			if (mqttClient != null && mqttClient.isConnected()) {
+				this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
+				this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn("An attempt to stop mqtt client without having one initialized or connected");
+			}
+		}
+		catch (MqttException ex) {
 			logger.error(ex, () -> "Error unsubscribing from " + Arrays.toString(topics));
-		} finally {
+		}
+		finally {
 			this.topicLock.unlock();
 		}
 	}
@@ -223,10 +230,13 @@ public void destroy() {
 	@Override
 	public void addTopic(String topic, int qos) {
 		this.topicLock.lock();
-		initializeMqttAsyncClientIfRequired();
 		try {
-			this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
-			super.addTopic(topic, qos);
+			if (mqttClient != null && mqttClient.isConnected()) {
+				this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
+				super.addTopic(topic, qos);
+			} else {
+				logger.warn(String.format("An attempt to add topic : '%s' with QoS : '%s' without having mqtt client initialized or connected, the topic will not be added!", topic, qos));
+			}
 		}
 		catch (MqttException ex) {
 			throw new MessagingException("Failed to subscribe to topic " + topic, ex);
@@ -239,9 +249,12 @@ public void addTopic(String topic, int qos) {
 	@Override
 	public void removeTopic(String... topic) {
 		this.topicLock.lock();
-		initializeMqttAsyncClient();
 		try {
-			this.mqttClient.unsubscribe(topic).waitForCompletion(getCompletionTimeout());
+			if (mqttClient != null && mqttClient.isConnected()) {
+				this.mqttClient.unsubscribe(topic).waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn(String.format("An attempt to remove topics : '%s' without having mqtt client initialized or connected", Arrays.toString(topic)));
+			}
 			super.removeTopic(topic);
 		}
 		catch (MqttException ex) {