diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java new file mode 100644 index 0000000000..d9f1ad8623 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java @@ -0,0 +1,56 @@ +/* + * Copyright 2016 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 + * + * http://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.endpoint; + +/** + * A message source that can limit the number of remote objects it fetches. + * + * @author Gary Russell + * @since 5.0 + * + */ +public abstract class AbstractFetchLimitingMessageSource extends AbstractMessageSource + implements MessageSourceManagement { + + private volatile int maxFetchSize = Integer.MIN_VALUE; + + @Override + public void setMaxFetchSize(int maxFetchSize) { + this.maxFetchSize = maxFetchSize; + } + + @Override + public int getMaxFetchSize() { + return this.maxFetchSize; + } + + @Override + protected Object doReceive() { + return doReceive(this.maxFetchSize); + } + + /** + * Subclasses must implement this method. Typically the returned value will be the + * payload of type T, but the returned value may also be a Message instance whose + * payload is of type T. + * @param maxFetchSize the maximum number of messages to fetch if a fetch is + * necessary. + * @return The value returned. + */ + protected abstract Object doReceive(int maxFetchSize); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java index d6d63c536d..699b98564b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java @@ -128,10 +128,13 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat } @Override - @SuppressWarnings("unchecked") public final Message receive() { + return buildMessage(doReceive()); + } + + @SuppressWarnings("unchecked") + protected Message buildMessage(Object result) { Message message = null; - Object result = this.doReceive(); Map headers = this.evaluateHeaders(); if (result instanceof Message) { try { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourceManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourceManagement.java new file mode 100644 index 0000000000..0884a9ec07 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourceManagement.java @@ -0,0 +1,51 @@ +/* + * Copyright 2016 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 + * + * http://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.endpoint; + +import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.support.management.MessageSourceMetrics; +import org.springframework.jmx.export.annotation.ManagedAttribute; + +/** + * Message sources implementing this interface have additional properties that + * can be set or examined using JMX. + * + * @author Gary Russell + * @since 5.0 + * + */ +@IntegrationManagedResource +public interface MessageSourceManagement extends MessageSourceMetrics { + + /** + * Set the maximum number of objects the source should fetch if it is necessary to + * fetch objects. Setting the + * maxFetchSize to 0 disables remote fetching, a negative value indicates no limit. + * @param maxFetchSize the max fetch size; a negative value means unlimited. + */ + @ManagedAttribute(description = "Maximum objects to fetch") + void setMaxFetchSize(int maxFetchSize); + + /** + * Return the max fetch size. + * @return the max fetch size. + * @see #setMaxFetchSize(int) + */ + @ManagedAttribute + int getMaxFetchSize(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceManagement.java new file mode 100644 index 0000000000..284d85bd6f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceManagement.java @@ -0,0 +1,45 @@ +/* + * Copyright 2016 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 + * + * http://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.support.management; + +import org.springframework.context.Lifecycle; +import org.springframework.integration.endpoint.MessageSourceManagement; + +/** + * An extension to {@link LifecycleMessageSourceMetrics} for sources that implement {@link MessageSourceManagement}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class LifecycleMessageSourceManagement extends LifecycleMessageSourceMetrics implements MessageSourceManagement { + + public LifecycleMessageSourceManagement(Lifecycle lifecycle, MessageSourceManagement delegate) { + super(lifecycle, delegate); + } + + @Override + public void setMaxFetchSize(int maxFetchSize) { + ((MessageSourceManagement) this.delegate).setMaxFetchSize(maxFetchSize); + } + + @Override + public int getMaxFetchSize() { + return ((MessageSourceManagement) this.delegate).getMaxFetchSize(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceMetrics.java index df4f97d2f7..bab8141c8e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceMetrics.java @@ -33,7 +33,7 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life private final Lifecycle lifecycle; - private final MessageSourceMetrics delegate; + protected final MessageSourceMetrics delegate; public LifecycleMessageSourceMetrics(Lifecycle lifecycle, MessageSourceMetrics delegate) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceManagement.java new file mode 100644 index 0000000000..e4544a48b5 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceManagement.java @@ -0,0 +1,48 @@ +/* + * Copyright 2016 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 + * + * http://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.support.management; + +import org.springframework.context.Lifecycle; +import org.springframework.integration.endpoint.MessageSourceManagement; + +/** + * An extension to {@link LifecycleTrackableMessageSourceMetrics} for sources + * that implement {@link MessageSourceManagement}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class LifecycleTrackableMessageSourceManagement extends LifecycleTrackableMessageSourceMetrics + implements MessageSourceManagement { + + public LifecycleTrackableMessageSourceManagement(Lifecycle lifecycle, MessageSourceManagement delegate) { + super(lifecycle, delegate); + } + + @Override + public void setMaxFetchSize(int maxFetchSize) { + ((MessageSourceManagement) this.delegate).setMaxFetchSize(maxFetchSize); + } + + @Override + public int getMaxFetchSize() { + return ((MessageSourceManagement) this.delegate).getMaxFetchSize(); + } + + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd index 4e3e897b85..ea5489a092 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd @@ -5009,6 +5009,18 @@ See 'SmartLifecycleRoleController'. + + + + +When fetching objects from some external resource, limit the number of such objects that +will be retrieved on each fetch. A negative value (default) indicates no limit; a value +of zero effectively disables fetching remote objects. + + + + + diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java index b822ee7dfa..6f8e7b99e1 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java @@ -82,6 +82,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst synchronizerBuilder.addPropertyValue("localFilenameGeneratorExpression", localFileGeneratorExpressionBuilder.getBeanDefinition()); } + IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "max-fetch-size"); return messageSourceBuilder.getBeanDefinition(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index dc5353f9d3..df300bd44f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java @@ -22,6 +22,7 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; import java.io.OutputStream; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -227,6 +228,17 @@ public abstract class AbstractInboundFileSynchronizer @Override public void synchronizeToLocalDirectory(final File localDirectory) { + synchronizeToLocalDirectory(localDirectory, Integer.MIN_VALUE); + } + + @Override + public void synchronizeToLocalDirectory(final File localDirectory, final int maxFetchSize) { + if (maxFetchSize == 0) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Max Fetch Size is zero - fetch to " + localDirectory.getAbsolutePath() + " ignored"); + } + return; + } final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class); try { int transferred = this.remoteFileTemplate.execute(new SessionCallback() { @@ -236,6 +248,14 @@ public abstract class AbstractInboundFileSynchronizer F[] files = session.list(remoteDirectory); if (!ObjectUtils.isEmpty(files)) { List filteredFiles = filterFiles(files); + if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) { + rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); + List newList = new ArrayList<>(maxFetchSize); + for (int i = 0; i < maxFetchSize; i++) { + newList.add(filteredFiles.get(i)); + } + filteredFiles = newList; + } for (F file : filteredFiles) { try { if (file != null) { @@ -245,17 +265,11 @@ public abstract class AbstractInboundFileSynchronizer } } catch (RuntimeException e) { - if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) { - ((ReversibleFileListFilter) AbstractInboundFileSynchronizer.this.filter) - .rollback(file, filteredFiles); - } + rollbackFromFileToListEnd(filteredFiles, file); throw e; } catch (IOException e) { - if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) { - ((ReversibleFileListFilter) AbstractInboundFileSynchronizer.this.filter) - .rollback(file, filteredFiles); - } + rollbackFromFileToListEnd(filteredFiles, file); throw e; } } @@ -265,6 +279,14 @@ public abstract class AbstractInboundFileSynchronizer return 0; } } + + public void rollbackFromFileToListEnd(List filteredFiles, F file) { + if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) { + ((ReversibleFileListFilter) AbstractInboundFileSynchronizer.this.filter) + .rollback(file, filteredFiles); + } + } + }); if (this.logger.isDebugEnabled()) { this.logger.debug(transferred + " files transferred"); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java index 7133da1136..764d892cc2 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java @@ -25,7 +25,7 @@ import java.util.regex.Pattern; import org.springframework.beans.factory.BeanInitializationException; import org.springframework.context.Lifecycle; -import org.springframework.integration.endpoint.AbstractMessageSource; +import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.CompositeFileListFilter; @@ -57,7 +57,7 @@ import org.springframework.util.Assert; * @author Gary Russell */ public abstract class AbstractInboundFileSynchronizingMessageSource - extends AbstractMessageSource implements Lifecycle { + extends AbstractFetchLimitingMessageSource implements Lifecycle { private volatile boolean running; @@ -182,21 +182,20 @@ public abstract class AbstractInboundFileSynchronizingMessageSource /** * Polls from the file source. If the result is not null, it will be returned. * If the result is null, it attempts to sync up with the remote directory to populate the file source. + * At most, maxFetchSize files will be fetched. * Then, it polls the file source again and returns the result, whether or not it is null. + * @param maxFetchSize the maximum files to fetch. */ @Override - public final Message doReceive() { - Assert.state(this.fileSource != null, "fileSource must not be null"); - Assert.state(this.synchronizer != null, "synchronizer must not be null"); + public final Message doReceive(int maxFetchSize) { Message message = this.fileSource.receive(); if (message == null) { - this.synchronizer.synchronizeToLocalDirectory(this.localDirectory); + this.synchronizer.synchronizeToLocalDirectory(this.localDirectory, maxFetchSize); message = this.fileSource.receive(); } return message; } - @SuppressWarnings("unchecked") private FileListFilter buildFilter() { Pattern completePattern = Pattern.compile("^.*(?(Arrays.asList( diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/InboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/InboundFileSynchronizer.java index c147811f12..e5db1fd006 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/InboundFileSynchronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/InboundFileSynchronizer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2016 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. @@ -22,10 +22,25 @@ import java.io.File; * Strategy for synchronizing from a remote File system to a local directory. * * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ +@FunctionalInterface public interface InboundFileSynchronizer { + /** + * Synchronize all available files to the local directory; + * @param localDirectory the directory. + */ void synchronizeToLocalDirectory(File localDirectory); + /** + * Synchronize up to maxFetchSize files to the local directory; + * @param localDirectory the directory. + * @param maxFetchSize the maximum files to fetch. + */ + default void synchronizeToLocalDirectory(File localDirectory, int maxFetchSize) { + synchronizeToLocalDirectory(localDirectory); + } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java index 011301238a..70d61ae2bc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java @@ -26,11 +26,13 @@ import java.io.File; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; @@ -94,6 +96,81 @@ public class AbstractRemoteFileSynchronizerTests { sync.close(); } + @Test + public void testMaxFetchSizeSynchronizer() throws Exception { + final AtomicInteger count = new AtomicInteger(); + AbstractInboundFileSynchronizer sync = createLimitingSynchronizer(count); + + sync.synchronizeToLocalDirectory(mock(File.class), 1); + assertEquals(1, count.get()); + sync.synchronizeToLocalDirectory(mock(File.class), 1); + assertEquals(2, count.get()); + sync.synchronizeToLocalDirectory(mock(File.class), 1); + assertEquals(3, count.get()); + sync.close(); + } + + @Test + public void testMaxFetchSizeSource() throws Exception { + final AtomicInteger count = new AtomicInteger(); + AbstractInboundFileSynchronizer sync = createLimitingSynchronizer(count); + + AbstractInboundFileSynchronizingMessageSource source = + new AbstractInboundFileSynchronizingMessageSource(sync) { + + @Override + public String getComponentType() { + return "foo"; + } + + }; + source.setLocalDirectory(new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID())); + source.setAutoCreateLocalDirectory(true); + source.setBeanFactory(mock(BeanFactory.class)); + source.setMaxFetchSize(1); + source.afterPropertiesSet(); + + source.receive(); + assertEquals(1, count.get()); + sync.synchronizeToLocalDirectory(mock(File.class), 1); + source.receive(); + sync.synchronizeToLocalDirectory(mock(File.class), 1); + source.receive(); + sync.close(); + } + + private AbstractInboundFileSynchronizer createLimitingSynchronizer(final AtomicInteger count) { + SessionFactory sf = new StringSessionFactory(); + AbstractInboundFileSynchronizer sync = new AbstractInboundFileSynchronizer(sf) { + + @Override + protected boolean isFile(String file) { + return true; + } + + @Override + protected String getFilename(String file) { + return file; + } + + @Override + protected long getModified(String file) { + return 0; + } + + @Override + protected void copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, File localDirectory, + Session session) throws IOException { + count.incrementAndGet(); + } + + }; + sync.setFilter(new AcceptOnceFileListFilter()); + sync.setRemoteDirectory("foo"); + sync.setBeanFactory(mock(BeanFactory.class)); + return sync; + } + private class StringSessionFactory implements SessionFactory { @Override diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd index bfb66f83a9..0cb7dd0e04 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd @@ -164,6 +164,7 @@ + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml index 46dce33a96..6d9abd6cc4 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml @@ -27,6 +27,7 @@ local-filename-generator-expression="#this.toUpperCase() + '.a' + @fooString" comparator="comparator" temporary-file-suffix=".foo" + max-fetch-size="42" local-filter="acceptAllFilter" remote-directory-expression="'foo/bar'"> diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index 62a06f6e9b..834212f564 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java @@ -122,6 +122,7 @@ public class FtpInboundChannelAdapterParserTests { } }); assertEquals("FOO.afoo", genMethod.get().invoke(fisync, "foo")); + assertEquals(42, inbound.getMaxFetchSize()); } @Test @@ -135,6 +136,7 @@ public class FtpInboundChannelAdapterParserTests { assertEquals("/", remoteFileSeparator); assertEquals("foo/bar", TestUtils.getPropertyValue(fisync, "remoteDirectoryExpression", Expression.class) .getExpressionString()); + assertEquals(Integer.MIN_VALUE, TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.maxFetchSize")); } @Test diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java index 468839c22c..65b729760b 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java @@ -35,6 +35,14 @@ import java.util.Calendar; import java.util.Collection; import java.util.List; +import org.apache.commons.net.ftp.FTPClient; +import org.apache.commons.net.ftp.FTPFile; +import org.hamcrest.Matchers; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.Expression; import org.springframework.expression.ExpressionParser; @@ -51,14 +59,6 @@ import org.springframework.integration.metadata.PropertiesPersistingMetadataStor import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; -import org.apache.commons.net.ftp.FTPClient; -import org.apache.commons.net.ftp.FTPFile; -import org.hamcrest.Matchers; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; -import org.mockito.Mockito; - /** * @author Oleg Zhurakousky * @author Gunnar Hillert @@ -142,7 +142,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests { assertNull(nothing); // two times because on the third receive (above) the internal queue will be empty, so it will attempt - verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectoy); + verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectoy, Integer.MIN_VALUE); assertTrue(new File("test/A.TEST.a").exists()); assertTrue(new File("test/B.TEST.a").exists()); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index d903d192e8..b848388df0 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -48,14 +48,17 @@ import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.MessageSourceManagement; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.history.MessageHistoryConfigurer; import org.springframework.integration.support.context.NamedComponent; import org.springframework.integration.support.management.IntegrationManagementConfigurer; import org.springframework.integration.support.management.LifecycleMessageHandlerMetrics; +import org.springframework.integration.support.management.LifecycleMessageSourceManagement; import org.springframework.integration.support.management.LifecycleMessageSourceMetrics; import org.springframework.integration.support.management.LifecycleTrackableMessageHandlerMetrics; +import org.springframework.integration.support.management.LifecycleTrackableMessageSourceManagement; import org.springframework.integration.support.management.LifecycleTrackableMessageSourceMetrics; import org.springframework.integration.support.management.MappingMessageRouterManagement; import org.springframework.integration.support.management.MessageChannelMetrics; @@ -993,6 +996,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements Applicati break; } } + if (endpointName == null) { + endpoint = null; + } if (name != null && endpoint != null && name.startsWith("_org.springframework.integration")) { name = getInternalComponentName(name); source = "internal"; @@ -1040,16 +1046,28 @@ public class IntegrationMBeanExporter extends MBeanExporter implements Applicati if (endpoint instanceof Lifecycle) { // Wrap the monitor in a lifecycle so it exposes the start/stop operations if (endpoint instanceof TrackableComponent) { - result = new LifecycleTrackableMessageSourceMetrics((Lifecycle) endpoint, monitor); + if (monitor instanceof MessageSourceManagement) { + result = new LifecycleTrackableMessageSourceManagement((Lifecycle) endpoint, + (MessageSourceManagement) monitor); + } + else { + result = new LifecycleTrackableMessageSourceMetrics((Lifecycle) endpoint, monitor); + } } else { - result = new LifecycleMessageSourceMetrics((Lifecycle) endpoint, monitor); + if (monitor instanceof MessageSourceManagement) { + result = new LifecycleMessageSourceManagement((Lifecycle) endpoint, + (MessageSourceManagement) monitor); + } + else { + result = new LifecycleMessageSourceMetrics((Lifecycle) endpoint, monitor); + } } } if (name == null) { name = monitor.toString(); - source = "handler"; + source = "source"; } if (endpointName != null) { diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceTests.java new file mode 100644 index 0000000000..3c3acda4fe --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceTests.java @@ -0,0 +1,128 @@ +/* + * Copyright 2016 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 + * + * http://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.monitor; + +import static org.hamcrest.Matchers.equalTo; +import static org.junit.Assert.assertThat; + +import java.util.Set; + +import javax.management.Attribute; +import javax.management.MBeanServer; +import javax.management.ObjectName; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource; +import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.jmx.support.MBeanServerFactoryBean; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 5.0 + * + */ +@RunWith(SpringRunner.class) +public class MessageSourceTests { + + @Autowired + private MBeanServer server; + + @Autowired + private MaxFetchSource source1; + + @Autowired + private MaxFetchSource source2; + + @Test + public void testMaxFetch() throws Exception { + Set query = this.server.queryNames(new ObjectName("foo:type=MessageSource,*"), null); + for (ObjectName instance : query) { + assertThat(this.server.getAttribute(instance, "MaxFetchSize"), equalTo(123)); + this.server.setAttribute(instance, new Attribute("MaxFetchSize", 456)); + assertThat(this.server.getAttribute(instance, "MaxFetchSize"), equalTo(456)); + } + assertThat(this.source1.getMaxFetchSize(), equalTo(456)); + assertThat(this.source2.getMaxFetchSize(), equalTo(456)); + } + + @Configuration + @EnableIntegrationMBeanExport(defaultDomain = "foo") + @EnableIntegration + public static class Config { + + @Bean + public MBeanServerFactoryBean fb() { + MBeanServerFactoryBean fb = new MBeanServerFactoryBean(); + fb.setLocateExistingServerIfPossible(true); + return fb; + } + + @Bean + @InboundChannelAdapter(channel = "out") + public MaxFetchSource source1() { + return new MaxFetchSource(); + } + + @Bean + public MaxFetchSource source2() { // raw source, no consumer + return new MaxFetchSource(); + } + + @Bean + public PollableChannel out() { + return new QueueChannel(); + } + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata defaultPoller() { + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(t -> null); + return pollerMetadata; + } + + } + + public static class MaxFetchSource extends AbstractFetchLimitingMessageSource { + + { + setMaxFetchSize(123); + } + + @Override + public String getComponentType() { + return null; + } + + @Override + protected Object doReceive(int maxFetchSize) { + return null; + } + + } + +} diff --git a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd index 2b7c4cb279..adbdc15651 100644 --- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd @@ -168,6 +168,7 @@ + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml index b8f86acf25..31bfa0cc61 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml @@ -42,6 +42,7 @@ temporary-file-suffix=".bar" comparator="comparator" local-filter="acceptAllFilter" + max-fetch-size="42" delete-remote-files="${delete.remote.files}" preserve-timestamp="true"> diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java index ab1ca39a13..7dbb1d1940 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java @@ -102,6 +102,7 @@ public class InboundChannelAdapterParserTests { Collection> filters = TestUtils.getPropertyValue(source, "fileSource.scanner.filter.fileFilters", Collection.class); assertThat(filters, hasItem(acceptAllFilter)); + assertEquals(42, source.getMaxFetchSize()); context.close(); } @@ -117,6 +118,7 @@ public class InboundChannelAdapterParserTests { .getPropertyValue(autoChannelAdapter, "source.synchronizer.remoteDirectoryExpression", Expression.class) .getExpressionString()); assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel")); + assertEquals(Integer.MIN_VALUE, TestUtils.getPropertyValue(autoChannelAdapter, "source.maxFetchSize")); context.close(); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java index 212645ca7a..a475c6ce73 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java @@ -137,7 +137,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { assertNull(nothing); // two times because on the third receive (above) the internal queue will be empty, so it will attempt - verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectory); + verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectory, Integer.MIN_VALUE); assertTrue(new File("test/a.test").exists()); assertTrue(new File("test/b.test").exists()); diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 9f54ba56da..2f67ffe89a 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -172,6 +172,7 @@ The _FTP Inbound Channel Adapter_ is a special listener that will connect to the local-filename-generator-expression="#this.toUpperCase() + '.a'" local-filter="myFilter" temporary-file-suffix=".writing" + max-fetch-size="-1" local-directory="."> @@ -184,6 +185,11 @@ If you want to override this behavior you can set the `local-filename-generator- Unlike outbound gateways and adapters where the root object of the SpEL Evaluation Context is a `Message`, this inbound adapter does not yet have the Message at the time of evaluation since that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). +The inbound channel adapter first retrieves the file to a local directory and then emits each file according to the poller configuration. +Starting with _version 5.0_ you can now limit the number of files fetched from the FTP server when new file retrievals are needed. +This can be beneficial when the target files are very large and/or when running in a clustered system with a persistent file list filter discussed below. +Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved. + Starting with _Spring Integration 3.0_, you can specify the `preserve-timestamp` attribute (default `false`); when `true`, the local file's modified timestamp will be set to the value retrieved from the server; otherwise it will be set to the current time. Starting with _version 4.2_, you can specify `remote-directory-expression` instead of `remote-directory`, allowing @@ -209,7 +215,7 @@ When used with a shared data store (such as `Redis` with the `RedisMetadataStore The above discussion refers to filtering the files before retrieving them. Once the files have been retrieved, an additional filter is applied to the files on the file system. -By default, this is an`AcceptOnceFileListFilter` which, as discussed, retains state in memory and does not consider the file's modified time. +By default, this is an `AcceptOnceFileListFilter` which, as discussed, retains state in memory and does not consider the file's modified time. Unless your application removes files after processing, the adapter will re-process the files on disk by default after an application restart. Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed. @@ -367,6 +373,7 @@ public class FtpJavaApplication { source.setLocalDirectory(new File("ftp-inbound")); source.setAutoCreateLocalDirectory(true); source.setLocalFilter(new AcceptOnceFileListFilter()); + source.setMaxFetchSize(1); return source; } diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index d9f596efef..5f42bb6c21 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -306,6 +306,7 @@ The _SFTP Inbound Channel Adapter_ is a special listener that will connect to th local-filename-generator-expression="#this.toUpperCase() + '.a'" local-filter="myFilter" temporary-file-suffix=".writing" + max-fetch-size="-1" delete-remote-files="false"> @@ -318,6 +319,11 @@ If you want to override this behavior you can set the `local-filename-generator- Unlike outbound gateways and adapters where the root object of the SpEL Evaluation Context is a `Message`, this inbound adapter does not yet have the Message at the time of evaluation since that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). +The inbound channel adapter first retrieves the file to a local directory and then emits each file according to the poller configuration. +Starting with _version 5.0_ you can now limit the number of files fetched from the FTP server when new file retrievals are needed. +This can be beneficial when the target files are very large and/or when running in a clustered system with a persistent file list filter discussed below. +Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved. + Starting with _Spring Integration 3.0_, you can specify the `preserve-timestamp` attribute (default `false`); when `true`, the local file's modified timestamp will be set to the value retrieved from the server; otherwise it will be set to the current time. Starting with _version 4.2_, you can specify `remote-directory-expression` instead of `remote-directory`, allowing @@ -476,6 +482,7 @@ public class SftpJavaApplication { source.setLocalDirectory(new File("ftp-inbound")); source.setAutoCreateLocalDirectory(true); source.setLocalFilter(new AcceptOnceFileListFilter()); + source.setMaxFetchSize(1); return source; } diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index a1e261f5c5..f1424f4f36 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -33,3 +33,7 @@ See <> for more information. Some inconsistencies with rendering IMAP mail content have been resolved. See <> for more information. + +==== (S)FTP Changes + +The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.