From 5f450ed9596a8e6741a251d6139eb686e2ce1fd6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 20 Jan 2017 15:00:17 -0500 Subject: [PATCH] INT-4212: FileWritingMessageHandler.flushWhenIdle JIRA: https://jira.spring.io/browse/INT-4212 Add an option to flush after the `flushInterval`, regardless of intermediate writes. Rename `lastFlush` to `firstWrite` Polishing **Cherry-pick to 4.3.x** --- .../file/FileWritingMessageHandler.java | 24 ++++++++++++++-- ...ngMessageHandlerBeanDefinitionBuilder.java | 3 +- .../FileWritingMessageHandlerFactoryBean.java | 11 +++++++- .../config/spring-integration-file-5.0.xsd | 16 ++++++++++- .../file/FileWritingMessageHandlerTests.java | 28 ++++++++++++++++++- ...boundChannelAdapterParserTests-context.xml | 1 + ...FileOutboundChannelAdapterParserTests.java | 4 ++- src/reference/asciidoc/file.adoc | 3 ++ 8 files changed, 83 insertions(+), 7 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index 6b0d10d9b6..39b9167ed8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -143,6 +143,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand private volatile long flushInterval = DEFAULT_FLUSH_INTERVAL; + private volatile boolean flushWhenIdle = true; + private volatile ScheduledFuture flushTask; private volatile MessageFlushPredicate flushPredicate = new DefaultFlushPredicate(); @@ -303,12 +305,27 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand * {@code flushInterval} and {@code flushInterval * 1.33} with an average of * {@code flushInterval * 1.167}. * @param flushInterval the interval. + * @see #setFlushWhenIdle(boolean) * @since 4.3 */ public void setFlushInterval(long flushInterval) { this.flushInterval = flushInterval; } + /** + * Determine whether the {@link #setFlushInterval(long) flushInterval} applies only + * to idle files (default) or whether to flush on that interval after the first + * write to a previously flushed or new file. + * @param flushWhenIdle false to flush on the interval after the first write + * to a closed file. + * @see #setFlushInterval(long) + * @see #setBufferSize(int) + * @since 4.3.7 + */ + public void setFlushWhenIdle(boolean flushWhenIdle) { + this.flushWhenIdle = flushWhenIdle; + } + @Override public void setTaskScheduler(TaskScheduler taskScheduler) { super.setTaskScheduler(taskScheduler); @@ -876,6 +893,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand private final BufferedOutputStream stream; + private final long firstWrite = System.currentTimeMillis(); + private volatile long lastWrite; FileState(BufferedWriter writer) { @@ -919,7 +938,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand while (iterator.hasNext()) { Entry entry = iterator.next(); FileState state = entry.getValue(); - if (state.lastWrite < expired) { + if (state.lastWrite < expired || + (!FileWritingMessageHandler.this.flushWhenIdle && state.firstWrite < expired)) { iterator.remove(); state.close(); if (FileWritingMessageHandler.this.logger.isDebugEnabled()) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java index 96f965bd9c..a609c2b496 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -73,6 +73,7 @@ abstract class FileWritingMessageHandlerBeanDefinitionBuilder { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "charset"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "buffer-size"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "flush-interval"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "flush-when-idle"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "flush-predicate"); String remoteFileNameGenerator = element.getAttribute("filename-generator"); String remoteFileNameGeneratorExpression = element.getAttribute("filename-generator-expression"); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java index d5566f9463..81fd1bc1c2 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -69,6 +69,8 @@ public class FileWritingMessageHandlerFactoryBean private volatile Long flushInterval; + private volatile Boolean flushWhenIdle; + private volatile MessageFlushPredicate flushPredicate; public void setFileExistsMode(String fileExistsModeAsString) { @@ -127,6 +129,10 @@ public class FileWritingMessageHandlerFactoryBean this.flushInterval = flushInterval; } + public void setFlushWhenIdle(boolean flushWhenIdle) { + this.flushWhenIdle = flushWhenIdle; + } + public void setFlushPredicate(MessageFlushPredicate flushPredicate) { this.flushPredicate = flushPredicate; } @@ -183,6 +189,9 @@ public class FileWritingMessageHandlerFactoryBean if (this.flushInterval != null) { handler.setFlushInterval(this.flushInterval); } + if (this.flushWhenIdle != null) { + handler.setFlushWhenIdle(this.flushWhenIdle); + } if (this.flushPredicate != null) { handler.setFlushPredicate(this.flushPredicate); } diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd index 963fcc4f59..827e067341 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd @@ -549,12 +549,26 @@ Only files matching this regular expression will be picked up by this adapter. - When using 'mode=APPEND_NO_FLUSH' if this time (ms) elapses + When using 'mode=APPEND_NO_FLUSH', if this time (ms) elapses without any new writes, the data is flushed and the file closed. Default 30000. + + + + When using 'mode=APPEND_NO_FLUSH', set to false to indicate the + 'flush-interval' starts from the first new write to a previously + flushed (or new) file. When true, the interval starts from the + last write (the file is flushed if it has no writes during the interval). + Default true. + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java index 2dc97258db..1f18f47484 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -29,7 +29,11 @@ import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.ArgumentMatchers.startsWith; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; import java.io.ByteArrayInputStream; import java.io.File; @@ -39,14 +43,17 @@ import java.io.IOException; import java.io.InputStream; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Matcher; +import org.apache.commons.logging.Log; import org.junit.Before; import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; @@ -490,6 +497,25 @@ public class FileWritingMessageHandlerTests { }); assertThat(file.length(), equalTo(24L)); assertTrue(called.get()); + + handler.stop(); + Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class)); + new DirectFieldAccessor(handler).setPropertyValue("logger", logger); + when(logger.isDebugEnabled()).thenReturn(true); + final AtomicInteger flushes = new AtomicInteger(); + doAnswer(i -> { + flushes.incrementAndGet(); + return null; + }).when(logger).debug(startsWith("Flushed:")); + handler.setFlushInterval(50); + handler.setFlushWhenIdle(false); + handler.start(); + for (int i = 0; i < 40; i++) { + handler.handleMessage(new GenericMessage("foo")); + Thread.sleep(5); + } + assertThat(flushes.get(), greaterThanOrEqualTo(2)); + handler.stop(); } void assertFileContentIsMatching(Message result) throws IOException { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml index 7fea9b0dbc..4a5839b473 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml @@ -21,6 +21,7 @@ diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java index 1fe287a6f8..03a763ea87 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -139,6 +139,7 @@ public class FileOutboundChannelAdapterParserTests { assertNotNull(expression); assertEquals("'foo.txt'", expression.getExpressionString()); assertEquals(Boolean.FALSE, handlerAccessor.getPropertyValue("deleteSourceFiles")); + assertEquals(Boolean.TRUE, handlerAccessor.getPropertyValue("flushWhenIdle")); } @Test @@ -156,6 +157,7 @@ public class FileOutboundChannelAdapterParserTests { assertEquals(expected, actual); assertTrue(handlerAccessor.getPropertyValue("fileNameGenerator") instanceof CustomFileNameGenerator); assertEquals(".writing", handlerAccessor.getPropertyValue("temporaryFileSuffix")); + assertEquals(Boolean.FALSE, handlerAccessor.getPropertyValue("flushWhenIdle")); } @Test diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 3a97d78f51..ac00ae0b83 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -661,6 +661,9 @@ or `FileWritingMessageHandler.MessageFlushPredicate` implementation. The predicates are called for each open file. See the java docs for these interfaces for more information. +When using `flushInterval`, the interval starts at the last write - the file is flushed only if it is idle for the interval. +Starting with _version 4.3.7_, and additional property `flushWhenIdle` can be set to `false`, meaning that the interval starts with the first write to a previously flushed (or new) file. + [[file-timestamps]] ==== File Timestamps