diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/AbstractStandardRotationPolicy.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/AbstractStandardRotationPolicy.java deleted file mode 100644 index 62f6e47b4d..0000000000 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/AbstractStandardRotationPolicy.java +++ /dev/null @@ -1,135 +0,0 @@ -/* - * Copyright 2019 the original author or authors. - * - * Licensed 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 - * - * https://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.springframework.integration.file.remote.aop; - -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.integration.core.MessageSource; -import org.springframework.integration.file.remote.session.DelegatingSessionFactory; -import org.springframework.util.Assert; - -/** - * Standard rotation policy; iterates over key/directory pairs; when the end is reached, - * starts again at the beginning. If the fair option is true the rotation occurs on every - * poll, regardless of result. Otherwise rotation occurs when the current pair returns no - * message. - * - * Subclasses implement {@code onRotation(MessageSource source)} to configure the - * {@link MessageSource} on each rotation. - * - * @author Gary Russell - * @author Michael Forstner - * @author Artem Bilan - * @author David Turanski - * - * @since 5.1.8 - */ -public abstract class AbstractStandardRotationPolicy implements RotationPolicy { - protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR final - - private final DelegatingSessionFactory factory; // NOSONAR final - - private final List keyDirectories = new ArrayList<>(); - - private final boolean fair; - - private volatile Iterator iterator; - - private volatile KeyDirectory current; - - private volatile boolean initialized; - - protected AbstractStandardRotationPolicy(DelegatingSessionFactory factory, List keyDirectories, - boolean fair) { - - Assert.notNull(factory, "factory cannot be null"); - Assert.notNull(keyDirectories, "keyDirectories cannot be null"); - Assert.isTrue(keyDirectories.size() > 0, "At least one KeyDirectory is required"); - this.factory = factory; - this.keyDirectories.addAll(keyDirectories); - this.fair = fair; - this.iterator = this.keyDirectories.iterator(); - } - - @Override - public void beforeReceive(MessageSource source) { - if (this.fair || !this.initialized) { - configureSource(source); - this.initialized = true; - } - if (this.logger.isTraceEnabled()) { - this.logger.trace("Next poll is for " + this.current); - } - this.factory.setThreadKey(this.current.getKey()); - } - - @Override - public void afterReceive(boolean messageReceived, MessageSource source) { - if (this.logger.isTraceEnabled()) { - this.logger.trace("Poll produced " - + (messageReceived ? "a" : "no") - + " message"); - } - this.factory.clearThreadKey(); - if (!this.fair && !messageReceived) { - configureSource(source); - } - } - - @Override - public KeyDirectory getCurrent() { - return this.current; - } - - - protected DelegatingSessionFactory getFactory() { - return this.factory; - } - - protected List getKeyDirectories() { - return this.keyDirectories; - } - - protected boolean isFair() { - return this.fair; - } - - protected Iterator getIterator() { - return this.iterator; - } - - protected boolean isInitialized() { - return this.initialized; - } - - protected void configureSource(MessageSource source) { - - if (!this.iterator.hasNext()) { - this.iterator = this.keyDirectories.iterator(); - } - this.current = this.iterator.next(); - - onRotation(source); - } - - protected abstract void onRotation(MessageSource source); -} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/KeyDirectory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/KeyDirectory.java deleted file mode 100644 index 1ff7985484..0000000000 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/KeyDirectory.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * Copyright 2019 the original author or authors. - * - * Licensed 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 - * - * https://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.springframework.integration.file.remote.aop; - -import org.springframework.integration.file.remote.session.DelegatingSessionFactory; -import org.springframework.util.Assert; - -/** - * A {@link DelegatingSessionFactory} key/directory pair. - */ -public class KeyDirectory { - - private final Object key; - - private final String directory; - - public KeyDirectory(Object key, String directory) { - Assert.notNull(key, "key cannot be null"); - Assert.notNull(directory, "directory cannot be null"); - this.key = key; - this.directory = directory; - } - - public Object getKey() { - return this.key; - } - - public String getDirectory() { - return this.directory; - } - - @Override - public String toString() { - return "KeyDirectory [key=" + this.key.toString() + ", directory=" + this.directory + "]"; - } - -} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java index 4c520995e1..d08bf28de3 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java @@ -16,8 +16,13 @@ package org.springframework.integration.file.remote.aop; +import java.util.ArrayList; +import java.util.Iterator; import java.util.List; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.integration.aop.AbstractMessageSourceAdvice; import org.springframework.integration.core.MessageSource; import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource; @@ -32,7 +37,6 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Michael Forstner * @author Artem Bilan - * @author David Turanski * * @since 5.0.7 * @@ -84,29 +88,167 @@ public class RotatingServerAdvice extends AbstractMessageSourceAdvice { return result; } - public static class StandardRotationPolicy extends AbstractStandardRotationPolicy { + /** + * Implementations can reconfigure the message source before and/or after + * a poll. + */ + public interface RotationPolicy { + /** + * Invoked before the message source receive() method. + * @param source the message source. + */ + void beforeReceive(MessageSource source); + + /** + * Invoked after the message source receive() method. + * @param messageReceived true if a message was received. + * @param source the message source. + */ + void afterReceive(boolean messageReceived, MessageSource source); + + } + + /** + * Standard rotation policy; iterates over key/directory pairs; when the end + * is reached, starts again at the beginning. If the fair option is true + * the rotation occurs on every poll, regardless of result. Otherwise rotation + * occurs when the current pair returns no message. + */ + public static class StandardRotationPolicy implements RotationPolicy { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected final DelegatingSessionFactory factory; + + private final List keyDirectories = new ArrayList<>(); + + private final boolean fair; + + private volatile Iterator iterator; + + private volatile KeyDirectory current; + + private volatile boolean initialized; public StandardRotationPolicy(DelegatingSessionFactory factory, List keyDirectories, boolean fair) { - super(factory, keyDirectories, fair); + + Assert.notNull(factory, "factory cannot be null"); + Assert.notNull(keyDirectories, "keyDirectories cannot be null"); + Assert.isTrue(keyDirectories.size() > 0, "At least one KeyDirectory is required"); + this.factory = factory; + this.keyDirectories.addAll(keyDirectories); + this.fair = fair; + this.iterator = this.keyDirectories.iterator(); + } + + protected Iterator getIterator() { + return this.iterator; + } + + protected void setIterator(Iterator iterator) { + this.iterator = iterator; + } + + protected boolean isInitialized() { + return this.initialized; + } + + protected void setInitialized(boolean initialized) { + this.initialized = initialized; + } + + protected DelegatingSessionFactory getFactory() { + return this.factory; + } + + protected List getKeyDirectories() { + return this.keyDirectories; + } + + protected boolean isFair() { + return this.fair; + } + + protected KeyDirectory getCurrent() { + return this.current; } @Override - protected void onRotation(MessageSource source) { + public void beforeReceive(MessageSource source) { + if (this.fair || !this.initialized) { + configureSource(source); + this.initialized = true; + } + if (this.logger.isTraceEnabled()) { + this.logger.trace("Next poll is for " + this.current); + } + this.factory.setThreadKey(this.current.getKey()); + } + + @Override + public void afterReceive(boolean messageReceived, MessageSource source) { + if (this.logger.isTraceEnabled()) { + this.logger.trace("Poll produced " + + (messageReceived ? "a" : "no") + + " message"); + } + this.factory.clearThreadKey(); + if (!this.fair && !messageReceived) { + configureSource(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(); if (source instanceof AbstractRemoteFileStreamingMessageSource) { - ((AbstractRemoteFileStreamingMessageSource) source).setRemoteDirectory(getCurrent().getDirectory()); + ((AbstractRemoteFileStreamingMessageSource) source).setRemoteDirectory(this.current.getDirectory()); } else { ((AbstractInboundFileSynchronizingMessageSource) source).getSynchronizer() - .setRemoteDirectory(getCurrent().getDirectory()); + .setRemoteDirectory(this.current.getDirectory()); } } } + + /** + * A {@link DelegatingSessionFactory} key/directory pair. + */ + public static class KeyDirectory { + + private final Object key; + + private final String directory; + + public KeyDirectory(Object key, String directory) { + Assert.notNull(key, "key cannot be null"); + Assert.notNull(directory, "directory cannot be null"); + this.key = key; + this.directory = directory; + } + + public Object getKey() { + return this.key; + } + + public String getDirectory() { + return this.directory; + } + + @Override + public String toString() { + return "KeyDirectory [key=" + this.key.toString() + ", directory=" + this.directory + "]"; + } + + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotationPolicy.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotationPolicy.java deleted file mode 100644 index ab21233ce6..0000000000 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotationPolicy.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Copyright 2019 the original author or authors. - * - * Licensed 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 - * - * https://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.springframework.integration.file.remote.aop; - -import org.springframework.integration.core.MessageSource; - -/** - * Implementations can reconfigure the message source before and/or after - * a poll. - * - * @author Gary Russell - * @author Michael Forstner - * @author Artem Bilan - * @author David Turanski - * - * @since 5.0.7 - */ -public interface RotationPolicy { - - /** - * Invoked before the message source receive() method. - * @param source the message source. - */ - void beforeReceive(MessageSource source); - - /** - * Invoked after the message source receive() method. - * @param messageReceived true if a message was received. - * @param source the message source. - */ - void afterReceive(boolean messageReceived, MessageSource source); - - - /** - * - * @return the current {@link KeyDirectory} - */ - KeyDirectory getCurrent(); - -} diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java index e75ab4a28f..b3a94805f6 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java @@ -43,8 +43,8 @@ import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.MessageChannels; import org.springframework.integration.dsl.Pollers; import org.springframework.integration.dsl.StandardIntegrationFlow; -import org.springframework.integration.file.remote.aop.KeyDirectory; import org.springframework.integration.file.remote.aop.RotatingServerAdvice; +import org.springframework.integration.file.remote.aop.RotatingServerAdvice.KeyDirectory; import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.DefaultSessionFactoryLocator; import org.springframework.integration.file.remote.session.DelegatingSessionFactory;