Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/185#discussion_r51156562
--- Diff:
nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/java/org/apache/nifi/processors/script/InvokeScriptProcessor.java
---
@@ -0,0 +1,835 @@
+/*
+ * 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.script;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.controller.ControllerServiceLookup;
+import org.apache.nifi.expression.AttributeExpression;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.stream.io.ByteArrayInputStream;
+import org.apache.nifi.util.file.monitor.LastModifiedMonitor;
+import org.apache.nifi.util.file.monitor.SynchronousFileWatcher;
+
+import javax.script.Invocable;
+import javax.script.ScriptEngine;
+import javax.script.ScriptEngineFactory;
+import javax.script.ScriptEngineManager;
+import javax.script.ScriptException;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+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;
+
+@Tags({"script", "invoke", "groovy", "python", "jython", "jruby", "ruby",
"javascript", "js", "lua", "luaj", "scala"})
+@CapabilityDescription("Invokes a script engine for a Processor defined in
the given script. The script must define "
+ + "a valid class that implements the Processor interface, and it
must set a variable 'processor' to an instance of "
+ + "the class. Processor methods such as onTrigger() will be
delegated to the scripted Processor instance. Also any "
+ + "Relationships or PropertyDescriptors defined by the scripted
processor will be added to the configuration dialog.")
+@DynamicProperty(name = "A script engine property to update", value = "The
value to set it to", supportsExpressionLanguage = true,
+ description = "Updates a script engine property specified by the
Dynamic Property's key with the value specified by the Dynamic Property's
value")
+@SeeAlso({ExecuteScript.class})
+public class InvokeScriptProcessor extends AbstractSessionFactoryProcessor
{
+
+ public static final Relationship REL_SUCCESS = new
Relationship.Builder()
+ .name("success")
+ .description("FlowFiles that were successfully processed")
+ .build();
+
+ public static final Relationship REL_FAILURE = new
Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles that were failed to process")
+ .build();
+
+
+ public static PropertyDescriptor SCRIPT_ENGINE;
+
+ public static final PropertyDescriptor SCRIPT_FILE = new
PropertyDescriptor.Builder()
+ .name("Script File")
+ .required(false)
+ .description("Path to script file to execute. Use either file
or body not both")
+ .addValidator(new StandardValidators.FileExistsValidator(true))
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor SCRIPT_BODY = new
PropertyDescriptor.Builder()
+ .name("Script Body")
+ .required(false)
+ .description("Body to script to execute. Use either file or
body not both")
+ .addValidator(Validator.VALID)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor MODULES = new
PropertyDescriptor.Builder()
+ .name("Module Directory")
+ .description("Path to a directory which contains modules
required by the script.")
+ .required(false)
+ .expressionLanguageSupported(true)
+ .addValidator(new
StandardValidators.DirectoryExistsValidator(true, false))
+ .build();
+
+ // A map from engine name to a custom configurator for that engine
+ private final Map<String, ScriptEngineConfigurator>
scriptEngineConfiguratorMap = new ConcurrentHashMap<>();
+
+ private final AtomicReference<Processor> processor = new
AtomicReference<>();
+ private final AtomicReference<Collection<ValidationResult>>
validationResults =
+ new AtomicReference(Collections.EMPTY_LIST);
+
+ private ControllerServiceLookup controllerServiceLookup;
+ private String initializationContextId;
+
+ private final AtomicBoolean isInitialized = new AtomicBoolean(false);
+ private final Lock lock = new ReentrantLock();
+ private SynchronousFileWatcher scriptWatcher;
+
+ private Map<String, ScriptEngineFactory> scriptEngineFactoryMap;
+ private ScriptEngineManager scriptEngineManager;
+ private ScriptEngine scriptEngine;
+ private String scriptEngineName;
+ private String scriptPath;
+ private String modulePath;
+ private String scriptBody;
+
+ private ScheduledExecutorService reloadService = null;
+
+ private List<PropertyDescriptor> descriptors;
+
+ /**
+ * Creates the resources needed by this processor. An attempt is made
to also initialize the scripted processor,
+ * but unless the properties (such as script engine name and script
file path) have already been specified, the
+ * script will not yet have been evaluated, so the script's
initialize() method will not be called.
+ */
+ protected void createResources() {
+
+ // Set up script file reloader service. This checks to see if the
script file has changed, and if so, tries
+ // to reload it
+ if (reloadService == null) {
+ reloadService = Executors.newScheduledThreadPool(1);
+
+ // monitor the script if configured for changes
+ reloadService.scheduleWithFixedDelay(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ final boolean hasLock = lock.tryLock();
+
+ // if a property is changing we don't need to
reload this iteration
+ if (hasLock) {
+ try {
+ if (scriptWatcher != null &&
scriptWatcher.checkAndReset()) {
+ if (isFile(scriptPath)) {
+ // reload the actual script
+ final boolean reloaded =
reloadScriptFile(scriptPath);
+
+ // log the script was reloaded
+ if (reloaded) {
+ getLogger().info("The
configured script has been successfully reloaded.");
+ }
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ } catch (final Throwable t) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to reload
configured script Processor: " + t;
+
+ logger.error(message);
+ if (logger.isDebugEnabled()) {
+ logger.error(message, t);
+ }
+ }
+ }
+ }, 30, 10, TimeUnit.SECONDS);
+ }
+
+ descriptors = new ArrayList<>();
+
+ // The following is required for JRuby, should be transparent to
everything else
+ System.setProperty("org.jruby.embed.localvariable.behavior",
"persistent");
+
+ // Create list of available engines
+ scriptEngineManager = new ScriptEngineManager();
+ List<ScriptEngineFactory> scriptEngineFactories =
scriptEngineManager.getEngineFactories();
+ if (scriptEngineFactories != null) {
+ scriptEngineFactoryMap = new HashMap<>();
+ List<AllowableValue> engineList = new ArrayList<>();
+ for (ScriptEngineFactory factory : scriptEngineFactories) {
+ ScriptEngine engine = factory.getScriptEngine();
+ if (engine instanceof Invocable) {
+ engineList.add(new
AllowableValue(factory.getLanguageName()));
+ scriptEngineFactoryMap.put(factory.getLanguageName(),
factory);
+ }
+ }
+ // Sort the list by name so the list always looks the same.
+ Collections.sort(engineList, new Comparator<AllowableValue>() {
+ @Override
+ public int compare(AllowableValue o1, AllowableValue o2) {
+ if (o1 == null) {
+ return o2 == null ? 0 : 1;
+ }
+ if (o2 == null) {
+ return -1;
+ }
+ return o1.getValue().compareTo(o2.getValue());
+ }
+ });
+
+ AllowableValue[] engines = engineList.toArray(new
AllowableValue[engineList.size()]);
+ SCRIPT_ENGINE = new PropertyDescriptor.Builder()
+ .name("Script Engine")
+ .required(true)
+ .description("The engine to execute the scripted
Processor")
+ .allowableValues(engines)
+ .defaultValue(engines[0].getValue())
+ .required(true)
+ .expressionLanguageSupported(false)
+ .build();
+ descriptors.add(SCRIPT_ENGINE);
+ }
+
+ descriptors.add(SCRIPT_FILE);
+ descriptors.add(SCRIPT_BODY);
+ descriptors.add(MODULES);
+
+ isInitialized.set(true);
+ }
+
+ /**
+ * Returns the valid relationships for this processor. SUCCESS and
FAILURE are always returned, and if the script
+ * processor has defined additional relationships, those will be added
as well.
+ *
+ * @return a Set of Relationships supported by this processor
+ */
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ final Processor instance = processor.get();
+ if (instance != null) {
+ try {
+ relationships.addAll(instance.getRelationships());
+ } catch (final Throwable t) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to get relationships from
scripted Processor: " + t;
+
+ logger.error(message);
+ if (logger.isDebugEnabled()) {
+ logger.error(message, t);
+ }
+ }
+ } else {
+ // Return defaults for now
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ }
+ return Collections.unmodifiableSet(relationships);
+ }
+
+ /**
+ * Returns a list of property descriptors supported by this processor.
The list always includes properties such as
+ * script engine name, script file name, script body name, script
arguments, and an external module path. If the
+ * scripted processor also defines supported properties, those are
added to the list as well.
+ *
+ * @return a List of PropertyDescriptor objects supported by this
processor
+ */
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+
+ synchronized (isInitialized) {
+ if (!isInitialized.get()) {
+ createResources();
+ }
+ }
+ List<PropertyDescriptor> supportedPropertyDescriptors = new
ArrayList<>();
+ supportedPropertyDescriptors.addAll(descriptors);
+
+ final Processor instance = processor.get();
+ if (instance != null) {
+ try {
+ final List<PropertyDescriptor> instanceDescriptors =
instance.getPropertyDescriptors();
+ if (instanceDescriptors != null) {
+
supportedPropertyDescriptors.addAll(instanceDescriptors);
+ }
+ } catch (final Throwable t) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to get property descriptors
from Processor: " + t;
+
+ logger.error(message);
+ if (logger.isDebugEnabled()) {
+ logger.error(message, t);
+ }
+ }
+ }
+
+ return Collections.unmodifiableList(supportedPropertyDescriptors);
+ }
+
+ /**
+ * Returns a PropertyDescriptor for the given name. This is for the
user to be able to define their own properties
+ * which will be available as variables in the script
+ *
+ * @param propertyDescriptorName used to lookup if any property
descriptors exist for that name
+ * @return a PropertyDescriptor object corresponding to the specified
dynamic property name
+ */
+ @Override
+ protected PropertyDescriptor
getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .required(false)
+
.addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING,
true))
+
.addValidator(StandardValidators.ATTRIBUTE_KEY_PROPERTY_NAME_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .dynamic(true)
+ .build();
+ }
+
+ /**
+ * Determines whether the given path refers to a valid file
+ *
+ * @param path a path to a file
+ * @return true if the path refers to a valid file, false otherwise
+ */
+ private boolean isFile(final String path) {
+ return path != null && Files.isRegularFile(Paths.get(path));
+ }
+
+ /**
+ * Performs setup operations when the processor is scheduled to run.
This includes evaluating the processor's
+ * properties, as well as reloading the script (from file or the
"Script Body" property)
+ *
+ * @param context the context in which to perform the setup operations
+ */
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ scriptEngineName = context.getProperty(SCRIPT_ENGINE).getValue();
+ scriptPath = context.getProperty(SCRIPT_FILE).getValue();
+ scriptBody = context.getProperty(SCRIPT_BODY).getValue();
+ modulePath = context.getProperty(MODULES).getValue();
+ setupEngine();
+ if (processor.get() == null) {
+ if (isFile(scriptPath)) {
+ reloadScriptFile(scriptPath);
+ } else {
+ reloadScriptBody(scriptBody);
+ }
+ }
+ }
+
+ /**
+ * Configures the specified script engine. First, the engine is loaded
and instantiated using the JSR-223
+ * javax.script APIs. Then, if any script configurators have been
defined for this engine, their init() method is
+ * called, and the configurator is saved for future calls.
+ *
+ * @see org.apache.nifi.processors.script.ScriptEngineConfigurator
+ */
+ private void setupEngine() {
+ ClassLoader originalContextClassLoader =
Thread.currentThread().getContextClassLoader();
+ try {
+ ProcessorLog log = getLogger();
+
+ // Need the right classloader when the engine is created. This
ensures the NAR's execution class loader
+ // (plus the module path) becomes the parent for the script
engine
+ ClassLoader scriptEngineModuleClassLoader =
createScriptEngineModuleClassLoader(modulePath);
+ if (scriptEngineModuleClassLoader != null) {
+
Thread.currentThread().setContextClassLoader(scriptEngineModuleClassLoader);
+ }
+ scriptEngine = getScriptEngine();
+ ServiceLoader<ScriptEngineConfigurator>
configuratorServiceLoader =
+ ServiceLoader.load(ScriptEngineConfigurator.class);
+ for (ScriptEngineConfigurator configurator :
configuratorServiceLoader) {
+ String configuratorScriptEngineName =
configurator.getScriptEngineName();
+ try {
+ if (configuratorScriptEngineName != null
+ &&
configuratorScriptEngineName.equalsIgnoreCase(scriptEngineName)) {
+ configurator.init(scriptEngine, modulePath);
+
scriptEngineConfiguratorMap.put(configurator.getScriptEngineName(),
configurator);
+ }
+ } catch (ScriptException se) {
+ log.error("Error initializing script engine
configurator {}",
+ new Object[]{configuratorScriptEngineName});
+ if (log.isDebugEnabled()) {
+ log.error("Error initializing script engine
configurator", se);
+ }
+ }
+ }
+
+ } finally {
+ // Restore original context class loader
+
Thread.currentThread().setContextClassLoader(originalContextClassLoader);
+ }
+ }
+
+ /**
+ * Provides a ScriptEngine corresponding to the currently selected
script engine name.
+ * ScriptEngineManager.getEngineByName() doesn't use find
ScriptEngineFactory.getName(), which
+ * is what we used to populate the list. So just search the list of
factories until a match is
+ * found, then create and return a script engine.
+ *
+ * @return a Script Engine corresponding to the currently specified
name, or null if none is found.
+ */
+ private ScriptEngine getScriptEngine() {
+ //
--- End diff --
This empty comment can be removed, I think.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
