Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Revert "GH-3027: RotatingServerAdvice enhancements"

Browse Source 
This reverts commit 03b6316d30.
This commit is contained in:
Artem Bilan
2019-08-15 10:55:22 -04:00
parent 03b6316d30
commit f7032c7127
5 changed files with 150 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/AbstractStandardRotationPolicy.java
View File

@@ -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<KeyDirectory> keyDirectories = new ArrayList<>();
private final boolean fair;
private volatile Iterator<KeyDirectory> iterator;
private volatile KeyDirectory current;
private volatile boolean initialized;
protected AbstractStandardRotationPolicy(DelegatingSessionFactory<?> factory, List<KeyDirectory> 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<KeyDirectory> getKeyDirectories() {
return this.keyDirectories;
}
protected boolean isFair() {
return this.fair;
}
protected Iterator<KeyDirectory> 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);
}

51
spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/KeyDirectory.java
View File

@@ -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 + "]";
}
}

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

@@ -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<KeyDirectory> keyDirectories = new ArrayList<>();
private final boolean fair;
private volatile Iterator<KeyDirectory> iterator;
private volatile KeyDirectory current;
private volatile boolean initialized;
public StandardRotationPolicy(DelegatingSessionFactory<?> factory, List<KeyDirectory> 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<KeyDirectory> getIterator() {
return this.iterator;
}
protected void setIterator(Iterator<KeyDirectory> 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<KeyDirectory> 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 + "]";
}
}
}

54
spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotationPolicy.java
View File

@@ -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();
}

2
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java
View File

@@ -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;