From f2fef2a7e92c9ae8233c495217a1fe6cef3a8826 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 6 May 2013 13:58:29 -0400 Subject: [PATCH] INT-2855 File Tailing Inbound Channel Adapter Two implementations: - native 'tail' command (e.g. tail -F -n 0 /foo/bar) - Apache commons-io Tailer Further changes: - Add documentation to section "what's new" - Add a JUnit @Rule so the OSDFTMP test doesn't fail on Windows --- build.gradle | 2 + .../file/config/FileNamespaceHandler.java | 5 +- ...eTailInboundChannelAdapterFactoryBean.java | 163 +++++++++++ .../FileTailInboundChannelAdapterParser.java | 55 ++++ ...acheCommonsFileTailingMessageProducer.java | 114 ++++++++ .../FileTailingMessageProducerSupport.java | 149 +++++++++++ ...SDelegatingFileTailingMessageProducer.java | 253 ++++++++++++++++++ .../integration/file/tail/package-info.java | 4 + .../config/spring-integration-file-3.0.xsd | 145 ++++++++++ ...boundChannelAdapterParserTests-context.xml | 54 ++++ ...eTailInboundChannelAdapterParserTests.java | 130 +++++++++ .../tail/FileTailingMessageProducerTests.java | 156 +++++++++++ .../integration/file/tail/TailAvailable.java | 32 +++ .../integration/file/tail/TailRule.java | 140 ++++++++++ src/reference/docbook/file.xml | 72 +++++ src/reference/docbook/whats-new.xml | 8 + 16 files changed, 1480 insertions(+), 2 deletions(-) create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/tail/package-info.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailAvailable.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java diff --git a/build.gradle b/build.gradle index 27b775abb7..f56eaf7367 100644 --- a/build.gradle +++ b/build.gradle @@ -35,6 +35,7 @@ subprojects { subproject -> aspectjVersion = '1.6.8' cglibVersion = '2.2' commonsNetVersion = '3.0.1' + commonsIoVersion = '2.4' derbyVersion = '10.9.1.0' easymockVersion = '2.3' groovyVersion = '2.1.0' @@ -203,6 +204,7 @@ project('spring-integration-file') { dependencies { compile project(":spring-integration-core") compile "org.springframework:spring-context:$springVersion" + compile("commons-io:commons-io:$commonsIoVersion", optional) testCompile project(":spring-integration-test") } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java index 8b86a87d92..5e8901b898 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -20,7 +20,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa /** * Namespace handler for Spring Integration's 'file' namespace. - * + * * @author Iwein Fuld * @author Mark Fisher */ @@ -32,6 +32,7 @@ public class FileNamespaceHandler extends AbstractIntegrationNamespaceHandler { registerBeanDefinitionParser("outbound-gateway", new FileOutboundGatewayParser()); registerBeanDefinitionParser("file-to-string-transformer", new FileToStringTransformerParser()); registerBeanDefinitionParser("file-to-bytes-transformer", new FileToByteArrayTransformerParser()); + registerBeanDefinitionParser("tail-inbound-channel-adapter", new FileTailInboundChannelAdapterParser()); } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java new file mode 100644 index 0000000000..da092a52fe --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java @@ -0,0 +1,163 @@ +/* + * Copyright 2002-2013 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.file.config; + +import java.io.File; + +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer; +import org.springframework.integration.file.tail.FileTailingMessageProducerSupport; +import org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.StringUtils; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBean + implements BeanNameAware { + + private volatile String nativeOptions; + + private volatile File file; + + private volatile TaskExecutor taskExecutor; + + private volatile TaskScheduler taskScheduler; + + private volatile Long delay; + + private volatile Long fileDelay; + + private volatile Boolean end; + + private volatile Boolean reopen; + + private volatile FileTailingMessageProducerSupport adapter; + + private volatile String beanName; + + private volatile MessageChannel outputChannel; + + private volatile Boolean autoStartup; + + private volatile Integer phase; + + public void setNativeOptions(String nativeOptions) { + this.nativeOptions = nativeOptions; + } + + public void setFile(File file) { + this.file = file; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + } + + public void setDelay(long delay) { + this.delay = delay; + } + + public void setFileDelay(long fileDelay) { + this.fileDelay = fileDelay; + } + + public void setEnd(Boolean end) { + this.end = end; + } + + public void setReopen(Boolean reopen) { + this.reopen = reopen; + } + + @Override + public void setBeanName(String name) { + this.beanName = name; + } + + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + @Override + public Class getObjectType() { + return this.adapter == null ? FileTailingMessageProducerSupport.class : this.adapter.getClass(); + } + + @Override + protected FileTailingMessageProducerSupport createInstance() throws Exception { + FileTailingMessageProducerSupport adapter; + if (this.delay == null && this.end == null && this.reopen == null) { + adapter = new OSDelegatingFileTailingMessageProducer(); + if (this.nativeOptions != null) { + ((OSDelegatingFileTailingMessageProducer) adapter).setOptions(this.nativeOptions); + } + } + else { + if (this.nativeOptions != null && StringUtils.hasText(this.nativeOptions) && logger.isWarnEnabled()) { + logger.warn("'native-options' are ignored with an Apache commons-io 'Tailer' adapter"); + } + adapter = new ApacheCommonsFileTailingMessageProducer(); + if (this.delay != null) { + ((ApacheCommonsFileTailingMessageProducer) adapter).setPollingDelay(this.delay); + } + if (this.end != null) { + ((ApacheCommonsFileTailingMessageProducer) adapter).setEnd(this.end); + } + if (this.reopen != null) { + ((ApacheCommonsFileTailingMessageProducer) adapter).setReopen(this.reopen); + } + } + adapter.setFile(this.file); + adapter.setTaskExecutor(this.taskExecutor); + if (this.taskScheduler != null) { + adapter.setTaskScheduler(this.taskScheduler); + } + if (this.fileDelay != null) { + adapter.setTailAttemptsDelay(this.fileDelay); + } + adapter.setOutputChannel(outputChannel); + adapter.setBeanName(this.beanName); + if (this.autoStartup != null) { + adapter.setAutoStartup(this.autoStartup); + } + if (this.phase != null) { + adapter.setPhase(this.phase); + } + adapter.afterPropertiesSet(); + this.adapter = adapter; + return adapter; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java new file mode 100644 index 0000000000..1c9571b562 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java @@ -0,0 +1,55 @@ +/* + * Copyright 2002-2013 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.file.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class FileTailInboundChannelAdapterParser extends AbstractChannelAdapterParser { + + @Override + protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(FileTailInboundChannelAdapterFactoryBean.class); + + if (StringUtils.hasText(channelName)) { + builder.addPropertyReference("outputChannel", channelName); + } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "native-options"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "file"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-scheduler"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "file-delay"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "end"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reopen"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); + + return builder.getBeanDefinition(); + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java new file mode 100644 index 0000000000..6f48ee9d49 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java @@ -0,0 +1,114 @@ +/* + * Copyright 2002-2013 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.file.tail; + +import org.apache.commons.io.input.Tailer; +import org.apache.commons.io.input.TailerListener; + +/** + * File tailer that delegates to the Apache Commons Tailer. + * + * @author Gary Russell + * @since 3.0 + * + */ +public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageProducerSupport + implements TailerListener { + + private volatile Tailer tailer; + + private volatile long pollingDelay = 1000; + + private volatile boolean end = true; + + private volatile boolean reopen = false; + + /** + * The delay between checks of the file for new content in milliseconds. + * @param pollingDelay The delay. + */ + public void setPollingDelay(long pollingDelay) { + this.pollingDelay = pollingDelay; + } + + /** + * If true, tail from the end of the file, otherwise + * include all lines from the beginning. Default true. + * @param end true or false + */ + public void setEnd(boolean end) { + this.end = end; + } + + /** + * If true, close and reopen the file between reading chunks; + * default false. + * @param reopen true or false. + */ + public void setReopen(boolean reopen) { + this.reopen = reopen; + } + + @Override + public String getComponentType() { + return super.getComponentType() + " (Apache)"; + } + + @Override + protected void doStart() { + super.doStart(); + Tailer tailer = new Tailer(this.getFile(), this, this.pollingDelay, this.end, this.reopen); + this.getTaskExecutor().execute(tailer); + this.tailer = tailer; + } + + @Override + protected void doStop() { + super.doStop(); + this.tailer.stop(); + } + + @Override + public void init(Tailer tailer) { + } + + @Override + public void fileNotFound() { + this.publish("File not found:" + this.getFile().getAbsolutePath()); + try { + Thread.sleep(this.getMissingFileDelay()); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + @Override + public void fileRotated() { + this.publish("File rotated:" + this.getFile().getAbsolutePath()); + } + + @Override + public void handle(String line) { + send(line); + } + + @Override + public void handle(Exception ex) { + this.publish(ex.getMessage()); + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java new file mode 100644 index 0000000000..a94c6e6da2 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java @@ -0,0 +1,149 @@ +/* + * Copyright 2002-2013 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.file.tail; + +import java.io.File; + +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.Message; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * Base class for file tailing inbound adapters. + * + * @author Gary Russell + * @since 3.0 + * + */ +public abstract class FileTailingMessageProducerSupport extends MessageProducerSupport + implements ApplicationEventPublisherAware { + + private volatile File file; + + private volatile ApplicationEventPublisher eventPublisher; + + private volatile TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + + private volatile long tailAttemptsDelay = 5000; + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.eventPublisher = applicationEventPublisher; + } + + /** + * The name of the file you wish to tail. + * @param file The absolute path of the file. + */ + public void setFile(File file) { + Assert.notNull("'file' cannot be null"); + this.file = file; + } + + protected File getFile() { + if (this.file == null) { + throw new IllegalStateException("No 'file' has been provided"); + } + return this.file; + } + + /** + * A task executor; default is a {@link SimpleAsyncTaskExecutor}. + * @param taskExecutor + */ + public void setTaskExecutor(TaskExecutor taskExecutor) { + Assert.notNull("'taskExecutor' cannot be null"); + this.taskExecutor = taskExecutor; + } + + /** + * The delay in milliseconds between attempts to tail a non-existent file, + * or between attempts to execute a process if it fails for any reason. + * @param missingFileDelay the delay. + */ + public void setTailAttemptsDelay(long tailAttemptsDelay) { + Assert.isTrue(tailAttemptsDelay > 0, "'tailAttemptsDelay' must be > 0"); + this.tailAttemptsDelay = tailAttemptsDelay; + } + + protected long getMissingFileDelay() { + return tailAttemptsDelay; + } + + protected TaskExecutor getTaskExecutor() { + return this.taskExecutor; + } + + @Override + public String getComponentType() { + return "file:tail-inbound-channel-adapter"; + } + + protected void send(String line) { + Message message = MessageBuilder.withPayload(line) + .setHeader(FileHeaders.FILENAME, this.file.getAbsolutePath()) + .build(); + super.sendMessage(message); + } + + protected void publish(String message) { + if (this.eventPublisher != null) { + FileTailingEvent event = new FileTailingEvent(this, message, this.file); + this.eventPublisher.publishEvent(event); + } + else { + logger.info("No publisher for event:" + message); + } + } + + public static class FileTailingEvent extends ApplicationEvent { + + private static final long serialVersionUID = -3382255736225946206L; + + private final String message; + + private final File file; + + public FileTailingEvent(Object source, String message, File file) { + super(source); + this.message = message; + this.file = file; + } + + protected String getMessage() { + return message; + } + + public File getFile() { + return file; + } + + @Override + public String toString() { + return "FileTailingEvent " + super.toString() + + " [message=" + this.message + + ", file=" + this.file.getAbsolutePath() + "]"; + } + + } +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java new file mode 100644 index 0000000000..16c0ec26ad --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java @@ -0,0 +1,253 @@ +/* + * Copyright 2002-2013 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.file.tail; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.util.Date; + +import org.springframework.integration.MessagingException; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.Assert; + +/** + * A file tailing message producer that delegates to the OS tail program. + * This is likely the most efficient mechanism on platforms that support it. + * Default options are "-F -n 0" (follow file name, no existing records). + * + * @author Gary Russell + * @since 3.0 + * + */ +public class OSDelegatingFileTailingMessageProducer extends FileTailingMessageProducerSupport + implements Runnable { + + private volatile Process process; + + private volatile String options = "-F -n 0"; + + private volatile String command = "ADAPTER_NOT_INITIALIZED"; + + private volatile BufferedReader reader; + + private volatile TaskScheduler scheduler; + + public void setOptions(String options) { + if (options == null) { + this.options = ""; + } + else { + this.options = options; + } + } + + @Override + public String getComponentType() { + return super.getComponentType() + " (native)"; + } + + @Override + protected void onInit() { + Assert.notNull(getFile(), "File cannot be null"); + super.onInit(); + this.command = "tail " + this.options + " " + this.getFile().getAbsolutePath(); + } + + @Override + protected void doStart() { + super.doStart(); + destroyProcess(); + this.getTaskExecutor().execute(new Runnable() { + + @Override + public void run() { + runExec(); + } + }); + } + + @Override + protected void doStop() { + super.doStop(); + destroyProcess(); + } + + private void destroyProcess() { + Process process = this.process; + if (process != null) { + process.destroy(); + this.process = null; + } + } + + /** + * Exec the native tail process. + */ + private void runExec() { + this.destroyProcess(); + if (logger.isInfoEnabled()) { + logger.info("Starting tail process"); + } + try { + Process process = Runtime.getRuntime().exec(this.command); + BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream())); + this.process = process; + this.startProcessMonitor(); + this.startStatusReader(); + this.reader = reader; + this.getTaskExecutor().execute(this); + } + catch (IOException e) { + throw new MessagingException("Failed to exec tail command: '" + this.command + "'", e); + } + } + + private TaskScheduler getRequiredTaskScheduler() { + if (this.scheduler == null) { + TaskScheduler taskScheduler = super.getTaskScheduler(); + if (taskScheduler == null) { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + taskScheduler = scheduler; + } + this.scheduler = taskScheduler; + } + return this.scheduler; + } + /** + * Runs a thread that waits for the Process result. + */ + private void startProcessMonitor() { + this.getTaskExecutor().execute(new Runnable() { + + @Override + public void run() { + Process process = OSDelegatingFileTailingMessageProducer.this.process; + if (process == null) { + if (logger.isDebugEnabled()) { + logger.debug("Process destroyed before starting process monitor"); + } + return; + } + + int result = Integer.MIN_VALUE; + try { + if (logger.isDebugEnabled()) { + logger.debug("Monitoring process " + process); + } + result = process.waitFor(); + if (logger.isInfoEnabled()) { + logger.info("tail process terminated with value " + result); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.error("Interrupted - stopping adapter", e); + stop(); + } + finally { + destroyProcess(); + } + if (isRunning()) { + if (logger.isInfoEnabled()) { + logger.info("Restarting tail process in " + getMissingFileDelay() + " milliseconds"); + } + getRequiredTaskScheduler().schedule(new Runnable() { + + @Override + public void run() { + runExec(); + } + }, new Date(System.currentTimeMillis() + getMissingFileDelay())); + } + } + }); + } + + /** + * Runs a thread that reads stderr - on some platforms status messages + * (file not available, rotations etc) are sent to stderr. + */ + private void startStatusReader() { + Process process = this.process; + if (process == null) { + if (logger.isDebugEnabled()) { + logger.debug("Process destroyed before starting stderr reader"); + } + return; + } + final BufferedReader errorReader = new BufferedReader(new InputStreamReader(process.getErrorStream())); + this.getTaskExecutor().execute(new Runnable() { + + @Override + public void run() { + String statusMessage; + if (logger.isDebugEnabled()) { + logger.debug("Reading stderr"); + } + try { + while ((statusMessage = errorReader.readLine()) != null) { + publish(statusMessage); + if (logger.isTraceEnabled()) { + logger.trace(statusMessage); + } + } + } + catch (IOException e) { + logger.error("Exception on tail error reader", e); + } + finally { + try { + errorReader.close(); + } + catch (IOException e) { + logger.error("Exception while closing stderr", e); + } + } + } + }); + } + + /** + * Reads lines from stdout and sends in a message to the output channel. + */ + @Override + public void run() { + String line; + try { + if (logger.isDebugEnabled()) { + logger.debug("Reading stdout"); + } + while ((line = this.reader.readLine()) != null) { + this.send(line); + } + } + catch (IOException e) { + logger.error("Exception on tail reader", e); + try { + this.reader.close(); + } + catch (IOException e1) { + + } + this.destroyProcess(); + } + } + + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/package-info.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/package-info.java new file mode 100644 index 0000000000..427fd622e0 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/package-info.java @@ -0,0 +1,4 @@ +/** + * Classes used for tailing file system files. + */ +package org.springframework.integration.file.tail; \ No newline at end of file diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-3.0.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-3.0.xsd index c72c6b5e19..ab6855010e 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-3.0.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-3.0.xsd @@ -193,6 +193,151 @@ Only files matching this regular expression will be picked up by this adapter. + + + + Configures an inbound channel adapter that 'tails' a file on the filesystem. + + + + + + + + + + + element. Therefore, if the "channel" attribute + is not provided, then the "id" attribute is required. + ]]> + + + + + + + + + + + Configures the adapter to exec 'tail' with these options (appended by the file name). + Default: "-F -n 0" (follow the filename and emit no + existing lines). This attribute is not allowed if 'delay' 'end' or 'reopen' is + specified, which cause the Apache commons-io 'Tailer' class to be used instead of + using a native 'tail' command. + + + + + + + The fully qualified name of the file to be tailed. + + + + + + + A reference to a TaskExecutor; the default is a SimpleAsyncTaskExecutor; the + native adapter uses three threads - one for reading stdout, one for + reading stderr and one for monitoring the process. + + + + + + + + + + + + A reference to a TaskScheduler; the default is the 'taskScheduler' bean + which is automatically configured for all Spring Integration applications. + The scheduler is used by the native adapter to reschedule + the 'tail' process after a failure according to the 'file-delay'. + This attribute is not allowed when using the Apache adapter. + + + + + + + + + + + + The delay in milliseconds between attempts to open the file when no file was found (Apache adapter). + For the native adapter, this is used as a delay before starting a new process after process failures. + On some platforms, when the file doesn't exist, the 'tail' process is suspended until the file + appears; on other platforms, the 'tail' process exits immediately if the file doesn't exist. + Default 5000. + + + + + + + Does not apply to the native adapter - the delay in milliseconds between polls when no new + data was detected in the file. Default 1000. Note: Setting this option forces the use of the Apache + Tailer implementation instead of the native 'tail' command. + + + + + + + Does not apply to the native adapter. + Set to 'true' to tail from the end of the file, 'false' to tail from the beginning of the file. + Default 'true'. Note: Setting this option forces the use of the Apache + Tailer implementation instead of the native 'tail' command. + + + + + + + + + + Does not apply to the native adapter. + If 'true', close and reopen the file between reading chunks. + Default 'false'. Note: Setting this option forces the use of the Apache + Tailer implementation instead of the native 'tail' command. + + + + + + + + + + Lifecycle attribute signaling if this component should be started during Application Context startup. + + + + + + + Lifecycle attribute signaling the phase in which this component should be started during + Application Context startup when 'auto-startup' is true. + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..60b7ebc22b --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..90fb8b250e --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java @@ -0,0 +1,130 @@ +/* + * Copyright 2002-2013 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.file.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; + +import java.io.File; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer; +import org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class FileTailInboundChannelAdapterParserTests { + + @Autowired @Qualifier("default") + private OSDelegatingFileTailingMessageProducer defaultAdapter; + + @Autowired @Qualifier("native") + private OSDelegatingFileTailingMessageProducer nativeAdapter; + + @Autowired + private ApacheCommonsFileTailingMessageProducer apacheDefault; + + @Autowired + private ApacheCommonsFileTailingMessageProducer apacheEndReopen; + + @Autowired + private TaskExecutor exec; + + @Autowired + private TaskScheduler sched; + + @Autowired + private TaskScheduler taskScheduler; + + @Test + public void testDefault() { + String fileName = TestUtils.getPropertyValue(defaultAdapter, "file", File.class).getAbsolutePath(); + String normalizedName = getNormalizedPath(fileName); + assertEquals("/tmp/baz", normalizedName); + assertEquals("tail -F -n 0 " + fileName, TestUtils.getPropertyValue(defaultAdapter, "command")); + assertSame(exec, TestUtils.getPropertyValue(defaultAdapter, "taskExecutor")); + assertFalse(TestUtils.getPropertyValue(defaultAdapter, "autoStartup", Boolean.class)); + assertEquals(123, TestUtils.getPropertyValue(defaultAdapter, "phase")); + } + + @Test + public void testNative() { + String fileName = TestUtils.getPropertyValue(nativeAdapter, "file", File.class).getAbsolutePath(); + String normalizedName = getNormalizedPath(fileName); + assertEquals("/tmp/foo", normalizedName); + assertEquals("tail -F -n 6 " + fileName, TestUtils.getPropertyValue(nativeAdapter, "command")); + assertSame(exec, TestUtils.getPropertyValue(nativeAdapter, "taskExecutor")); + assertSame(sched, TestUtils.getPropertyValue(nativeAdapter, "taskScheduler")); + assertFalse(TestUtils.getPropertyValue(nativeAdapter, "autoStartup", Boolean.class)); + assertEquals(123, TestUtils.getPropertyValue(nativeAdapter, "phase")); + assertEquals(456L, TestUtils.getPropertyValue(nativeAdapter, "tailAttemptsDelay")); + } + + @Test + public void testApacheDefault() { + String fileName = TestUtils.getPropertyValue(apacheDefault, "file", File.class).getAbsolutePath(); + String normalizedName = getNormalizedPath(fileName); + assertEquals("/tmp/bar", normalizedName); + assertSame(exec, TestUtils.getPropertyValue(apacheDefault, "taskExecutor")); + assertEquals(2000L, TestUtils.getPropertyValue(apacheDefault, "pollingDelay")); + assertEquals(10000L, TestUtils.getPropertyValue(apacheDefault, "tailAttemptsDelay")); + assertFalse(TestUtils.getPropertyValue(apacheDefault, "autoStartup", Boolean.class)); + assertEquals(123, TestUtils.getPropertyValue(apacheDefault, "phase")); + assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(apacheDefault, "end")); + assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(apacheDefault, "reopen")); + } + + @Test + public void testApacheEndReopen() { + String fileName = TestUtils.getPropertyValue(apacheEndReopen, "file", File.class).getAbsolutePath(); + String normalizedName = getNormalizedPath(fileName); + assertEquals("/tmp/qux", normalizedName); + assertSame(exec, TestUtils.getPropertyValue(apacheEndReopen, "taskExecutor")); + assertEquals(2000L, TestUtils.getPropertyValue(apacheEndReopen, "pollingDelay")); + assertEquals(10000L, TestUtils.getPropertyValue(apacheEndReopen, "tailAttemptsDelay")); + assertFalse(TestUtils.getPropertyValue(apacheEndReopen, "autoStartup", Boolean.class)); + assertEquals(123, TestUtils.getPropertyValue(apacheEndReopen, "phase")); + assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(apacheEndReopen, "end")); + assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(apacheEndReopen, "reopen")); + } + + /** + * Fix up windows paths. + */ + private String getNormalizedPath(String fileName) { + String absolutePath = fileName.replaceAll("\\\\", "/"); + int index = absolutePath.indexOf(":"); + if (index >= 0) { + absolutePath = absolutePath.substring(index + 1); + } + return absolutePath; + } +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java new file mode 100644 index 0000000000..238de37e99 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java @@ -0,0 +1,156 @@ +/* + * Copyright 2002-2013 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.file.tail; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +import java.io.File; +import java.io.FileOutputStream; +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.integration.Message; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.file.tail.FileTailingMessageProducerSupport.FileTailingEvent; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class FileTailingMessageProducerTests { + + private static final String TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES = "-F -n 99999999"; + + @Rule + public TailRule tailRule = new TailRule(TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES); + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final String tmpDir = System.getProperty("java.io.tmpdir"); + + private File testDir; + + private FileTailingMessageProducerSupport adapter; + + @Before + public void setup() { + File f = new File(tmpDir, "FileTailingMessageProducerTests"); + f.mkdir(); + this.testDir = f; + } + + @After + public void tearDown() { + if (this.adapter != null) { + adapter.stop(); + } + } + + @Test + @TailAvailable + public void testOS() throws Exception { + OSDelegatingFileTailingMessageProducer adapter = new OSDelegatingFileTailingMessageProducer(); + adapter.setOptions(TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES); + testGuts(adapter, "reader"); + } + + @Test + public void testApache() throws Exception { + ApacheCommonsFileTailingMessageProducer adapter = new ApacheCommonsFileTailingMessageProducer(); + adapter.setPollingDelay(100); + adapter.setEnd(false); + testGuts(adapter, "tailer"); + } + + private void testGuts(FileTailingMessageProducerSupport adapter, String field) + throws Exception { + this.adapter = adapter; + final List events = new ArrayList(); + adapter.setApplicationEventPublisher(new ApplicationEventPublisher() { + @Override + public void publishEvent(ApplicationEvent event) { + FileTailingEvent tailEvent = (FileTailingEvent) event; + logger.warn(event); + events.add(tailEvent); + } + }); + adapter.setFile(new File(testDir, "foo")); + QueueChannel outputChannel = new QueueChannel(); + adapter.setOutputChannel(outputChannel); + adapter.setTailAttemptsDelay(500); + adapter.afterPropertiesSet(); + File file = new File(testDir, "foo"); + File renamed = new File(testDir, "bar"); + file.delete(); + renamed.delete(); + adapter.start(); + waitForField(adapter, field); + FileOutputStream foo = new FileOutputStream(file); + for (int i = 0; i < 50; i++) { + foo.write(("hello" + i + "\n").getBytes()); + } + foo.flush(); + for (int i = 0; i < 50; i++) { + Message message = outputChannel.receive(5000); + assertNotNull("expected a non-null message", message); + assertEquals("hello" + i, message.getPayload()); + } + file.renameTo(renamed); + foo.close(); + foo = new FileOutputStream(file); + if (adapter instanceof ApacheCommonsFileTailingMessageProducer) { + Thread.sleep(1000); + } + for (int i = 50; i < 100; i++) { + foo.write(("hello" + i + "\n").getBytes()); + } + foo.flush(); + for (int i = 50; i < 100; i++) { + Message message = outputChannel.receive(3000); + assertNotNull("expected a non-null message", message); + assertEquals("hello" + i, message.getPayload()); + } + foo.close(); + } + + private void waitForField(FileTailingMessageProducerSupport adapter, String field) throws Exception { + int n = 0; + DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); + while (n < 100) { + if (accessor.getPropertyValue(field) == null) { + Thread.sleep(100); + } + else { + return; + } + } + fail("adapter failed to start"); + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailAvailable.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailAvailable.java new file mode 100644 index 0000000000..81479a0399 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailAvailable.java @@ -0,0 +1,32 @@ +/* + * Copyright 2002-2013 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.file.tail; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +@Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.METHOD}) +public @interface TailAvailable { + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java new file mode 100644 index 0000000000..fc3f6a4cbc --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java @@ -0,0 +1,140 @@ +/* + * Copyright 2002-2013 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.file.tail; + +import static org.junit.Assert.assertFalse; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.junit.rules.TestWatcher; +import org.junit.runner.Description; +import org.junit.runners.model.Statement; + +/** + * Ignores tests annotated with {@link TailAvailable} if 'tail' with the requested options + * does not work on this platform. + * @author Gary Russell + * @since 3.0 + * + */ +public class TailRule extends TestWatcher { + + private static final Log logger = LogFactory.getLog(TailRule.class); + + private static final String tmpDir = System.getProperty("java.io.tmpdir"); + + private final String commandToTest; + + public TailRule(String optionsToTest) { + this.commandToTest = "tail " + optionsToTest + " "; + } + + @Override + public Statement apply(Statement base, Description description) { + if (description.getAnnotation(TailAvailable.class) != null) { + if (!tailWorksOnThisMachine()) { + return new Statement() { + + @Override + public void evaluate() throws Throwable { + // skip + }}; + } + } + return super.apply(base, description); + } + + private boolean tailWorksOnThisMachine() { + if (tmpDir.contains(":")) { + return false; + } + File testDir = new File(tmpDir, "FileTailingMessageProducerTests"); + testDir.mkdir(); + final File file = new File(testDir, "foo"); + int result = -99; + try { + OutputStream fos = new FileOutputStream(file); + fos.write("foo".getBytes()); + fos.close(); + final AtomicReference c = new AtomicReference(); + final CountDownLatch latch = new CountDownLatch(1); + Future future = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Process call() throws Exception { + final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath()); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + c.set(process.getInputStream().read()); + latch.countDown(); + } + catch (IOException e) { + logger.error("Error reading test stream", e); + } + } + }); + return process; + } + }); + try { + Process process = future.get(10, TimeUnit.SECONDS); + if (latch.await(10, TimeUnit.SECONDS)) { + Integer read = c.get(); + if (read != null && read == 'f') { + result = 0; + } + } + process.destroy(); + } + catch (ExecutionException e) { + result = -999; + } + file.delete(); + } + catch (Exception e) { + logger.error("failed to test tail", e); + } + if (result != 0) { + logger.warn("tail command is not available on this platform; result:" + result); + } + return result == 0; + } + + public static class TestRule { + + @Test + public void test1() { + TailRule rule = new TailRule("-BLAH"); + assertFalse(rule.tailWorksOnThisMachine()); + } + } +} diff --git a/src/reference/docbook/file.xml b/src/reference/docbook/file.xml index a3425acaae..6491570c1f 100644 --- a/src/reference/docbook/file.xml +++ b/src/reference/docbook/file.xml @@ -149,6 +149,78 @@ This gives you full freedom to choose the ordering, listing and locking strategies. +
+ 'Tail'ing Files + + Another popular use case is to get 'lines' from the end (or tail) of a file. Two implementations are provided; + the first, OSDelegatingFileTailingMessageProducer, uses the native tail + command (on operating systems that have one). This is likely the most efficient implementation on those + platforms. For operating systems that do not have a tail command, the second implementation + ApacheCommonsFileTailingMessageProducer which uses the Apache commons-io + Tailer class. + + + In both cases, file system events, such as files being unavailable etc, are published as + ApplicationEvents using the normal Spring event publishing mechanism. + Examples of such events are: + + + [message=tail: cannot open `/tmp/foo' for reading: + No such file or directory, file=/tmp/foo] + + + [message=tail: `/tmp/foo' has become accessible, file=/tmp/foo] + + + [message=tail: `/tmp/foo' has become inaccessible: + No such file or directory, file=/tmp/foo] + + + [message=tail: `/tmp/foo' has appeared; + following end of new file, file=/tmp/foo] + + + This sequence of events might occur, for example, when a file is rotated. + + + Not all platforms supporting a tail command provide these status messages. + + + Example configurations: + + ]]> + + This creates a native adapter with default '-F -n 0' options (follow the file name from the current end). + + ]]> + + This creates a native adapter with '-F -n 6' options (follow the file name, emit up to 6 lines before the current end). + If the tail command fails (on some platforms, a missing file causes the tail to fail, even with + -F specified), the command will be retried every 10 seconds. + + ]]> + + This creates a commons-io Tailer adapter that examines the file for new lines every + 2 seconds, and checks for existence of a missing file every 10 seconds. The file will be tailed from the + beginning (end="false") instead of the end (which is the default). The file will be + reopened for each chunk (the default is to keep the file open). + +
Writing files diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 66cb225ec4..9d549d129a 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -60,6 +60,14 @@ .
+
+ 'Tail' Support + + File 'tail'ing inbound channel adapters are now provided to generate messages when + lines are added to the end of text files. + . + +