GH-3027: Rotator enhancements

* Implement `RotatingServerAdvice` enhancements

* Update javadocs

* Introduce a new `onRotation()` callback for possible overriding
* Make a default implementation do no throw an exception when 
non-standard `MessageSource` impl is provided.
* Adjust its JavaDocs to describe a behavior

Author: David Turanski

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

@@ -171,7 +171,7 @@ protected boolean isFair() {
 			return this.fair;
 		}
 
-		protected KeyDirectory getCurrent() {
+		public KeyDirectory getCurrent() {
 			return this.current;
 		}
 
@@ -201,18 +201,27 @@ public void afterReceive(boolean messageReceived, MessageSource<?> source) {
 		}
 
 		protected void configureSource(MessageSource<?> source) {
-			Assert.isTrue(source instanceof AbstractInboundFileSynchronizingMessageSource
-							|| source instanceof AbstractRemoteFileStreamingMessageSource,
-					"source must be an AbstractInboundFileSynchronizingMessageSource or a "
-							+ "AbstractRemoteFileStreamingMessageSource");
 			if (!this.iterator.hasNext()) {
 				this.iterator = this.keyDirectories.iterator();
 			}
 			this.current = this.iterator.next();
+			onRotation(source);
+		}
+
+		/**
+		 * Update the state of the {@link MessageSource} after the server is rotated, if necessary.
+		 * The default implementation updates the remote directory for known MessageSource implementations that require it,
+		 * specifically, instances of {@link AbstractRemoteFileStreamingMessageSource}, and
+		 * {@link AbstractInboundFileSynchronizingMessageSource}, and does nothing otherwise.
+		 * Subclasses may override this method to support other MessageSource types.
+		 *
+		 * @param source the MessageSource.
+		 */
+		protected void onRotation(MessageSource<?> source) {
 			if (source instanceof AbstractRemoteFileStreamingMessageSource) {
 				((AbstractRemoteFileStreamingMessageSource<?>) source).setRemoteDirectory(this.current.getDirectory());
 			}
-			else {
+			else if (source instanceof AbstractInboundFileSynchronizingMessageSource) {
 				((AbstractInboundFileSynchronizingMessageSource<?>) source).getSynchronizer()
 						.setRemoteDirectory(this.current.getDirectory());
 			}