You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2015/03/09 15:11:53 UTC
[1/3] incubator-nifi git commit: NIFI-380 fixed unit test to use a
more os portable command. Modify execute process to enable error stream
redirection.
Repository: incubator-nifi
Updated Branches:
refs/heads/develop 342ca1791 -> 3533a4a58
NIFI-380 fixed unit test to use a more os portable command. Modify execute process to enable error stream redirection.
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/3533a4a5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/3533a4a5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/3533a4a5
Branch: refs/heads/develop
Commit: 3533a4a58ee2533b32c7f6cc6f758a4a5be93da4
Parents: a066d9b
Author: joewitt <jo...@apache.org>
Authored: Thu Mar 5 22:56:41 2015 -0500
Committer: joewitt <jo...@apache.org>
Committed: Mon Mar 9 10:01:57 2015 -0400
----------------------------------------------------------------------
.../processors/standard/ExecuteProcess.java | 272 ++++++++++---------
.../index.html | 42 +--
.../processors/standard/TestExecuteProcess.java | 6 +-
3 files changed, 168 insertions(+), 152 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3533a4a5/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
index c510641..31efd18 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
@@ -38,7 +38,6 @@ import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -58,7 +57,6 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-
@Tags({"command", "process", "source", "external", "invoke", "script"})
@CapabilityDescription("Runs an operating system command specified by the user and writes the output of that command to a FlowFile. If the command is expected "
+ "to be long-running, the Processor can output the partial data on a specified interval. When this option is used, the output is expected to be in textual "
@@ -66,94 +64,105 @@ import org.apache.nifi.processor.util.StandardValidators;
public class ExecuteProcess extends AbstractProcessor {
public static final PropertyDescriptor COMMAND = new PropertyDescriptor.Builder()
- .name("Command")
- .description("Specifies the command to be executed; if just the name of an executable is provided, it must be in the user's environment PATH.")
- .required(true)
- .expressionLanguageSupported(false)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
+ .name("Command")
+ .description("Specifies the command to be executed; if just the name of an executable is provided, it must be in the user's environment PATH.")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
public static final PropertyDescriptor COMMAND_ARGUMENTS = new PropertyDescriptor.Builder()
- .name("Command Arguments")
- .description("The arguments to supply to the executable delimited by white space. White space can be escaped by enclosing it in double-quotes.")
- .required(false)
- .expressionLanguageSupported(false)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
+ .name("Command Arguments")
+ .description("The arguments to supply to the executable delimited by white space. White space can be escaped by enclosing it in double-quotes.")
+ .required(false)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
public static final PropertyDescriptor WORKING_DIR = new PropertyDescriptor.Builder()
- .name("Working Directory")
- .description("The directory to use as the current working directory when executing the command")
- .expressionLanguageSupported(false)
- .addValidator(StandardValidators.createDirectoryExistsValidator(false, true))
- .required(false)
- .build();
-
+ .name("Working Directory")
+ .description("The directory to use as the current working directory when executing the command")
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.createDirectoryExistsValidator(false, true))
+ .required(false)
+ .build();
+
public static final PropertyDescriptor BATCH_DURATION = new PropertyDescriptor.Builder()
- .name("Batch Duration")
- .description("If the process is expected to be long-running and produce textual output, a batch duration can be specified so "
- + "that the output will be captured for this amount of time and a FlowFile will then be sent out with the results "
- + "and a new FlowFile will be started, rather than waiting for the process to finish before sending out the results")
- .required(false)
- .expressionLanguageSupported(false)
- .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
- .build();
-
+ .name("Batch Duration")
+ .description("If the process is expected to be long-running and produce textual output, a batch duration can be specified so "
+ + "that the output will be captured for this amount of time and a FlowFile will then be sent out with the results "
+ + "and a new FlowFile will be started, rather than waiting for the process to finish before sending out the results")
+ .required(false)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor REDIRECT_ERROR_STREAM = new PropertyDescriptor.Builder()
+ .name("Redirect Error Stream")
+ .description("If true will redirect any error stream output of the process to the output stream. "
+ + "This is particularly helpful for processes which write extensively to the error stream or for troubleshooting.")
+ .required(false)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .build();
+
public static final Relationship REL_SUCCESS = new Relationship.Builder()
- .name("success")
- .description("All created FlowFiles are routed to this relationship")
- .build();
-
+ .name("success")
+ .description("All created FlowFiles are routed to this relationship")
+ .build();
private volatile ExecutorService executor;
-
+
@Override
public Set<Relationship> getRelationships() {
return Collections.singleton(REL_SUCCESS);
}
-
+
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> properties = new ArrayList<>();
properties.add(COMMAND);
properties.add(COMMAND_ARGUMENTS);
properties.add(BATCH_DURATION);
+ properties.add(REDIRECT_ERROR_STREAM);
return properties;
}
-
+
@Override
protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
return new PropertyDescriptor.Builder()
- .name(propertyDescriptorName)
- .description("Sets the environment variable '" + propertyDescriptorName + "' for the process' environment")
- .dynamic(true)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
+ .name(propertyDescriptorName)
+ .description("Sets the environment variable '" + propertyDescriptorName + "' for the process' environment")
+ .dynamic(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
}
-
+
static List<String> splitArgs(final String input) {
- if ( input == null ) {
+ if (input == null) {
return Collections.emptyList();
}
-
+
final List<String> args = new ArrayList<>();
-
+
final String trimmed = input.trim();
boolean inQuotes = false;
-
+
final StringBuilder sb = new StringBuilder();
- for (int i=0; i < trimmed.length(); i++) {
+ for (int i = 0; i < trimmed.length(); i++) {
final char c = trimmed.charAt(i);
switch (c) {
case ' ':
case '\t':
case '\r':
case '\n': {
- if ( inQuotes ) {
+ if (inQuotes) {
sb.append(c);
} else {
final String arg = sb.toString().trim();
- if ( !arg.isEmpty() ) {
+ if (!arg.isEmpty()) {
args.add(arg);
}
sb.setLength(0);
@@ -168,20 +177,20 @@ public class ExecuteProcess extends AbstractProcessor {
break;
}
}
-
+
final String finalArg = sb.toString().trim();
- if ( !finalArg.isEmpty() ) {
+ if (!finalArg.isEmpty()) {
args.add(finalArg);
}
-
+
return args;
}
-
+
@OnScheduled
public void setupExecutor(final ProcessContext context) {
executor = Executors.newFixedThreadPool(context.getMaxConcurrentTasks() * 2, new ThreadFactory() {
private final ThreadFactory defaultFactory = Executors.defaultThreadFactory();
-
+
@Override
public Thread newThread(final Runnable r) {
final Thread t = defaultFactory.newThread(r);
@@ -190,73 +199,67 @@ public class ExecuteProcess extends AbstractProcessor {
}
});
}
-
+
@OnUnscheduled
public void shutdownExecutor() {
executor.shutdown();
}
-
+
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final String command = context.getProperty(COMMAND).getValue();
final List<String> args = splitArgs(context.getProperty(COMMAND_ARGUMENTS).getValue());
-
+ final Boolean redirectErrorStream = context.getProperty(REDIRECT_ERROR_STREAM).asBoolean();
+
final List<String> commandStrings = new ArrayList<>(args.size() + 1);
commandStrings.add(command);
commandStrings.addAll(args);
final String commandString = StringUtils.join(commandStrings, " ");
-
+
final ProcessBuilder builder = new ProcessBuilder(commandStrings);
final String workingDirName = context.getProperty(WORKING_DIR).getValue();
- if ( workingDirName != null ) {
+ if (workingDirName != null) {
builder.directory(new File(workingDirName));
}
-
+
final Map<String, String> environment = new HashMap<>();
- for ( final Map.Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet() ) {
- if ( entry.getKey().isDynamic() ) {
+ for (final Map.Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet()) {
+ if (entry.getKey().isDynamic()) {
environment.put(entry.getKey().getName(), entry.getValue());
}
}
-
- if ( !environment.isEmpty() ) {
+
+ if (!environment.isEmpty()) {
builder.environment().putAll(environment);
}
-
+
final long startNanos = System.nanoTime();
final Process process;
try {
- process = builder.start();
+ process = builder.redirectErrorStream(redirectErrorStream).start();
} catch (final IOException ioe) {
- getLogger().error("Failed to create process due to {}", new Object[] {ioe});
+ getLogger().error("Failed to create process due to {}", new Object[]{ioe});
context.yield();
return;
}
-
+
final Long batchNanos = context.getProperty(BATCH_DURATION).asTimePeriod(TimeUnit.NANOSECONDS);
-
+
// Submit task to read error stream from process
- final AtomicReference<String> errorStream = new AtomicReference<>();
- executor.submit(new Runnable() {
- @Override
- public void run() {
- final StringBuilder sb = new StringBuilder();
- try (final BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
- String line;
- while ((line = reader.readLine()) != null) {
- if ( sb.length() < 4000 ) {
- sb.append(line);
- sb.append("\n");
+ if (!redirectErrorStream) {
+ executor.submit(new Runnable() {
+ @Override
+ public void run() {
+ try (final BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+ while (reader.read() >= 0) {
}
+ } catch (final IOException ioe) {
}
- } catch (final IOException ioe) {
}
-
- errorStream.set(sb.toString());
- }
- });
-
+ });
+ }
+
// Submit task to read output of Process and write to FlowFile.
final ProxyOutputStream proxyOut = new ProxyOutputStream(getLogger());
final AtomicBoolean failure = new AtomicBoolean(false);
@@ -265,16 +268,16 @@ public class ExecuteProcess extends AbstractProcessor {
@Override
public Object call() throws IOException {
try {
- if ( batchNanos == null ) {
+ if (batchNanos == null) {
// if we aren't batching, just copy the stream from the process to the flowfile.
try (final BufferedInputStream bufferedIn = new BufferedInputStream(process.getInputStream())) {
final byte[] buffer = new byte[4096];
int len;
while ((len = bufferedIn.read(buffer)) > 0) {
- if ( !isScheduled() ) {
+ if (!isScheduled()) {
return null;
}
-
+
proxyOut.write(buffer, 0, len);
}
}
@@ -286,9 +289,9 @@ public class ExecuteProcess extends AbstractProcessor {
// to read lines of text and write them as lines of text.
try (final BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
String line;
-
+
while ((line = reader.readLine()) != null) {
- if ( !isScheduled() ) {
+ if (!isScheduled()) {
return null;
}
@@ -302,22 +305,22 @@ public class ExecuteProcess extends AbstractProcessor {
} finally {
finishedCopying.set(true);
}
-
+
return null;
}
});
-
+
// continue to do this loop until both the process has finished and we have finished copying
// the output from the process to the FlowFile. Unfortunately, even after calling Process.exitValue(),
// there can be data buffered on the InputStream; so we will wait until the stream is empty as well.
int flowFileCount = 0;
while (!finishedCopying.get() || isAlive(process)) {
- if ( !isScheduled() ) {
+ if (!isScheduled()) {
getLogger().info("User stopped processor; will terminate process immediately");
process.destroy();
break;
}
-
+
// Create a FlowFile that we can write to and set the OutputStream for the FlowFile
// as the delegate for the ProxyOuptutStream, then wait until the process finishes
// or until the specified amount of time
@@ -327,31 +330,33 @@ public class ExecuteProcess extends AbstractProcessor {
public void process(final OutputStream flowFileOut) throws IOException {
try (final OutputStream out = new BufferedOutputStream(flowFileOut)) {
proxyOut.setDelegate(out);
-
- if ( batchNanos == null ) {
+
+ if (batchNanos == null) {
// we are not creating batches; wait until process terminates.
Integer exitCode = null;
while (exitCode == null) {
try {
exitCode = process.waitFor();
- } catch (final InterruptedException ie) {}
+ } catch (final InterruptedException ie) {
+ }
}
} else {
// wait the allotted amount of time.
try {
TimeUnit.NANOSECONDS.sleep(batchNanos);
- } catch (final InterruptedException ie) {}
+ } catch (final InterruptedException ie) {
+ }
}
-
+
proxyOut.setDelegate(null); // prevent from writing to this stream
}
}
});
-
- if ( flowFile.getSize() == 0L ) {
+
+ if (flowFile.getSize() == 0L) {
// If no data was written to the file, remove it
session.remove(flowFile);
- } else if ( failure.get() ) {
+ } else if (failure.get()) {
// If there was a failure processing the output of the Process, remove the FlowFile
session.remove(flowFile);
getLogger().error("Failed to read data from Process, so will not generate FlowFile");
@@ -359,15 +364,15 @@ public class ExecuteProcess extends AbstractProcessor {
} else {
// All was good. Generate event and transfer FlowFile.
session.getProvenanceReporter().create(flowFile, "Created from command: " + commandString);
- getLogger().info("Created {} and routed to success", new Object[] {flowFile});
+ getLogger().info("Created {} and routed to success", new Object[]{flowFile});
session.transfer(flowFile, REL_SUCCESS);
flowFileCount++;
}
-
+
// Commit the session so that the FlowFile is transferred to the next processor
session.commit();
}
-
+
final int exitCode;
final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
try {
@@ -376,20 +381,19 @@ public class ExecuteProcess extends AbstractProcessor {
getLogger().warn("Process was interrupted before finishing");
return;
}
-
+
try {
future.get();
} catch (final ExecutionException e) {
- getLogger().error("Failed to copy output from Process to FlowFile due to {}", new Object[] {e.getCause()});
+ getLogger().error("Failed to copy output from Process to FlowFile due to {}", new Object[]{e.getCause()});
} catch (final InterruptedException ie) {
getLogger().error("Interrupted while waiting to copy data form Process to FlowFile");
return;
}
-
- getLogger().info("Process finished with exit code {} after creating {} FlowFiles in {} millis", new Object[] {exitCode, flowFileCount, millis});
+
+ getLogger().info("Process finished with exit code {} after creating {} FlowFiles in {} millis", new Object[]{exitCode, flowFileCount, millis});
}
-
private boolean isAlive(final Process process) {
// unfortunately, java provides no straight-forward way to test if a Process is alive.
// In Java 8, Process.isAlive() is introduced, but NiFi needs to run against Java 7,
@@ -401,46 +405,48 @@ public class ExecuteProcess extends AbstractProcessor {
return true;
}
}
-
-
+
/**
- * Output stream that is used to wrap another output stream in a way that the
- * underlying output stream can be swapped out for a different one when needed
+ * Output stream that is used to wrap another output stream in a way that
+ * the underlying output stream can be swapped out for a different one when
+ * needed
*/
private static class ProxyOutputStream extends OutputStream {
+
private final ProcessorLog logger;
-
+
private final Lock lock = new ReentrantLock();
private OutputStream delegate;
public ProxyOutputStream(final ProcessorLog logger) {
this.logger = logger;
}
-
+
public void setDelegate(final OutputStream delegate) {
lock.lock();
try {
- logger.trace("Switching delegate from {} to {}", new Object[] {this.delegate, delegate});
+ logger.trace("Switching delegate from {} to {}", new Object[]{this.delegate, delegate});
this.delegate = delegate;
} finally {
lock.unlock();
}
}
-
+
private void sleep(final long millis) {
try {
Thread.sleep(millis);
- } catch (final InterruptedException ie) {}
+ } catch (final InterruptedException ie) {
+ }
}
-
+
@Override
public void write(final int b) throws IOException {
lock.lock();
try {
while (true) {
- if ( delegate != null ) {
- logger.trace("Writing to {}", new Object[] {delegate});
+ if (delegate != null) {
+ logger.trace("Writing to {}", new Object[]{delegate});
delegate.write(b);
return;
@@ -454,15 +460,15 @@ public class ExecuteProcess extends AbstractProcessor {
lock.unlock();
}
}
-
+
@Override
public void write(final byte[] b, final int off, final int len) throws IOException {
lock.lock();
try {
while (true) {
- if ( delegate != null ) {
- logger.trace("Writing to {}", new Object[] {delegate});
-
+ if (delegate != null) {
+ logger.trace("Writing to {}", new Object[]{delegate});
+
delegate.write(b, off, len);
return;
} else {
@@ -475,22 +481,22 @@ public class ExecuteProcess extends AbstractProcessor {
lock.unlock();
}
}
-
+
@Override
public void write(final byte[] b) throws IOException {
write(b, 0, b.length);
}
-
+
@Override
public void close() throws IOException {
}
-
+
@Override
public void flush() throws IOException {
lock.lock();
try {
while (true) {
- if ( delegate != null ) {
+ if (delegate != null) {
delegate.flush();
return;
} else {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3533a4a5/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html
index 9129855..3526cd1 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html
@@ -25,11 +25,11 @@
<!-- Processor Documentation ================================================== -->
<h2> Description:</h2>
<p>
- Runs an operating system command specified by the user and writes the output of that command to a FlowFile. If the command is expected
- to be long-running, the Processor can output the partial data on a specified interval. When this option is used, the output is expected to be in textual
+ Runs an operating system command specified by the user and writes the output of that command to a FlowFile. If the command is expected
+ to be long-running, the Processor can output the partial data on a specified interval. When this option is used, the output is expected to be in textual
format, as it typically does not make sense to split binary data on arbitrary time-based intervals.
</p>
-
+
<p>
<strong>Properties:</strong>
</p>
@@ -60,20 +60,30 @@
<li>Supports expression language: false</li>
</ul>
</li>
- <li>
- Batch Duration>
- <ul>
- <li>
- If the process is expected to be long-running and produce textual output, a batch duration can be specified so
- that the output will be captured for this amount of time and a FlowFile will then be sent out with the results
- and a new FlowFile will be started, rather than waiting for the process to finish before sending out the results.
- If no value is provided, the process will run to completion and the entire output of the process will be written
- to a single FlowFile.
- </li>
- <li>Default value: none</li>
+ <li>Batch Duration
+ <ul>
+ <li>
+ If the process is expected to be long-running and produce textual output, a batch duration can be specified so
+ that the output will be captured for this amount of time and a FlowFile will then be sent out with the results
+ and a new FlowFile will be started, rather than waiting for the process to finish before sending out the results.
+ If no value is provided, the process will run to completion and the entire output of the process will be written
+ to a single FlowFile.
+ </li>
+ <li>Default value: none</li>
<li>Supports expression language: false</li>
- </ul>
- </li>
+ </ul>
+ </li>
+ <li>Redirect Error Stream
+ <ul>
+ <li>
+ If true will redirect any error stream output of the process to the output stream.
+ This is particularly helpful for processes which write extensively to the error stream or for troubleshooting.
+ </li>
+ <li>Default value: false</li>
+ <li>Allowed Values: true, false</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
</ul>
<p>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3533a4a5/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
index 0aa2ee3..897973c 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
@@ -59,12 +59,12 @@ public class TestExecuteProcess {
}
@Test
- public void testPing() {
+ public void testEcho() {
System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi", "TRACE");
final TestRunner runner = TestRunners.newTestRunner(ExecuteProcess.class);
- runner.setProperty(ExecuteProcess.COMMAND, "ping");
- runner.setProperty(ExecuteProcess.COMMAND_ARGUMENTS, "127.0.0.1");
+ runner.setProperty(ExecuteProcess.COMMAND, "echo");
+ runner.setProperty(ExecuteProcess.COMMAND_ARGUMENTS, "test-args");
runner.setProperty(ExecuteProcess.BATCH_DURATION, "500 millis");
runner.run();
[2/3] incubator-nifi git commit: NIFI-380: Added documentation
Posted by jo...@apache.org.
NIFI-380: Added documentation
Signed-off-by: joewitt <jo...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/a066d9b6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/a066d9b6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/a066d9b6
Branch: refs/heads/develop
Commit: a066d9b601ae1b8910bd7e526d1c631e4e4e3aae
Parents: 12b44ee
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Mar 4 20:09:08 2015 -0500
Committer: joewitt <jo...@apache.org>
Committed: Mon Mar 9 10:01:57 2015 -0400
----------------------------------------------------------------------
.../processors/standard/ExecuteProcess.java | 27 ++++--
.../index.html | 90 ++++++++++++++++++++
2 files changed, 110 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/a066d9b6/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
index 6f68f7d..c510641 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
@@ -57,7 +57,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.stream.io.StreamUtils;
@Tags({"command", "process", "source", "external", "invoke", "script"})
@@ -66,13 +65,8 @@ import org.apache.nifi.stream.io.StreamUtils;
+ "format, as it typically does not make sense to split binary data on arbitrary time-based intervals.")
public class ExecuteProcess extends AbstractProcessor {
- public static final Relationship REL_SUCCESS = new Relationship.Builder()
- .name("success")
- .description("All created FlowFiles are routed to this relationship")
- .build();
-
public static final PropertyDescriptor COMMAND = new PropertyDescriptor.Builder()
- .name("Command Path")
+ .name("Command")
.description("Specifies the command to be executed; if just the name of an executable is provided, it must be in the user's environment PATH.")
.required(true)
.expressionLanguageSupported(false)
@@ -105,6 +99,12 @@ public class ExecuteProcess extends AbstractProcessor {
.addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
.build();
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("All created FlowFiles are routed to this relationship")
+ .build();
+
+
private volatile ExecutorService executor;
@Override
@@ -317,6 +317,10 @@ public class ExecuteProcess extends AbstractProcessor {
process.destroy();
break;
}
+
+ // Create a FlowFile that we can write to and set the OutputStream for the FlowFile
+ // as the delegate for the ProxyOuptutStream, then wait until the process finishes
+ // or until the specified amount of time
FlowFile flowFile = session.create();
flowFile = session.write(flowFile, new OutputStreamCallback() {
@Override
@@ -345,17 +349,22 @@ public class ExecuteProcess extends AbstractProcessor {
});
if ( flowFile.getSize() == 0L ) {
+ // If no data was written to the file, remove it
session.remove(flowFile);
} else if ( failure.get() ) {
+ // If there was a failure processing the output of the Process, remove the FlowFile
session.remove(flowFile);
getLogger().error("Failed to read data from Process, so will not generate FlowFile");
+ break;
} else {
+ // All was good. Generate event and transfer FlowFile.
session.getProvenanceReporter().create(flowFile, "Created from command: " + commandString);
getLogger().info("Created {} and routed to success", new Object[] {flowFile});
session.transfer(flowFile, REL_SUCCESS);
flowFileCount++;
}
+ // Commit the session so that the FlowFile is transferred to the next processor
session.commit();
}
@@ -394,6 +403,10 @@ public class ExecuteProcess extends AbstractProcessor {
}
+ /**
+ * Output stream that is used to wrap another output stream in a way that the
+ * underlying output stream can be swapped out for a different one when needed
+ */
private static class ProxyOutputStream extends OutputStream {
private final ProcessorLog logger;
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/a066d9b6/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html
new file mode 100644
index 0000000..9129855
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.ExecuteProcess/index.html
@@ -0,0 +1,90 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>ExecuteProcess</title>
+
+ <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <!-- Processor Documentation ================================================== -->
+ <h2> Description:</h2>
+ <p>
+ Runs an operating system command specified by the user and writes the output of that command to a FlowFile. If the command is expected
+ to be long-running, the Processor can output the partial data on a specified interval. When this option is used, the output is expected to be in textual
+ format, as it typically does not make sense to split binary data on arbitrary time-based intervals.
+ </p>
+
+ <p>
+ <strong>Properties:</strong>
+ </p>
+ <p>In the list below, the names of required properties appear
+ in bold. Any other properties (not in bold) are considered optional.
+ If a property has a default value, it is indicated. If a property
+ supports the use of the NiFi Expression Language (or simply,
+ "expression language"), that is also indicated.</p>
+ <ul>
+ <li><strong>Command</strong>
+ <ul>
+ <li>Specifies the command to be executed; if just the name of an executable is provided, it must be in the user's environment PATH.</li>
+ <li>Default value: none</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li>Command Arguments
+ <ul>
+ <li>The arguments to supply to the executable delimited by white space. White space can be escaped by enclosing it in double-quotes.</li>
+ <li>Default value: none</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li>Working Directory
+ <ul>
+ <li>The directory to use as the current working directory when executing the command</li>
+ <li>Default value: none (which means whatever NiFi's root installation directory is)</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li>
+ Batch Duration>
+ <ul>
+ <li>
+ If the process is expected to be long-running and produce textual output, a batch duration can be specified so
+ that the output will be captured for this amount of time and a FlowFile will then be sent out with the results
+ and a new FlowFile will be started, rather than waiting for the process to finish before sending out the results.
+ If no value is provided, the process will run to completion and the entire output of the process will be written
+ to a single FlowFile.
+ </li>
+ <li>Default value: none</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ </ul>
+
+ <p>
+ <strong>Relationships:</strong>
+ </p>
+ <ul>
+ <li>success
+ <ul>
+ <li>All FlowFiles that are created are routed to this relationship.</li>
+ </ul>
+ </li>
+ </ul>
+ </body>
+</html>
[3/3] incubator-nifi git commit: NIFI-380: Initial import of
ExecuteProcess processor
Posted by jo...@apache.org.
NIFI-380: Initial import of ExecuteProcess processor
Signed-off-by: joewitt <jo...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/12b44ee0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/12b44ee0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/12b44ee0
Branch: refs/heads/develop
Commit: 12b44ee0b8446346794a25da5b0ab7d5329287a1
Parents: 342ca17
Author: Mark Payne <ma...@hotmail.com>
Authored: Tue Feb 24 21:30:18 2015 -0500
Committer: joewitt <jo...@apache.org>
Committed: Mon Mar 9 10:01:57 2015 -0400
----------------------------------------------------------------------
.../processors/standard/ExecuteProcess.java | 494 +++++++++++++++++++
.../org.apache.nifi.processor.Processor | 1 +
.../processors/standard/TestExecuteProcess.java | 78 +++
3 files changed, 573 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/12b44ee0/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
new file mode 100644
index 0000000..6f68f7d
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteProcess.java
@@ -0,0 +1,494 @@
+/*
+ * 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.nifi.processors.standard;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.stream.io.StreamUtils;
+
+
+@Tags({"command", "process", "source", "external", "invoke", "script"})
+@CapabilityDescription("Runs an operating system command specified by the user and writes the output of that command to a FlowFile. If the command is expected "
+ + "to be long-running, the Processor can output the partial data on a specified interval. When this option is used, the output is expected to be in textual "
+ + "format, as it typically does not make sense to split binary data on arbitrary time-based intervals.")
+public class ExecuteProcess extends AbstractProcessor {
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("All created FlowFiles are routed to this relationship")
+ .build();
+
+ public static final PropertyDescriptor COMMAND = new PropertyDescriptor.Builder()
+ .name("Command Path")
+ .description("Specifies the command to be executed; if just the name of an executable is provided, it must be in the user's environment PATH.")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor COMMAND_ARGUMENTS = new PropertyDescriptor.Builder()
+ .name("Command Arguments")
+ .description("The arguments to supply to the executable delimited by white space. White space can be escaped by enclosing it in double-quotes.")
+ .required(false)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor WORKING_DIR = new PropertyDescriptor.Builder()
+ .name("Working Directory")
+ .description("The directory to use as the current working directory when executing the command")
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.createDirectoryExistsValidator(false, true))
+ .required(false)
+ .build();
+
+ public static final PropertyDescriptor BATCH_DURATION = new PropertyDescriptor.Builder()
+ .name("Batch Duration")
+ .description("If the process is expected to be long-running and produce textual output, a batch duration can be specified so "
+ + "that the output will be captured for this amount of time and a FlowFile will then be sent out with the results "
+ + "and a new FlowFile will be started, rather than waiting for the process to finish before sending out the results")
+ .required(false)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .build();
+
+ private volatile ExecutorService executor;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return Collections.singleton(REL_SUCCESS);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new ArrayList<>();
+ properties.add(COMMAND);
+ properties.add(COMMAND_ARGUMENTS);
+ properties.add(BATCH_DURATION);
+ return properties;
+ }
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .description("Sets the environment variable '" + propertyDescriptorName + "' for the process' environment")
+ .dynamic(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ }
+
+ static List<String> splitArgs(final String input) {
+ if ( input == null ) {
+ return Collections.emptyList();
+ }
+
+ final List<String> args = new ArrayList<>();
+
+ final String trimmed = input.trim();
+ boolean inQuotes = false;
+
+ final StringBuilder sb = new StringBuilder();
+ for (int i=0; i < trimmed.length(); i++) {
+ final char c = trimmed.charAt(i);
+ switch (c) {
+ case ' ':
+ case '\t':
+ case '\r':
+ case '\n': {
+ if ( inQuotes ) {
+ sb.append(c);
+ } else {
+ final String arg = sb.toString().trim();
+ if ( !arg.isEmpty() ) {
+ args.add(arg);
+ }
+ sb.setLength(0);
+ }
+ break;
+ }
+ case '"':
+ inQuotes = !inQuotes;
+ break;
+ default:
+ sb.append(c);
+ break;
+ }
+ }
+
+ final String finalArg = sb.toString().trim();
+ if ( !finalArg.isEmpty() ) {
+ args.add(finalArg);
+ }
+
+ return args;
+ }
+
+ @OnScheduled
+ public void setupExecutor(final ProcessContext context) {
+ executor = Executors.newFixedThreadPool(context.getMaxConcurrentTasks() * 2, new ThreadFactory() {
+ private final ThreadFactory defaultFactory = Executors.defaultThreadFactory();
+
+ @Override
+ public Thread newThread(final Runnable r) {
+ final Thread t = defaultFactory.newThread(r);
+ t.setName("ExecuteProcess " + getIdentifier() + " Task");
+ return t;
+ }
+ });
+ }
+
+ @OnUnscheduled
+ public void shutdownExecutor() {
+ executor.shutdown();
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ final String command = context.getProperty(COMMAND).getValue();
+ final List<String> args = splitArgs(context.getProperty(COMMAND_ARGUMENTS).getValue());
+
+ final List<String> commandStrings = new ArrayList<>(args.size() + 1);
+ commandStrings.add(command);
+ commandStrings.addAll(args);
+
+ final String commandString = StringUtils.join(commandStrings, " ");
+
+ final ProcessBuilder builder = new ProcessBuilder(commandStrings);
+ final String workingDirName = context.getProperty(WORKING_DIR).getValue();
+ if ( workingDirName != null ) {
+ builder.directory(new File(workingDirName));
+ }
+
+ final Map<String, String> environment = new HashMap<>();
+ for ( final Map.Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet() ) {
+ if ( entry.getKey().isDynamic() ) {
+ environment.put(entry.getKey().getName(), entry.getValue());
+ }
+ }
+
+ if ( !environment.isEmpty() ) {
+ builder.environment().putAll(environment);
+ }
+
+ final long startNanos = System.nanoTime();
+ final Process process;
+ try {
+ process = builder.start();
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to create process due to {}", new Object[] {ioe});
+ context.yield();
+ return;
+ }
+
+ final Long batchNanos = context.getProperty(BATCH_DURATION).asTimePeriod(TimeUnit.NANOSECONDS);
+
+ // Submit task to read error stream from process
+ final AtomicReference<String> errorStream = new AtomicReference<>();
+ executor.submit(new Runnable() {
+ @Override
+ public void run() {
+ final StringBuilder sb = new StringBuilder();
+ try (final BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ if ( sb.length() < 4000 ) {
+ sb.append(line);
+ sb.append("\n");
+ }
+ }
+ } catch (final IOException ioe) {
+ }
+
+ errorStream.set(sb.toString());
+ }
+ });
+
+ // Submit task to read output of Process and write to FlowFile.
+ final ProxyOutputStream proxyOut = new ProxyOutputStream(getLogger());
+ final AtomicBoolean failure = new AtomicBoolean(false);
+ final AtomicBoolean finishedCopying = new AtomicBoolean(false);
+ final Future<?> future = executor.submit(new Callable<Object>() {
+ @Override
+ public Object call() throws IOException {
+ try {
+ if ( batchNanos == null ) {
+ // if we aren't batching, just copy the stream from the process to the flowfile.
+ try (final BufferedInputStream bufferedIn = new BufferedInputStream(process.getInputStream())) {
+ final byte[] buffer = new byte[4096];
+ int len;
+ while ((len = bufferedIn.read(buffer)) > 0) {
+ if ( !isScheduled() ) {
+ return null;
+ }
+
+ proxyOut.write(buffer, 0, len);
+ }
+ }
+ } else {
+ // we are batching, which means that the output of the process is text. It doesn't make sense to grab
+ // arbitrary batches of bytes from some process and send it along as a piece of data, so we assume that
+ // setting a batch during means text.
+ // Also, we don't want that text to get split up in the middle of a line, so we use BufferedReader
+ // to read lines of text and write them as lines of text.
+ try (final BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
+ String line;
+
+ while ((line = reader.readLine()) != null) {
+ if ( !isScheduled() ) {
+ return null;
+ }
+
+ proxyOut.write((line + "\n").getBytes(StandardCharsets.UTF_8));
+ }
+ }
+ }
+ } catch (final IOException ioe) {
+ failure.set(true);
+ throw ioe;
+ } finally {
+ finishedCopying.set(true);
+ }
+
+ return null;
+ }
+ });
+
+ // continue to do this loop until both the process has finished and we have finished copying
+ // the output from the process to the FlowFile. Unfortunately, even after calling Process.exitValue(),
+ // there can be data buffered on the InputStream; so we will wait until the stream is empty as well.
+ int flowFileCount = 0;
+ while (!finishedCopying.get() || isAlive(process)) {
+ if ( !isScheduled() ) {
+ getLogger().info("User stopped processor; will terminate process immediately");
+ process.destroy();
+ break;
+ }
+ FlowFile flowFile = session.create();
+ flowFile = session.write(flowFile, new OutputStreamCallback() {
+ @Override
+ public void process(final OutputStream flowFileOut) throws IOException {
+ try (final OutputStream out = new BufferedOutputStream(flowFileOut)) {
+ proxyOut.setDelegate(out);
+
+ if ( batchNanos == null ) {
+ // we are not creating batches; wait until process terminates.
+ Integer exitCode = null;
+ while (exitCode == null) {
+ try {
+ exitCode = process.waitFor();
+ } catch (final InterruptedException ie) {}
+ }
+ } else {
+ // wait the allotted amount of time.
+ try {
+ TimeUnit.NANOSECONDS.sleep(batchNanos);
+ } catch (final InterruptedException ie) {}
+ }
+
+ proxyOut.setDelegate(null); // prevent from writing to this stream
+ }
+ }
+ });
+
+ if ( flowFile.getSize() == 0L ) {
+ session.remove(flowFile);
+ } else if ( failure.get() ) {
+ session.remove(flowFile);
+ getLogger().error("Failed to read data from Process, so will not generate FlowFile");
+ } else {
+ session.getProvenanceReporter().create(flowFile, "Created from command: " + commandString);
+ getLogger().info("Created {} and routed to success", new Object[] {flowFile});
+ session.transfer(flowFile, REL_SUCCESS);
+ flowFileCount++;
+ }
+
+ session.commit();
+ }
+
+ final int exitCode;
+ final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ try {
+ exitCode = process.waitFor();
+ } catch (final InterruptedException ie) {
+ getLogger().warn("Process was interrupted before finishing");
+ return;
+ }
+
+ try {
+ future.get();
+ } catch (final ExecutionException e) {
+ getLogger().error("Failed to copy output from Process to FlowFile due to {}", new Object[] {e.getCause()});
+ } catch (final InterruptedException ie) {
+ getLogger().error("Interrupted while waiting to copy data form Process to FlowFile");
+ return;
+ }
+
+ getLogger().info("Process finished with exit code {} after creating {} FlowFiles in {} millis", new Object[] {exitCode, flowFileCount, millis});
+ }
+
+
+ private boolean isAlive(final Process process) {
+ // unfortunately, java provides no straight-forward way to test if a Process is alive.
+ // In Java 8, Process.isAlive() is introduced, but NiFi needs to run against Java 7,
+ // so we have this solution in the mean time.
+ try {
+ process.exitValue();
+ return false;
+ } catch (final IllegalThreadStateException itse) {
+ return true;
+ }
+ }
+
+
+ private static class ProxyOutputStream extends OutputStream {
+ private final ProcessorLog logger;
+
+ private final Lock lock = new ReentrantLock();
+ private OutputStream delegate;
+
+ public ProxyOutputStream(final ProcessorLog logger) {
+ this.logger = logger;
+ }
+
+ public void setDelegate(final OutputStream delegate) {
+ lock.lock();
+ try {
+ logger.trace("Switching delegate from {} to {}", new Object[] {this.delegate, delegate});
+
+ this.delegate = delegate;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void sleep(final long millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (final InterruptedException ie) {}
+ }
+
+ @Override
+ public void write(final int b) throws IOException {
+ lock.lock();
+ try {
+ while (true) {
+ if ( delegate != null ) {
+ logger.trace("Writing to {}", new Object[] {delegate});
+
+ delegate.write(b);
+ return;
+ } else {
+ lock.unlock();
+ sleep(1L);
+ lock.lock();
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void write(final byte[] b, final int off, final int len) throws IOException {
+ lock.lock();
+ try {
+ while (true) {
+ if ( delegate != null ) {
+ logger.trace("Writing to {}", new Object[] {delegate});
+
+ delegate.write(b, off, len);
+ return;
+ } else {
+ lock.unlock();
+ sleep(1L);
+ lock.lock();
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void write(final byte[] b) throws IOException {
+ write(b, 0, b.length);
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+
+ @Override
+ public void flush() throws IOException {
+ lock.lock();
+ try {
+ while (true) {
+ if ( delegate != null ) {
+ delegate.flush();
+ return;
+ } else {
+ lock.unlock();
+ sleep(1L);
+ lock.lock();
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/12b44ee0/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 2d5855c..f81ccec 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -24,6 +24,7 @@ org.apache.nifi.processors.standard.EvaluateRegularExpression
org.apache.nifi.processors.standard.EvaluateXPath
org.apache.nifi.processors.standard.EvaluateXQuery
org.apache.nifi.processors.standard.ExecuteStreamCommand
+org.apache.nifi.processors.standard.ExecuteProcess
org.apache.nifi.processors.standard.GenerateFlowFile
org.apache.nifi.processors.standard.GetFile
org.apache.nifi.processors.standard.GetFTP
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/12b44ee0/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
new file mode 100644
index 0000000..0aa2ee3
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
@@ -0,0 +1,78 @@
+/*
+ * 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.nifi.processors.standard;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+
+public class TestExecuteProcess {
+
+ @Test
+ public void testSplitArgs() {
+ final List<String> nullArgs = ExecuteProcess.splitArgs(null);
+ assertNotNull(nullArgs);
+ assertTrue(nullArgs.isEmpty());
+
+ final List<String> zeroArgs = ExecuteProcess.splitArgs(" ");
+ assertNotNull(zeroArgs);
+ assertTrue(zeroArgs.isEmpty());
+
+ final List<String> singleArg = ExecuteProcess.splitArgs(" hello ");
+ assertEquals(1, singleArg.size());
+ assertEquals("hello", singleArg.get(0));
+
+ final List<String> twoArg = ExecuteProcess.splitArgs(" hello good-bye ");
+ assertEquals(2, twoArg.size());
+ assertEquals("hello", twoArg.get(0));
+ assertEquals("good-bye", twoArg.get(1));
+
+ final List<String> singleQuotedArg = ExecuteProcess.splitArgs(" \"hello\" ");
+ assertEquals(1, singleQuotedArg.size());
+ assertEquals("hello", singleQuotedArg.get(0));
+
+ final List<String> twoQuotedArg = ExecuteProcess.splitArgs(" hello \"good bye\"");
+ assertEquals(2, twoQuotedArg.size());
+ assertEquals("hello", twoQuotedArg.get(0));
+ assertEquals("good bye", twoQuotedArg.get(1));
+ }
+
+ @Test
+ public void testPing() {
+ System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi", "TRACE");
+
+ final TestRunner runner = TestRunners.newTestRunner(ExecuteProcess.class);
+ runner.setProperty(ExecuteProcess.COMMAND, "ping");
+ runner.setProperty(ExecuteProcess.COMMAND_ARGUMENTS, "127.0.0.1");
+ runner.setProperty(ExecuteProcess.BATCH_DURATION, "500 millis");
+
+ runner.run();
+
+ final List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ExecuteProcess.REL_SUCCESS);
+ for ( final MockFlowFile flowFile : flowFiles ) {
+ System.out.println(flowFile);
+ System.out.println(new String(flowFile.toByteArray()));
+ }
+ }
+}