This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new f70be09 CAMEL-12551: stream component in file mode should support
file-rollover when an existing file is re-written. Using similar file watcher
logic we have in camel-core for route reload strategy.
f70be09 is described below
commit f70be09f45a7a6a512f60c5cfa1b8b7c82c59026
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Jun 4 16:10:25 2018 +0200
CAMEL-12551: stream component in file mode should support file-rollover
when an existing file is re-written. Using similar file watcher logic we have
in camel-core for route reload strategy.
---
.../src/main/docs/stream-component.adoc | 40 ++--
.../component/stream/FileWatcherStrategy.java | 220 +++++++++++++++++++++
.../camel/component/stream/StreamConsumer.java | 46 ++++-
.../camel/component/stream/StreamEndpoint.java | 22 ++-
.../camel/component/stream/ScanStreamFileTest.java | 47 ++++-
5 files changed, 337 insertions(+), 38 deletions(-)
diff --git a/components/camel-stream/src/main/docs/stream-component.adoc
b/components/camel-stream/src/main/docs/stream-component.adoc
index dbb7af2..e0f5a69 100644
--- a/components/camel-stream/src/main/docs/stream-component.adoc
+++ b/components/camel-stream/src/main/docs/stream-component.adoc
@@ -19,7 +19,7 @@ for this component:
</dependency>
------------------------------------------------------------
-### URI format
+=== URI format
[source,java]
-----------------------
@@ -44,7 +44,7 @@ producers (that is, it cannot appear in `from()`).
You can append query options to the URI in the following format,
`?option=value&option=value&...`
-### Options
+=== Options
// component options: START
@@ -72,7 +72,7 @@ with the following path and query parameters:
|===
-==== Query Parameters (18 parameters):
+==== Query Parameters (19 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -82,12 +82,13 @@ with the following path and query parameters:
| *fileName* (common) | When using the stream:file URI format, this option
specifies the filename to stream to/from. | | String
| *url* (common) | When using the stream:url URI format, this option specifies
the URL to stream to/from. The input/output stream will be opened using the JDK
URLConnection facility. | | String
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions occurred while the
consumer is trying to pickup incoming messages, or the likes, will now be
processed as a message and handled by the routing Error Handler. By default the
consumer will use the org.apache.camel.spi.ExceptionHandler to deal with
exceptions, that will be logged at WARN or ERROR level and ignored. | false |
boolean
+| *fileWatcher* (consumer) | To use JVM file watcher to listen for file change
events to support re-loading files that may be overwritten, somewhat like tail
--retry | false | boolean
| *groupLines* (consumer) | To group X number of lines in the consumer. For
example to group 10 lines and therefore only spit out an Exchange with 10
lines, instead of 1 Exchange per line. | | int
| *groupStrategy* (consumer) | Allows to use a custom GroupStrategy to control
how to group lines. | | GroupStrategy
| *initialPromptDelay* (consumer) | Initial delay in milliseconds before
showing the message prompt. This delay occurs only once. Can be used during
system startup to avoid message prompts being written while other logging is
done to the system out. | 2000 | long
| *promptDelay* (consumer) | Optional delay in milliseconds before showing the
message prompt. | | long
| *promptMessage* (consumer) | Message prompt to use when reading from
stream:in; for example, you could set this to Enter a command: | | String
-| *retry* (consumer) | Will retry opening the file if it's overwritten,
somewhat like tail --retry | false | boolean
+| *retry* (consumer) | Will retry opening the stream if it's overwritten,
somewhat like tail --retry If reading from files then you should also enable
the fileWatcher option, to make it work reliable. | false | boolean
| *scanStream* (consumer) | To be used for continuously reading a stream such
as the unix tail command. | false | boolean
| *scanStreamDelay* (consumer) | Delay in milliseconds between read attempts
when using scanStream. | | long
| *exceptionHandler* (consumer) | To let the consumer use a custom
ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this
options is not in use. By default the consumer will deal with exceptions, that
will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
@@ -100,7 +101,7 @@ with the following path and query parameters:
// endpoint options: END
-### Message content
+=== Message content
The *stream:* component supports either `String` or `byte[]` for writing
to streams. Just add either `String` or `byte[]` content to the
@@ -113,13 +114,13 @@ add a `java.io.OutputStream` object to
`message.in.header` in the key
`header`. +
See samples for an example.
-### Samples
+=== Samples
In the following sample we route messages from the `direct:in` endpoint
to the `System.out` stream:
[source,java]
----------------------------------------------------------------
+----
// Route messages to the standard output.
from("direct:in").to("stream:out");
@@ -130,7 +131,7 @@ template.sendBody("direct:in", "Hello Text World");
// Send byte[] payload to the standard output.
// No newline will be added after the message.
template.sendBody("direct:in", "Hello Bytes World".getBytes());
----------------------------------------------------------------
+----
The following sample demonstrates how the header type can be used to
determine which stream to use. In the sample we use our own output
@@ -140,18 +141,15 @@ The following sample demonstrates how to continuously
read a file stream
(analogous to the UNIX `tail` command):
[source,java]
-------------------------------------------------------------------------------------------------------------------------------------
-from("stream:file?fileName=/server/logs/server.log&scanStream=true&scanStreamDelay=1000").to("bean:logService?method=parseLogLine");
-------------------------------------------------------------------------------------------------------------------------------------
-
-One gotcha with scanStream (pre Camel 2.7) or scanStream + retry is the
-file will be re-opened and scanned with each iteration of
-scanStreamDelay. Until NIO2 is available we cannot reliably detect when
-a file is deleted/recreated.
+----
+from("stream:file?fileName=/server/logs/server.log&scanStream=true&scanStreamDelay=1000")
+ .to("bean:logService?method=parseLogLine");
+----
-### See Also
+If you want to re-load the file if it rollover/rewritten then you should also
turn on the `fileWatcher` and `retry` options.
-* Configuring Camel
-* Component
-* Endpoint
-* Getting Started
+[source,java]
+----
+from("stream:file?fileName=/server/logs/server.log&scanStream=true&scanStreamDelay=1000&retry=true&fileWatcher=true")
+ .to("bean:logService?method=parseLogLine");
+----
diff --git
a/components/camel-stream/src/main/java/org/apache/camel/component/stream/FileWatcherStrategy.java
b/components/camel-stream/src/main/java/org/apache/camel/component/stream/FileWatcherStrategy.java
new file mode 100644
index 0000000..e3b9311
--- /dev/null
+++
b/components/camel-stream/src/main/java/org/apache/camel/component/stream/FileWatcherStrategy.java
@@ -0,0 +1,220 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.stream;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.WatchEvent;
+import java.nio.file.WatchKey;
+import java.nio.file.WatchService;
+import java.util.Locale;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
+import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.support.ServiceSupport;
+import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * To watch for file changes/rollover via JDK file watcher API.
+ * This is used to know for example of streaming from a file, that gets
rolled-over, so we know about this,
+ * and can begin reading the file again from the beginning.
+ */
+public class FileWatcherStrategy extends ServiceSupport implements
CamelContextAware {
+
+ @FunctionalInterface
+ public interface OnChangeEvent {
+
+ void onChange(File file);
+
+ }
+
+ private static final Logger LOG =
LoggerFactory.getLogger(FileWatcherStrategy.class);
+ private CamelContext camelContext;
+ private final String directory;
+ private final OnChangeEvent onChangeEvent;
+ private WatchService watcher;
+ private ExecutorService executorService;
+ private WatchFileChangesTask task;
+ private long pollTimeout = 1000;
+
+ public FileWatcherStrategy(String directory, OnChangeEvent onChangeEvent) {
+ this.directory = directory;
+ this.onChangeEvent = onChangeEvent;
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ public long getPollTimeout() {
+ return pollTimeout;
+ }
+
+ /**
+ * Sets the poll timeout in millis. The default value is 1000.
+ */
+ public void setPollTimeout(long pollTimeout) {
+ this.pollTimeout = pollTimeout;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (directory == null) {
+ // no folder configured
+ return;
+ }
+
+ File dir = new File(directory);
+ if (dir.exists() && dir.isDirectory()) {
+ LOG.info("Starting ReloadStrategy to watch directory: {}", dir);
+
+ WatchEvent.Modifier modifier = null;
+
+ // if its mac OSX then attempt to apply workaround or warn its
slower
+ String os = ObjectHelper.getSystemProperty("os.name", "");
+ if (os.toLowerCase(Locale.US).startsWith("mac")) {
+ // this modifier can speedup the scanner on mac osx (as java
on mac has no native file notification integration)
+ Class<WatchEvent.Modifier> clazz =
getCamelContext().getClassResolver().resolveClass("com.sun.nio.file.SensitivityWatchEventModifier",
WatchEvent.Modifier.class);
+ if (clazz != null) {
+ WatchEvent.Modifier[] modifiers = clazz.getEnumConstants();
+ for (WatchEvent.Modifier mod : modifiers) {
+ if ("HIGH".equals(mod.name())) {
+ modifier = mod;
+ break;
+ }
+ }
+ }
+ if (modifier != null) {
+ LOG.info("On Mac OS X the JDK WatchService is slow by
default so enabling SensitivityWatchEventModifier.HIGH as workaround");
+ } else {
+ LOG.warn("On Mac OS X the JDK WatchService is slow and it
may take up till 10 seconds to notice file changes");
+ }
+ }
+
+ try {
+ Path path = dir.toPath();
+ watcher = path.getFileSystem().newWatchService();
+ registerPathToWatcher(modifier, path, watcher);
+
+ task = new WatchFileChangesTask(watcher, path, onChangeEvent);
+
+ executorService =
getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
"FileWatcherStrategy");
+ executorService.submit(task);
+ } catch (IOException e) {
+ throw ObjectHelper.wrapRuntimeCamelException(e);
+ }
+ }
+ }
+
+ private WatchKey registerPathToWatcher(WatchEvent.Modifier modifier, Path
path, WatchService watcher) throws IOException {
+ WatchKey key;
+ if (modifier != null) {
+ key = path.register(watcher, new
WatchEvent.Kind<?>[]{ENTRY_CREATE, ENTRY_MODIFY}, modifier);
+ } else {
+ key = path.register(watcher, ENTRY_CREATE, ENTRY_MODIFY);
+ }
+ return key;
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (executorService != null) {
+
getCamelContext().getExecutorServiceManager().shutdownGraceful(executorService);
+ executorService = null;
+ }
+ if (watcher != null) {
+ IOHelper.close(watcher);
+ }
+ }
+
+ /**
+ * Background task which watches for file changes
+ */
+ protected class WatchFileChangesTask implements Runnable {
+
+ private final WatchService watcher;
+ private final Path folder;
+ private volatile boolean running;
+ private OnChangeEvent changeEvent;
+
+ public WatchFileChangesTask(WatchService watcher, Path folder,
OnChangeEvent changeEvent) {
+ this.watcher = watcher;
+ this.folder = folder;
+ this.changeEvent = changeEvent;
+ }
+
+ public boolean isRunning() {
+ return running;
+ }
+
+ public void run() {
+ LOG.debug("FileWatcherStrategy is starting watching folder: {}",
folder);
+
+ // allow to run while starting Camel
+ while (isStarting() || isRunAllowed()) {
+ running = true;
+
+ WatchKey key;
+ try {
+ LOG.trace("FileWatcherStrategy is polling for file changes
in directory: {}", folder);
+ // wait for a key to be available
+ key = watcher.poll(pollTimeout, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException ex) {
+ break;
+ }
+
+ if (key != null) {
+ Path pathToReload = folder;
+
+ for (WatchEvent<?> event : key.pollEvents()) {
+ WatchEvent<Path> we = (WatchEvent<Path>) event;
+ Path path = we.context();
+ File file = pathToReload.resolve(path).toFile();
+ LOG.trace("Modified/Created/Deleted file: {}", file);
+ changeEvent.onChange(file);
+ }
+
+ // the key must be reset after processed
+ boolean valid = key.reset();
+ if (!valid) {
+ break;
+ }
+ }
+ }
+
+ running = false;
+
+ LOG.info("FileWatcherStrategy is stopping watching folder: {}",
folder);
+ }
+ }
+
+}
diff --git
a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
index 78e5ec3..c2cd374 100644
---
a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
+++
b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
@@ -34,9 +34,10 @@ import java.util.concurrent.ExecutorService;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.util.FileUtil;
import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
-
+import org.apache.camel.util.ServiceHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,11 +51,14 @@ public class StreamConsumer extends DefaultConsumer
implements Runnable {
private static final String INVALID_URI = "Invalid uri, valid form:
'stream:{" + TYPES + "}'";
private static final List<String> TYPES_LIST =
Arrays.asList(TYPES.split(","));
private ExecutorService executor;
+ private FileWatcherStrategy fileWatcher;
+ private volatile boolean watchFileChanged;
private volatile InputStream inputStream = System.in;
private volatile InputStream inputStreamToClose;
+ private volatile File file;
private StreamEndpoint endpoint;
private String uri;
- private boolean initialPromptDone;
+ private volatile boolean initialPromptDone;
private final List<String> lines = new CopyOnWriteArrayList<>();
public StreamConsumer(StreamEndpoint endpoint, Processor processor, String
uri) throws Exception {
@@ -68,6 +72,22 @@ public class StreamConsumer extends DefaultConsumer
implements Runnable {
protected void doStart() throws Exception {
super.doStart();
+ // use file watch service if we read from file
+ if (endpoint.isFileWatcher()) {
+ String dir = new File(endpoint.getFileName()).getParent();
+ fileWatcher = new FileWatcherStrategy(dir, (file) -> {
+ String onlyName = file.getName();
+ String target = FileUtil.stripPath(endpoint.getFileName());
+ LOG.trace("File changed: {}", onlyName);
+ if (onlyName.equals(target)) {
+ // file is changed
+ watchFileChanged = true;
+ }
+ });
+ fileWatcher.setCamelContext(getEndpoint().getCamelContext());
+ }
+ ServiceHelper.startService(fileWatcher);
+
// if we scan the stream we are lenient and can wait for the stream to
be available later
if (!endpoint.isScanStream()) {
initializeStream();
@@ -87,6 +107,7 @@ public class StreamConsumer extends DefaultConsumer
implements Runnable {
endpoint.getCamelContext().getExecutorServiceManager().shutdownNow(executor);
executor = null;
}
+ ServiceHelper.stopAndShutdownService(fileWatcher);
lines.clear();
// do not close regular inputStream as it may be System.in etc.
@@ -145,9 +166,22 @@ public class StreamConsumer extends DefaultConsumer
implements Runnable {
if (!eos && isRunAllowed()) {
index = processLine(line, false, index);
} else if (eos && isRunAllowed() && endpoint.isRetry()) {
- //try and re-open stream
- br = initializeStream();
+ boolean reOpen = true;
+ if (endpoint.isFileWatcher()) {
+ reOpen = watchFileChanged;
+ }
+ if (reOpen) {
+ LOG.debug("File: {} changed/rollover, re-reading file
from beginning", file);
+ br = initializeStream();
+ // we have re-initialized the stream so lower changed
flag
+ if (endpoint.isFileWatcher()) {
+ watchFileChanged = false;
+ }
+ } else {
+ LOG.trace("File: {} not changed since last read",
file);
+ }
}
+
// sleep only if there is no input
if (eos) {
try {
@@ -271,10 +305,10 @@ public class StreamConsumer extends DefaultConsumer
implements Runnable {
FileInputStream fileStream;
- File file = new File(fileName);
+ file = new File(fileName);
if (LOG.isDebugEnabled()) {
- LOG.debug("File to be scanned : {}, path : {}", file.getName(),
file.getAbsolutePath());
+ LOG.debug("File to be scanned: {}, path: {}", file.getName(),
file.getAbsolutePath());
}
if (file.canRead()) {
diff --git
a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
index ddb8270..2e0ca70 100644
---
a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
+++
b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
@@ -50,6 +50,8 @@ public class StreamEndpoint extends DefaultEndpoint {
private boolean scanStream;
@UriParam(label = "consumer")
private boolean retry;
+ @UriParam(label = "consumer")
+ private boolean fileWatcher;
@UriParam(label = "producer")
private boolean closeOnDone;
@UriParam(label = "consumer")
@@ -82,6 +84,9 @@ public class StreamEndpoint extends DefaultEndpoint {
public Consumer createConsumer(Processor processor) throws Exception {
StreamConsumer answer = new StreamConsumer(this, processor,
getEndpointUri());
+ if (isFileWatcher() && !"file".equals(getKind())) {
+ throw new IllegalArgumentException("File watcher is only possible
if reading streams from files");
+ }
configureConsumer(answer);
return answer;
}
@@ -224,12 +229,25 @@ public class StreamEndpoint extends DefaultEndpoint {
}
/**
- * Will retry opening the file if it's overwritten, somewhat like tail
--retry
+ * Will retry opening the stream if it's overwritten, somewhat like tail
--retry
+ * <p/>
+ * If reading from files then you should also enable the fileWatcher
option, to make it work reliable.
*/
public void setRetry(boolean retry) {
this.retry = retry;
}
-
+
+ public boolean isFileWatcher() {
+ return fileWatcher;
+ }
+
+ /**
+ * To use JVM file watcher to listen for file change events to support
re-loading files that may be overwritten, somewhat like tail --retry
+ */
+ public void setFileWatcher(boolean fileWatcher) {
+ this.fileWatcher = fileWatcher;
+ }
+
public boolean isCloseOnDone() {
return closeOnDone;
}
diff --git
a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java
b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java
index 60096f0..d902b5b 100644
---
a/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java
+++
b/components/camel-stream/src/test/java/org/apache/camel/component/stream/ScanStreamFileTest.java
@@ -54,11 +54,15 @@ public class ScanStreamFileTest extends CamelTestSupport {
// a scanStream=true is never finished
mock.message(1).header(StreamConstants.STREAM_COMPLETE).isEqualTo(false);
+ context.startAllRoutes();
+
FileOutputStream fos = new FileOutputStream(file);
try {
fos.write("Hello\n".getBytes());
Thread.sleep(150);
fos.write("World\n".getBytes());
+ // ensure it does not read the file again
+ Thread.sleep(1000);
} finally {
fos.close();
}
@@ -69,20 +73,42 @@ public class ScanStreamFileTest extends CamelTestSupport {
@Test
public void testScanRefreshedFile() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedMinimumMessageCount(3);
+ mock.expectedMessageCount(5);
+
+ // write file during started route
FileOutputStream fos = refreshFile(null);
try {
- fos.write("Hello\n".getBytes());
- Thread.sleep(150);
- fos = refreshFile(fos);
- fos.write("there\n".getBytes());
- Thread.sleep(150);
- fos = refreshFile(fos);
- fos.write("World\n".getBytes());
+ fos.write("Hello\nWorld\n".getBytes());
Thread.sleep(150);
+
+ context.startAllRoutes();
+
+ // roll-over file
+ Thread.sleep(1500);
fos = refreshFile(fos);
+ fos.write("Bye\nWorld\n".getBytes());
fos.write("!\n".getBytes());
+ // ensure it does not read the file again
+ Thread.sleep(1500);
+ } finally {
+ fos.close();
+ }
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testScanFileAlreadyWritten() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(4);
+
+ FileOutputStream fos = refreshFile(null);
+ try {
+ fos.write("Hello\nthere\nWorld\n!\n".getBytes());
+ context.startAllRoutes();
+ // ensure it does not read the file again
+ Thread.sleep(1000);
} finally {
fos.close();
}
@@ -102,7 +128,10 @@ public class ScanStreamFileTest extends CamelTestSupport {
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
-
from("stream:file?fileName=target/stream/scanstreamfile.txt&scanStream=true&scanStreamDelay=200&retry=true").to("mock:result");
+
from("stream:file?fileName=target/stream/scanstreamfile.txt&scanStream=true&scanStreamDelay=200&retry=true&fileWatcher=true")
+ .routeId("foo").noAutoStartup()
+ .to("log:line")
+ .to("mock:result");
}
};
}
--
To stop receiving notification emails like this one, please contact
[email protected].
