Some various polishing

* Add `RotatingServerAdvice.StandardRotationPolicy.getCurrent()` method
for better end-user experience when this class is extended
* Add NPE protection into the MQTT Channel Adapters.
Some code style polishing for them

**Cherry-pick to 5.0.x**

spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java

@@ -36,6 +36,7 @@ import org.springframework.util.Assert;
  *
  * @author Gary Russell
  * @author Michael Forstner
+ * @author Artem Bilan
  *
  * @since 5.0.7
  *
@@ -118,7 +119,7 @@ public class RotatingServerAdvice extends AbstractMessageSourceAdvice {
 
 		protected final Log logger = LogFactory.getLog(getClass());
 
-		private final DelegatingSessionFactory<?> factory;
+		protected final DelegatingSessionFactory<?> factory;
 
 		private final List<KeyDirectory> keyDirectories = new ArrayList<>();
 
@@ -170,6 +171,10 @@ public class RotatingServerAdvice extends AbstractMessageSourceAdvice {
 			return this.fair;
 		}
 
+		protected KeyDirectory getCurrent() {
+			return this.current;
+		}
+
 		@Override
 		public void beforeReceive(MessageSource<?> source) {
 			if (this.fair || !this.initialized) {

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

@@ -162,6 +162,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 				if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
 						|| (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN)
 								&& this.cleanSession)) {
+
 					this.client.unsubscribe(getTopic());
 				}
 			}
@@ -249,8 +250,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 				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));
+								+ " requested: " + Arrays.toString(requestedQos)
+								+ " granted: " + Arrays.toString(grantedQos));
 					}
 					break;
 				}
@@ -294,7 +295,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		}
 	}
 
-	private void scheduleReconnect() {
+	private synchronized void scheduleReconnect() {
+		cancelReconnect();
 		try {
 			this.reconnectFuture = getTaskScheduler().schedule(() -> {
 				try {
@@ -324,12 +326,14 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		if (isRunning()) {
 			this.logger.error("Lost connection: " + cause.getMessage() + "; retrying...");
 			this.connected = false;
-			try {
-				this.client.setCallback(null);
-				this.client.close();
-			}
-			catch (MqttException e) {
-				// NOSONAR
+			if (this.client != null) {
+				try {
+					this.client.setCallback(null);
+					this.client.close();
+				}
+				catch (MqttException e) {
+					// NOSONAR
+				}
 			}
 			this.client = null;
 			scheduleReconnect();
@@ -340,7 +344,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 	}
 
 	@Override
-	public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
+	public void messageArrived(String topic, MqttMessage mqttMessage) {
 		Message<?> message = this.getConverter().toMessage(topic, mqttMessage);
 		try {
 			sendMessage(message);

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java

@@ -39,6 +39,7 @@ import org.springframework.util.Assert;
  *
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 4.0
  *
  */
@@ -51,13 +52,13 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 
 	private final MqttPahoClientFactory clientFactory;
 
-	private volatile IMqttAsyncClient client;
-
 	private volatile boolean async;
 
 	private volatile boolean asyncEvents;
 
-	private volatile ApplicationEventPublisher applicationEventPublisher;
+	private ApplicationEventPublisher applicationEventPublisher;
+
+	private volatile IMqttAsyncClient client;
 
 	/**
 	 * Use this constructor for a single url (although it may be overridden
@@ -181,7 +182,6 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 			catch (MqttException e) {
 				if (client != null) {
 					client.close();
-					client = null;
 				}
 				throw new MessagingException("Failed to connect", e);
 			}
@@ -215,18 +215,20 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 	@Override
 	public synchronized void connectionLost(Throwable cause) {
 		logger.error("Lost connection; will attempt reconnect on next request");
-		try {
-			this.client.setCallback(null);
-			this.client.close();
+		if (this.client != null) {
+			try {
+				this.client.setCallback(null);
+				this.client.close();
+			}
+			catch (MqttException e) {
+				// NOSONAR
+			}
+			this.client = null;
 		}
-		catch (MqttException e) {
-			// NOSONAR
-		}
-		this.client = null;
 	}
 
 	@Override
-	public void messageArrived(String topic, MqttMessage message) throws Exception {
+	public void messageArrived(String topic, MqttMessage message) {
 
 	}