This is an automated email from the ASF dual-hosted git repository.
andrewmlim pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 0d0a119737 NIFI-1954: Add additionalDetails page to ExecuteScript
(#6044)
0d0a119737 is described below
commit 0d0a119737efc951ba2e0aadd21a01b1459f20e5
Author: Matt Burgess <[email protected]>
AuthorDate: Mon Aug 15 11:44:30 2022 -0400
NIFI-1954: Add additionalDetails page to ExecuteScript (#6044)
* NIFI-1954: Add additionalDetails page to ExecuteScript
* NIFI-1954: Incorporated review comments
---
.../additionalDetails.html | 698 +++++++++++++++++++++
1 file changed, 698 insertions(+)
diff --git
a/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ExecuteScript/additionalDetails.html
b/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ExecuteScript/additionalDetails.html
new file mode 100644
index 0000000000..24494ba1d2
--- /dev/null
+++
b/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/resources/docs/org.apache.nifi.processors.script.ExecuteScript/additionalDetails.html
@@ -0,0 +1,698 @@
+<!DOCTYPE html>
+<html lang="en" xmlns="http://www.w3.org/1999/html";>
+<!--
+ 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>ExecuteScript</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css"
type="text/css"/>
+ <style>
+h2 {margin-top: 4em}
+h3 {margin-top: 3em}
+td {text-align: left}
+ </style>
+</head>
+
+<body>
+
+<h1>ExecuteScript</h1>
+
+<h3>Description</h3>
+<p>
+ The ExecuteScript Processor provides the ability to use a scripting
language in order to leverage the NiFi API to perform tasks such as the
following:
+</p>
+<ul>
+ <li>Read content and/or attributes from an incoming FlowFile</li>
+ <li>Create a new FlowFile (with or without a parent)</li>
+ <li>Write content and/or attributes to an outgoing FlowFile</li>
+ <li>Interact with the ProcessSession to transfer FlowFiles to
relationships</li>
+ <li>Read/write to the State Manager to keep track of variables across
executions of the processor</li>
+</ul>
+
+<p>
+ <b>Notes:</b>
+<ul>
+ <li>The engine listed as "python" in the list of available script engines
is actually Jython, not Python. When using Jython, you cannot import pure
(CPython) modules such as pandas</li>
+ <li>Lua does not allow for referencing static members of a class, so the
REL_SUCCESS and REL_FAILURE relationships are made available via script
bindings (aka variables), see the Variable Bindings section for more
details</li>
+ <li>ExecuteScript uses the JSR-223 Script Engine API to evaluate scripts,
so the use of idiomatic language structure is sometimes limited. For example,
in the case of Groovy, there is a separate ExecuteGroovyScript processor that
allows you to do many more idiomatic Groovy tasks. For example, it's easier to
interact with Controller Services via ExecuteGroovyScript vs. ExecuteScript
(see the ExecuteGroovyScript documentation for more details)</li>
+ <li>JavaScript is provided by the Nashorn engine, which is not available
in later Java versions, so it may not show up in the Script Engines list
depending on the JRE used</li>
+</ul>
+</p>
+<h3>Variable Bindings</h3>
+<p>
+ The Processor expects a user defined script that is evaluated when the
processor is triggered. The following variables are available to the scripts:
+</p>
+<table>
+ <tr>
+ <th>Variable Name</th>
+ <th>Description</th>
+ <th>Variable Class</th>
+ </tr>
+ <tr>
+ <td><b>session</b></td>
+ <td>This is a reference to the ProcessSession assigned to the
processor. The session allows you to perform operations on FlowFiles such as
create(), putAttribute(), and transfer(), as well as read() and write()</td>
+ <td><a
href="https://www.javadoc.io/doc/org.apache.nifi/nifi-api/latest/org/apache/nifi/processor/ProcessSession.html";>ProcessSession</a></td>
+ </tr>
+ <tr>
+ <td><b>context</b></td>
+ <td>This is a reference to the ProcessContext for the processor. It
can be used to retrieve processor properties, relationships, Controller
Services, and the State Manager.</td>
+ <td><a
href="https://www.javadoc.io/doc/org.apache.nifi/nifi-api/latest/org/apache/nifi/processor/ProcessContext.html";>ProcessContext</a></td>
+ </tr>
+ <tr>
+ <td><b>log</b></td>
+ <td>This is a reference to the ComponentLog for the processor. Use it
to log messages to NiFi, such as log.info('Hello world!')</td>
+ <td><a
href="https://www.javadoc.io/doc/org.apache.nifi/nifi-api/latest/org/apache/nifi/logging/ComponentLog.html";>ComponentLog</a></td>
+ </tr>
+ <tr>
+ <td><b>REL_SUCCESS</b></td>
+ <td>This is a reference to the "success" relationship defined for the
processor. It could also be inherited by referencing the static member of the
parent class (ExecuteScript), but some engines such as Lua do not allow for
referencing static members, so this is a convenience variable. It also saves
having to use the fully-qualified name for the relationship.</td>
+ <td><a
href="https://www.javadoc.io/doc/org.apache.nifi/nifi-api/latest/org/apache/nifi/processor/Relationship.html";>Relationship</a></td>
+ </tr>
+ <tr>
+ <td><b>REL_FAILURE</b></td>
+ <td>This is a reference to the "failure" relationship defined for the
processor. As with REL_SUCCESS, it could also be inherited by referencing the
static member of the parent class (ExecuteScript), but some engines such as Lua
do not allow for referencing static members, so this is a convenience variable.
It also saves having to use the fully-qualified name for the relationship.</td>
+ <td><a
href="https://www.javadoc.io/doc/org.apache.nifi/nifi-api/latest/org/apache/nifi/processor/Relationship.html";>Relationship</a></td>
+ </tr>
+ <tr>
+ <td><i>Dynamic Properties</i></td>
+ <td>Any dynamic (user-defined) properties defined in ExecuteScript are
passed to the script engine as variables set to the PropertyValue object
corresponding to the dynamic property. This allows you to get the String value
of the property, but also to evaluate the property with respect to NiFi
Expression Language, cast the value as an appropriate data type (e.g.,
Boolean), etc. Because the dynamic property name becomes the variable name for
the script, you must be aware of the va [...]
+ Interaction with these variables is done via the NiFi Java API.
The 'Dynamic Properties' section below will discuss the relevant API calls as
they are introduced. </td>
+ <td><a
href="https://www.javadoc.io/doc/org.apache.nifi/nifi-api/latest/org/apache/nifi/components/PropertyValue.html";>PropertyValue</a></td>
+ </tr>
+</table>
+
+<h2>Example Scripts</h2>
+
+<p><strong>Get an incoming FlowFile from the session</strong></p>
+<p><strong>Use Case</strong>: You have incoming connection(s) to ExecuteScript
and want to retrieve one FlowFile from the queue(s) for processing.</p>
+<p><strong>Approach</strong>: Use the get() method from the session object.
This method returns the FlowFile that is next highest priority FlowFile to
process. If there is no FlowFile to process, the method will return null. NOTE:
It is possible to have null returned even if there is a steady flow of
FlowFiles into the processor. This can happen if there are multiple concurrent
tasks for the processor, and the other task(s) have already retrieved the
FlowFiles. If the script requires a F [...]
+<p><em> Groovy</em></p>
+<pre>flowFile = session.get()
+if(!flowFile) return
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFile = session.get()
+if (flowFile != None):
+ # All processing code starts at this indent
+# implicit return at the end
+</pre>
+<p><em> Javascript</em></p>
+<pre>var flowFile = session.get();
+if (flowFile != null) {
+ // All processing code goes here
+}
+</pre>
+<p><em> JRuby</em></p>
+<pre>flowFile = session.get()
+if flowFile != nil
+ # All processing code goes here
+end
+</pre>
+<p> </p>
+<p> </p>
+<p><strong>Get multiple incoming FlowFiles from the session</strong>:</p>
+<p><strong>Use Case</strong>: You have incoming connection(s) to ExecuteScript
and want to retrieve multiple FlowFiles from the queue(s) for processing.</p>
+<p><strong>Approach</strong>: Use the get(<em>maxResults</em>) method from the
session object. This method returns up to <em>maxResults</em> FlowFiles from
the work queue. If no FlowFiles are available, an empty list is returned (the
method does not return null). NOTE: If multiple incoming queues are present,
the behavior is unspecified in terms of whether all queues or only a single
queue will be polled in a single call. Having said that, the observed behavior
(for both NiFi 1.1.0+ and [...]
+<p><strong>Examples</strong>:</p>
+<p><em> Groovy</em></p>
+<pre>flowFileList = session.get(100)
+if(!flowFileList.isEmpty()) {
+ flowFileList.each { flowFile ->
+ // Process each FlowFile here
+ }
+}
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFileList = session.get(100)
+if not flowFileList.isEmpty():
+ for flowFile in flowFileList:
+ # Process each FlowFile here
+</pre>
+<p><em>Javascript</em></p>
+<pre>flowFileList = session.get(100)
+if(!flowFileList.isEmpty()) {
+ for each (var flowFile in flowFileList) {
+ // Process each FlowFile here
+ }
+}
+</pre>
+<p><em>JRuby</em></p>
+<pre>flowFileList = session.get(100)
+if !(flowFileList.isEmpty())
+ flowFileList.each { |flowFile|
+ # Process each FlowFile here
+ }
+end
+</pre>
+<p> </p>
+<p> </p>
+<p><strong>Create a new FlowFile</strong></p>
+<p><strong>Use Case</strong>: You want to generate a new FlowFile to send to
the next processor.</p>
+<p><strong>Approach</strong>: Use the create() method from the session object.
This method returns a new FlowFile object, which you can perform further
processing on</p>
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>flowFile = session.create()
+// Additional processing here
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFile = session.create()
+# Additional processing here
+</pre>
+<p><em>Javascript</em></p>
+<pre>var flowFile = session.create();
+// Additional processing here
+</pre>
+<p><em>JRuby</em></p>
+<pre>flowFile = session.create()
+# Additional processing here
+</pre>
+<p> </p>
+<p> </p>
+<p><strong>Create a new FlowFile from a parent FlowFile</strong></p>
+<p><strong>Use Case</strong>: You want to generate new FlowFile(s) based on an
incoming FlowFile.</p>
+<p><strong>Approach</strong>: Use the create(<em>parentFlowFile</em>) method
from the session object. This method takes a parent FlowFile reference and
returns a new child FlowFile object. The newly created FlowFile will inherit
all of the parent's attributes except for the UUID. This method will
automatically generate a Provenance FORK event or a Provenance JOIN event,
depending on whether or not other FlowFiles are generated from the same parent
before the ProcessSession is committed.</p>
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>flowFile = session.get()
+if(!flowFile) return
+newFlowFile = session.create(flowFile)
+// Additional processing here
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFile = session.get()
+if (flowFile != None):
+ newFlowFile = session.create(flowFile)
+ # Additional processing here
+</pre>
+<p><em>Javascript</em></p>
+<pre>var flowFile = session.get();
+if (flowFile != null) {
+ var newFlowFile = session.create(flowFile);
+ // Additional processing here
+}
+</pre>
+<p><em>JRuby</em></p>
+<pre>flowFile = session.get()
+if flowFile != nil
+ newFlowFile = session.create(flowFile)
+ # Additional processing here
+end
+</pre>
+<p> </p>
+<p> </p>
+<p><strong>Add an attribute to a FlowFile</strong></p>
+<p><strong>Use Case</strong>: You have a FlowFile to which you'd like to add a
custom attribute.</p>
+<p><strong>Approach</strong>: Use the putAttribute(<em>flowFile</em>,
<em>attributeKey</em>, <em>attributeValue</em>) method from the session object.
This method updates the given FlowFile's attributes with the given key/value
pair. NOTE: The "uuid" attribute is fixed for a FlowFile and cannot be
modified; if the key is named "uuid", it will be ignored.</p>
+<p>Also this is a good point to mention that FlowFile objects are immutable;
this means that if you update a FlowFile's attributes (or otherwise alter it)
via the API, you will get a new reference to the new version of the FlowFile.
This is very important when it comes to transferring FlowFiles to
relationships. You must keep a reference to the latest version of a FlowFile,
and you <u>must</u> transfer or remove the latest version of all FlowFiles
retrieved from or created by the session [...]
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>flowFile = session.get()
+if(!flowFile) return
+flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFile = session.get()
+if (flowFile != None):
+ flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
+# implicit return at the end
+</pre>
+<p><em>Javascript</em></p>
+<pre>var flowFile = session.get();
+if (flowFile != null) {
+ flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
+}
+</pre>
+<p><em>JRuby</em></p>
+<pre>flowFile = session.get()
+if flowFile != nil
+ flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
+end
+</pre>
+<p> </p>
+<p> </p>
+<p><strong>Add multiple attributes to a FlowFile</strong></p>
+<p><strong>Use Case</strong>: You have a FlowFile to which you'd like to add
custom attributes.</p>
+<p><strong>Approach</strong>: Use the putAllAttributes(<em>flowFile</em>,
<em>attributeMap</em>) method from the session object. This method updates the
given FlowFile's attributes with the key/value pairs from the given Map. NOTE:
The "uuid" attribute is fixed for a FlowFile and cannot be modified; if the key
is named "uuid", it will be ignored.</p>
+<p>The technique here is to create a Map (aka dictionary in Jython, hash in
JRuby) of the attribute key/value pairs you'd like to update, then call
putAllAttributes() on it. This is much more efficient than calling
putAttribute() for each key/value pair, as the latter case will cause the
framework to create a temporary version of the FlowFile for each attribute
added (see above recipe for discussion on FlowFile immutability). The examples
show a map of two entries myAttr1 and myAttr2, se [...]
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>attrMap = ['myAttr1': '1', 'myAttr2': Integer.toString(2)]
+flowFile = session.get()
+if(!flowFile) return
+flowFile = session.putAllAttributes(flowFile, attrMap)
+</pre>
+<p><em>Jython</em></p>
+<pre>attrMap = {'myAttr1':'1', 'myAttr2':str(2)}
+flowFile = session.get()
+if (flowFile != None):
+ flowFile = session.putAllAttributes(flowFile, attrMap)
+# implicit return at the end
+</pre>
+<p><em>Javascript</em></p>
+<pre>var number2 = 2;
+var attrMap = {'myAttr1':'1', 'myAttr2': number2.toString()}
+var flowFile = session.get()
+if (flowFile != null) {
+ flowFile = session.putAllAttributes(flowFile, attrMap)
+}
+</pre>
+<p><em>JRuby</em></p>
+<pre>attrMap = {'myAttr1' => '1', 'myAttr2' => 2.to_s}
+flowFile = session.get()
+if flowFile != nil
+ flowFile = session.putAllAttributes(flowFile, attrMap)
+end
+</pre>
+<p> </p>
+<p> </p>
+<p><strong>Get an attribute from a FlowFile</strong></p>
+<p><strong>Use Case</strong>: You have a FlowFile from which you'd like to
inspect an attribute.</p>
+<p><strong>Approach</strong>: Use the getAttribute(<em>attributeKey</em>)
method from the FlowFile object. This method returns the String value for the
given attributeKey, or null if the attributeKey is not found. The examples show
the retrieval of the value for the "filename" attribute.</p>
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>flowFile = session.get()
+if(!flowFile) return
+myAttr = flowFile.getAttribute('filename')
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFile = session.get()
+if (flowFile != None):
+ myAttr = flowFile.getAttribute('filename')
+# implicit return at the end
+</pre>
+<p><em>Javascript</em></p>
+<pre>var flowFile = session.get()
+if (flowFile != null) {
+ var myAttr = flowFile.getAttribute('filename')
+}
+</pre>
+<p><em>JRuby</em></p>
+<pre>flowFile = session.get()
+if flowFile != nil
+ myAttr = flowFile.getAttribute('filename')
+end
+</pre>
+<p> </p>
+<p> </p>
+<p><strong>Get all attributes from a FlowFile</strong></p>
+<p><strong>Use Case</strong>: You have a FlowFile from which you'd like to
retrieve its attributes.</p>
+<p><strong>Approach</strong>: Use the getAttributes() method from the FlowFile
object. This method returns a Map with String keys and String values,
representing the key/value pairs of attributes for the FlowFile. The examples
show an iteration over the Map of all attributes for a FlowFile.</p>
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>flowFile = session.get()
+if(!flowFile) return
+flowFile.getAttributes().each { key,value ->
+ // Do something with the key/value pair
+}
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFile = session.get()
+if (flowFile != None):
+ for key,value in flowFile.getAttributes().iteritems():
+ # Do something with key and/or value
+# implicit return at the end
+</pre>
+<p><em>Javascript</em></p>
+<pre>var flowFile = session.get()
+if (flowFile != null) {
+ var attrs = flowFile.getAttributes();
+ for each (var attrKey in attrs.keySet()) {
+ // Do something with attrKey (the key) and/or attrs[attrKey] (the value)
+ }
+}
+</pre>
+<p><em>JRuby</em></p>
+<pre>flowFile = session.get()
+if flowFile != nil
+ flowFile.getAttributes().each { |key,value|
+ # Do something with key and/or value
+ }
+end
+</pre>
+<p> </p>
+<p><strong>Transfer a FlowFile to a relationship</strong></p>
+<p><strong>Use Case</strong>: After processing a FlowFile (new or incoming),
you want to transfer the FlowFile to a relationship ("success" or "failure").
In this simple case let us assume there is a variable called "errorOccurred"
that indicates which relationship to which the FlowFile should be transferred.
Additional error handling techniques will be discussed in part 2 of this
series.</p>
+<p><strong>Approach</strong>: Use the transfer(<em>flowFile</em>,
<em>relationship</em>) method from the session object. From the documentation:
this method transfers the given FlowFile to the appropriate destination
processor work queue(s) based on the given relationship. If the relationship
leads to more than one destination the state of the FlowFile is replicated such
that each destination receives an exact copy of the FlowFile though each will
have its own unique identity.</p>
+<p><u>NOTE: ExecuteScript will perform a session.commit() at the end of each
execution to ensure the operations have been committed. You do not need to (and
should not) perform a session.commit() within the script.</u></p>
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>flowFile = session.get()
+if(!flowFile) return
+// Processing occurs here
+if(errorOccurred) {
+ session.transfer(flowFile, REL_FAILURE)
+}
+else {
+ session.transfer(flowFile, REL_SUCCESS)
+}
+</pre>
+<p><em>Jython</em></p>
+<pre>flowFile = session.get()
+if (flowFile != None):
+ # All processing code starts at this indent
+ if errorOccurred:
+ session.transfer(flowFile, REL_FAILURE)
+ else:
+ session.transfer(flowFile, REL_SUCCESS)
+# implicit return at the end
+</pre>
+<p><em>Javascript</em></p>
+<pre>var flowFile = session.get();
+if (flowFile != null) {
+ // All processing code goes here
+ if(errorOccurred) {
+ session.transfer(flowFile, REL_FAILURE)
+ }
+ else {
+ session.transfer(flowFile, REL_SUCCESS)
+ }
+}
+</pre>
+<p><em>JRuby</em></p>
+<pre>flowFile = session.get()
+if flowFile != nil
+ # All processing code goes here
+ if errorOccurred
+ session.transfer(flowFile, REL_FAILURE)
+ else
+ session.transfer(flowFile, REL_SUCCESS)
+ end
+end
+</pre>
+<p> </p>
+<p><strong>Send a message to the log at a specified logging level</strong></p>
+<p><strong>Use Case</strong>: You want to report some event that has occurred
during processing to the logging framework.</p>
+<p><strong>Approach</strong>: Use the log variable with the warn(), trace(),
debug(), info(), or error() methods. These methods can take a single String, or
a String followed by an array of Objects, or a String followed by an array of
Objects followed by a Throwable. The first one is used for simple messages. The
second is used when you have some dynamic objects/values that you want to log.
To refer to these in the message string use "{}" in the message. These are
evaluated against the O [...]
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>log.info('Found these things: {} {} {}', ['Hello',1,true] as Object[])
+</pre>
+<p><em>Jython</em></p>
+<pre>from java.lang import Object
+from jarray import array
+objArray = ['Hello',1,True]
+javaArray = array(objArray, Object)
+log.info('Found these things: {} {} {}', javaArray)
+</pre>
+<p><em>Javascript</em></p>
+<pre>var ObjectArrayType = Java.type("java.lang.Object[]");
+var objArray = new ObjectArrayType(3);
+objArray[0] = 'Hello';
+objArray[1] = 1;
+objArray[2] = true;
+log.info('Found these things: {} {} {}', objArray)
+</pre>
+<p><em>JRuby</em></p>
+<pre>log.info('Found these things: {} {} {}', ['Hello',1,true].to_java)
+</pre>
+<p> </p>
+<p><strong>Read the contents of an incoming FlowFile using a
callback</strong></p>
+<p><strong>Use Case</strong>: You have incoming connection(s) to ExecuteScript
and want to retrieve the contents of a FlowFile from the queue(s) for
processing.</p>
+<p><strong>Approach</strong>: Use the read(<em>flowFile</em>,
<em>inputStreamCallback</em>) method from the session object. An
InputStreamCallback object is needed to pass into the read() method. Note that
because InputStreamCallback is an object, the contents are only visible to that
object by default. If you need to use the data outside the read() method, use a
more globally-scoped variable. The examples will store the full contents of the
incoming FlowFile into a String (using Apache [...]
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+<pre>import org.apache.commons.io.IOUtils
+import java.nio.charset.StandardCharsets
+flowFile = session.get()
+if(!flowFile)return
+def text = ''
+// Cast a closure with an inputStream parameter to InputStreamCallback
+session.read(flowFile, {inputStream ->
+ text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
+ // Do something with text here
+} as InputStreamCallback)</pre>
+<p><em>Jython</em></p>
+
+<pre>from org.apache.commons.io import IOUtils
+from java.nio.charset import StandardCharsets
+from org.apache.nifi.processor.io import InputStreamCallback
+
+# Define a subclass of InputStreamCallback for use in session.read()
+class PyInputStreamCallback(InputStreamCallback):
+ def __init__(self):
+ pass
+ def process(self, inputStream):
+ text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
+ # Do something with text here
+# end class
+flowFile = session.get()
+if(flowFile != None):
+ session.read(flowFile, PyInputStreamCallback())
+# implicit return at the end</pre>
+<p><em>Javascript</em></p>
+
+<pre>var InputStreamCallback =
Java.type("org.apache.nifi.processor.io.InputStreamCallback")
+var IOUtils = Java.type("org.apache.commons.io.IOUtils")
+var StandardCharsets = Java.type("java.nio.charset.StandardCharsets")
+
+var flowFile = session.get();
+if(flowFile != null) {
+ // Create a new InputStreamCallback, passing in a function to define the
interface method
+ session.read(flowFile,
+ new InputStreamCallback(function(inputStream) {
+ var text = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
+ // Do something with text here
+ }));
+}</pre>
+
+<p><em>JRuby</em></p>
+
+<pre>java_import org.apache.commons.io.IOUtils
+java_import org.apache.nifi.processor.io.InputStreamCallback
+
+# Define a subclass of InputStreamCallback for use in session.read()
+class JRubyInputStreamCallback
+ include InputStreamCallback
+ def process(inputStream)
+ text = IOUtils.toString(inputStream)
+ # Do something with text here
+ end
+end
+jrubyInputStreamCallback = JRubyInputStreamCallback.new
+flowFile = session.get()
+if flowFile != nil
+ session.read(flowFile, jrubyInputStreamCallback)
+end</pre>
+<p> </p>
+<p><strong>Write content to an outgoing FlowFile using a callback</strong></p>
+<p><strong>Use Case</strong>: You want to generate content for an outgoing
FlowFile.</p>
+<p><strong>Approach</strong>: Use the write(<em>flowFile</em>,
<em>outputStreamCallback</em>) method from the session object. An
OutputStreamCallback object is needed to pass into the write() method. Note
that because OutputStreamCallback is an object, the contents are only visible
to that object by default. If you need to use the data outside the write()
method, use a more globally-scoped variable. The examples will write a sample
String to a FlowFile.</p>
+<p><strong>Examples</strong>:</p>
+<p><em>Groovy</em></p>
+
+<pre>import org.apache.commons.io.IOUtils
+import java.nio.charset.StandardCharsets
+
+flowFile = session.get()
+if(!flowFile) return
+def text = 'Hello world!'
+// Cast a closure with an outputStream parameter to OutputStreamCallback
+flowFile = session.write(flowFile, {outputStream ->
+ outputStream.write(text.getBytes(StandardCharsets.UTF_8))
+} as OutputStreamCallback)</pre>
+<p><em>Jython</em></p>
+
+<pre>from org.apache.commons.io import IOUtils
+from java.nio.charset import StandardCharsets
+from org.apache.nifi.processor.io import OutputStreamCallback
+
+# Define a subclass of OutputStreamCallback for use in session.write()
+class PyOutputStreamCallback(OutputStreamCallback):
+ def __init__(self):
+ pass
+ def process(self, outputStream):
+ outputStream.write(bytearray('Hello World!'.encode('utf-8')))
+# end class
+flowFile = session.get()
+if(flowFile != None):
+ flowFile = session.write(flowFile, PyOutputStreamCallback())
+# implicit return at the end</pre>
+
+<p><em>Javascript</em></p>
+
+<pre>var OutputStreamCallback =
Java.type("org.apache.nifi.processor.io.OutputStreamCallback");
+var IOUtils = Java.type("org.apache.commons.io.IOUtils");
+var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");
+
+var flowFile = session.get();
+if(flowFile != null) {
+ // Create a new OutputStreamCallback, passing in a function to define the
interface method
+ flowFile = session.write(flowFile,
+ new OutputStreamCallback(function(outputStream) {
+ outputStream.write("Hello World!".getBytes(StandardCharsets.UTF_8))
+ }));
+}</pre>
+
+<p><em>JRuby</em></p>
+
+<pre>java_import org.apache.commons.io.IOUtils
+java_import java.nio.charset.StandardCharsets
+java_import org.apache.nifi.processor.io.OutputStreamCallback
+
+# Define a subclass of OutputStreamCallback for use in session.write()
+class JRubyOutputStreamCallback
+ include OutputStreamCallback
+ def process(outputStream)
+ outputStream.write("Hello
World!".to_java.getBytes(StandardCharsets::UTF_8))
+ end
+end
+jrubyOutputStreamCallback = JRubyOutputStreamCallback.new
+flowFile = session.get()
+if flowFile != nil
+ flowFile = session.write(flowFile, jrubyOutputStreamCallback)
+end</pre>
+<p> </p>
+<p><strong>Overwrite an incoming FlowFile with updated content using a
callback</strong></p>
+<p><strong>Use Case</strong>: You want to reuse the incoming FlowFile but want
to modify its content for the outgoing FlowFile.</p>
+<p><strong>Approach</strong>: Use the write(<em>flowFile</em>,
<em>streamCallback</em>) method from the session object. An StreamCallback
object is needed to pass into the write() method. StreamCallback provides both
an InputStream (from the incoming FlowFile) and an outputStream (for the next
version of that FlowFile), so you can use the InputStream to get the current
contents of the FlowFile, then modify them and write them back out to the
FlowFile. This overwrites the contents of the [...]
+
+<p><strong>Examples</strong>:</p>
+
+<p><em>Groovy</em></p>
+
+<pre>import org.apache.commons.io.IOUtils
+import java.nio.charset.StandardCharsets
+
+flowFile = session.get()
+if(!flowFile) return
+def text = 'Hello world!'
+// Cast a closure with an inputStream and outputStream parameter to
StreamCallback
+flowFile = session.write(flowFile, {inputStream, outputStream ->
+ text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
+ outputStream.write(text.reverse().getBytes(StandardCharsets.UTF_8))
+} as StreamCallback)
+session.transfer(flowFile, REL_SUCCESS)</pre>
+
+<p><em>Jython</em></p>
+
+<pre>from org.apache.commons.io import IOUtils
+from java.nio.charset import StandardCharsets
+from org.apache.nifi.processor.io import StreamCallback
+
+# Define a subclass of StreamCallback for use in session.write()
+class PyStreamCallback(StreamCallback):
+ def __init__(self):
+ pass
+ def process(self, inputStream, outputStream):
+ text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
+ outputStream.write(bytearray('Hello World!'[::-1].encode('utf-8')))
+# end class
+flowFile = session.get()
+if(flowFile != None):
+ flowFile = session.write(flowFile, PyStreamCallback())
+# implicit return at the end</pre>
+
+<p><em>Javascript</em></p>
+
+<pre>var StreamCallback =
Java.type("org.apache.nifi.processor.io.StreamCallback");
+var IOUtils = Java.type("org.apache.commons.io.IOUtils");
+var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");
+
+var flowFile = session.get();
+if(flowFile != null) {
+ // Create a new StreamCallback, passing in a function to define the
interface method
+ flowFile = session.write(flowFile,
+ new StreamCallback(function(inputStream, outputStream) {
+ var text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
+
outputStream.write(text.split("").reverse().join("").getBytes(StandardCharsets.UTF_8))
+ }));
+}</pre>
+
+<p><em>JRuby</em></p>
+
+<pre>java_import org.apache.commons.io.IOUtils
+java_import java.nio.charset.StandardCharsets
+java_import org.apache.nifi.processor.io.StreamCallback
+
+# Define a subclass of StreamCallback for use in session.write()
+class JRubyStreamCallback
+ include StreamCallback
+ def process(inputStream, outputStream)
+ text = IOUtils.toString(inputStream)
+
outputStream.write((text.reverse!).to_java.getBytes(StandardCharsets::UTF_8))
+ end
+end
+jrubyStreamCallback = JRubyStreamCallback.new
+flowFile = session.get()
+if flowFile != nil
+ flowFile = session.write(flowFile, jrubyStreamCallback)
+end</pre>
+<p> </p>
+<p><strong>Handle errors during script processing</strong></p>
+<p><strong>Use Case</strong>: An error occurs in the script (either by data
validation or a thrown exception), and you want the script to handle it
gracefully.</p>
+<p><strong>Approach</strong>: For exceptions, use the exception-handling
mechanism for the scripting language (often they are try/catch block(s)). For
data validation, you can use a similar approach, but define a boolean variable
like "valid" and an if/else clause rather than a try/catch clause.
ExecuteScript defines "success" and "failure" relationships; often your
processing will transfer "good" FlowFiles to success and "bad" FlowFiles to
failure (logging an error in the latter case).</p>
+<p><strong>Examples</strong>:</p>
+
+<p><em>Groovy</em></p>
+
+<pre>flowFile = session.get()
+if(!flowFile) return
+try {
+ // Something that might throw an exception here
+
+ // Last operation is transfer to success (failures handled in the catch
block)
+ session.transfer(flowFile, REL_SUCCESS)
+} catch(e) {
+ log.error('Something went wrong', e)
+ session.transfer(flowFile, REL_FAILURE)
+}</pre>
+
+<p><em>Jython</em></p>
+
+<pre>flowFile = session.get()
+if(flowFile != None):
+ try:
+ # Something that might throw an exception here
+
+ # Last operation is transfer to success (failures handled in the catch
block)
+ session.transfer(flowFile, REL_SUCCESS)
+ except:
+ log.error('Something went wrong', e)
+ session.transfer(flowFile, REL_FAILURE)
+# implicit return at the end</pre>
+
+<p><em>Javascript</em></p>
+
+<pre>var flowFile = session.get();
+if(flowFile != null) {
+ try {
+ // Something that might throw an exception here
+
+ // Last operation is transfer to success (failures handled in the catch
block)
+ session.transfer(flowFile, REL_SUCCESS)
+} catch(e) {
+ log.error('Something went wrong', e)
+ session.transfer(flowFile, REL_FAILURE)
+}
+}</pre>
+
+<p><em>JRuby</em></p>
+
+<pre>flowFile = session.get()
+if flowFile != nil
+ begin
+ # Something that might raise an exception here
+
+ # Last operation is transfer to success (failures handled in the rescue
block)
+ session.transfer(flowFile, REL_SUCCESS)
+ rescue Exception => e
+ log.error('Something went wrong', e)
+ session.transfer(flowFile, REL_FAILURE)
+ end
+end</pre>
+
+</body>
+</html>
\ No newline at end of file
