You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2016/06/19 00:47:33 UTC
[1/5] nifi git commit: NIFI-1974 - initial commit variable registry
for custom property support
Repository: nifi
Updated Branches:
refs/heads/0.x 90c749d73 -> 2cdb1c190
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
index cebedaf..4dfac22 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
@@ -29,15 +29,19 @@ import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.repository.StandardFlowFileRecord;
import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.junit.Test;
+
public class TestStandardPropertyValue {
private final ControllerServiceLookup lookup = new TestControllerServiceLookup();
+ private final VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
@Test
public void testSubstituteAttributesWithOneMatchingArg() {
- final PropertyValue value = new StandardPropertyValue("Hello, ${audience}!", lookup);
+ final PropertyValue value = new StandardPropertyValue("Hello, ${audience}!", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
attributes.put("audience", "World");
assertEquals("Hello, World!", value.evaluateAttributeExpressions(createFlowFile(attributes)).getValue());
@@ -45,7 +49,7 @@ public class TestStandardPropertyValue {
@Test
public void testMissingEndBraceEvaluatesToStringLiteral() {
- final PropertyValue value = new StandardPropertyValue("Hello, ${audience!", lookup);
+ final PropertyValue value = new StandardPropertyValue("Hello, ${audience!", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
attributes.put("audience", "World");
assertEquals("Hello, ${audience!", value.evaluateAttributeExpressions(createFlowFile(attributes)).getValue());
@@ -53,7 +57,7 @@ public class TestStandardPropertyValue {
@Test
public void testEscaped() {
- final PropertyValue value = new StandardPropertyValue("Hello, $${audience}!", lookup);
+ final PropertyValue value = new StandardPropertyValue("Hello, $${audience}!", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
attributes.put("audience", "World");
assertEquals("Hello, ${audience}!", value.evaluateAttributeExpressions(createFlowFile(attributes)).getValue());
@@ -61,7 +65,7 @@ public class TestStandardPropertyValue {
@Test
public void testSubstituteAttributesWithMultipleMatchingArgs() {
- final PropertyValue value = new StandardPropertyValue("Hello, ${audience}${comma}${question}!", lookup);
+ final PropertyValue value = new StandardPropertyValue("Hello, ${audience}${comma}${question}!", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
attributes.put("audience", "World");
attributes.put("comma", ",");
@@ -71,14 +75,14 @@ public class TestStandardPropertyValue {
@Test
public void testSubstituteAttributesWithNoMatch() {
- final PropertyValue value = new StandardPropertyValue("Hello, ${audience}${comma}${question:replaceNull('')}!", lookup);
+ final PropertyValue value = new StandardPropertyValue("Hello, ${audience}${comma}${question:replaceNull('')}!", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
assertEquals("Hello, !", value.evaluateAttributeExpressions(createFlowFile(attributes)).getValue());
}
@Test
public void testSubstituteAttributesRecursively() {
- final PropertyValue value = new StandardPropertyValue("Hello, ${'${a}${b}'}!", lookup);
+ final PropertyValue value = new StandardPropertyValue("Hello, ${'${a}${b}'}!", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
attributes.put("a", "b");
attributes.put("b", "World");
@@ -88,7 +92,7 @@ public class TestStandardPropertyValue {
@Test
public void testGetValueAsIntegerAfterSubstitute() {
- final PropertyValue value = new StandardPropertyValue("1${value}", lookup);
+ final PropertyValue value = new StandardPropertyValue("1${value}", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
attributes.put("value", "39");
assertEquals(139, value.evaluateAttributeExpressions(createFlowFile(attributes)).asInteger().intValue());
@@ -96,7 +100,7 @@ public class TestStandardPropertyValue {
@Test(expected = NumberFormatException.class)
public void testGetValueAsIntegerAfterSubstitutingWithNonInteger() {
- final PropertyValue value = new StandardPropertyValue("1${value}", lookup);
+ final PropertyValue value = new StandardPropertyValue("1${value}", lookup, variableRegistry);
final Map<String, String> attributes = new HashMap<>();
attributes.put("value", "Yes");
final PropertyValue substituted = value.evaluateAttributeExpressions(createFlowFile(attributes));
@@ -105,7 +109,7 @@ public class TestStandardPropertyValue {
@Test
public void testFileSize() {
- final PropertyValue value = new StandardPropertyValue("${fileSize}", lookup);
+ final PropertyValue value = new StandardPropertyValue("${fileSize}", lookup, variableRegistry);
final FlowFile flowFile = new StandardFlowFileRecord.Builder().size(1024 * 1024L).build();
final long val = value.evaluateAttributeExpressions(flowFile).asLong().longValue();
assertEquals(1024 * 1024L, val);
@@ -115,7 +119,7 @@ public class TestStandardPropertyValue {
public void testFlowFileEntryYear() {
final Calendar now = Calendar.getInstance();
final int year = now.get(Calendar.YEAR);
- final PropertyValue value = new StandardPropertyValue("${entryDate:toNumber():toDate():format('yyyy')}", lookup);
+ final PropertyValue value = new StandardPropertyValue("${entryDate:toNumber():toDate():format('yyyy')}", lookup, variableRegistry);
final FlowFile flowFile = new StandardFlowFileRecord.Builder().entryDate(now.getTimeInMillis()).build();
final int val = value.evaluateAttributeExpressions(flowFile).asInteger().intValue();
assertEquals(year, val);
@@ -125,7 +129,7 @@ public class TestStandardPropertyValue {
public void testSystemProperty() {
System.setProperty("Prop1", "Foo");
System.setProperty("Prop2", "Bar");
- final PropertyValue value = new StandardPropertyValue("${Prop1}${Prop2}${abc}", lookup);
+ final PropertyValue value = new StandardPropertyValue("${Prop1}${Prop2}${abc}", lookup, VariableRegistryUtils.createVariableRegistry());
final Map<String, String> attributes = new HashMap<>();
attributes.put("abc", "Baz");
final FlowFile flowFile = createFlowFile(attributes);
@@ -171,5 +175,6 @@ public class TestStandardPropertyValue {
public boolean isControllerServiceEnabling(String serviceIdentifier) {
return false;
}
+
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
index 24d2295..7b64982 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
@@ -185,4 +185,8 @@ nifi.cluster.manager.safemode.duration=${nifi.cluster.manager.safemode.duration}
nifi.kerberos.krb5.file=${nifi.kerberos.krb5.file}
nifi.kerberos.service.principal=${nifi.kerberos.service.principal}
nifi.kerberos.keytab.location=${nifi.kerberos.keytab.location}
-nifi.kerberos.authentication.expiration=${nifi.kerberos.authentication.expiration}
\ No newline at end of file
+nifi.kerberos.authentication.expiration=${nifi.kerberos.authentication.expiration}
+
+# external properties files for variable registry
+# supports a comma delimited list of file locations
+nifi.variable.registry.properties=
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebConfigurationContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebConfigurationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebConfigurationContext.java
index 9076063..d1d0796 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebConfigurationContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebConfigurationContext.java
@@ -30,6 +30,9 @@ import org.apache.nifi.cluster.manager.impl.WebClusterManager;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.reporting.ReportingTaskProvider;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryFactory;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.user.NiFiUser;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.api.dto.ControllerServiceDTO;
@@ -54,6 +57,8 @@ import org.springframework.security.core.context.SecurityContextHolder;
import javax.ws.rs.HttpMethod;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
+
+import java.io.IOException;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.net.URI;
@@ -244,6 +249,17 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
return componentFacade;
}
+ @Override
+ public VariableRegistry getVariableRegistry() {
+ VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ try {
+ variableRegistry.addRegistry(VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths()));
+ }catch(IOException ioe){
+ logger.error("Exception occurred loading custom properties",ioe);
+ }
+ return variableRegistry;
+ }
+
/**
* Facade over accessing different types of NiFi components.
*/
@@ -848,5 +864,4 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
public void setReportingTaskProvider(ReportingTaskProvider reportingTaskProvider) {
this.reportingTaskProvider = reportingTaskProvider;
}
-
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/StandardSearchContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/StandardSearchContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/StandardSearchContext.java
index f673075..1ef12ee 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/StandardSearchContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/StandardSearchContext.java
@@ -53,7 +53,7 @@ public class StandardSearchContext implements SearchContext {
@Override
public PropertyValue getProperty(PropertyDescriptor property) {
final String configuredValue = processorNode.getProperty(property);
- return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, controllerServiceLookup, null);
+ return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, controllerServiceLookup,null);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/test/java/org/apache/nifi/processors/script/TestInvokeGroovy.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/test/java/org/apache/nifi/processors/script/TestInvokeGroovy.java b/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/test/java/org/apache/nifi/processors/script/TestInvokeGroovy.java
index 2dc700d..e57ad8a 100644
--- a/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/test/java/org/apache/nifi/processors/script/TestInvokeGroovy.java
+++ b/nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/test/java/org/apache/nifi/processors/script/TestInvokeGroovy.java
@@ -18,6 +18,8 @@ package org.apache.nifi.processors.script;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.MockProcessContext;
import org.apache.nifi.util.MockProcessorInitializationContext;
@@ -37,9 +39,12 @@ import static org.junit.Assert.assertTrue;
public class TestInvokeGroovy extends BaseScriptTest {
+ private VariableRegistry variableRegistry;
+
@Before
public void setup() throws Exception {
super.setupInvokeScriptProcessor();
+ variableRegistry = VariableRegistryUtils.createVariableRegistry();
}
/**
@@ -72,7 +77,7 @@ public class TestInvokeGroovy extends BaseScriptTest {
@Test
public void testScriptDefinedAttribute() throws Exception {
InvokeScriptedProcessor processor = new InvokeScriptedProcessor();
- MockProcessContext context = new MockProcessContext(processor);
+ MockProcessContext context = new MockProcessContext(processor, variableRegistry);
MockProcessorInitializationContext initContext = new MockProcessorInitializationContext(processor, context);
processor.initialize(initContext);
@@ -81,7 +86,7 @@ public class TestInvokeGroovy extends BaseScriptTest {
context.setProperty(InvokeScriptedProcessor.SCRIPT_FILE, "target/test/resources/groovy/test_reader.groovy");
context.setProperty(InvokeScriptedProcessor.MODULES, "target/test/resources/groovy");
// State Manger is unused, and a null reference is specified
- processor.customValidate(new MockValidationContext(context, null));
+ processor.customValidate(new MockValidationContext(context, null, variableRegistry));
processor.setup(context);
List<PropertyDescriptor> descriptors = processor.getSupportedPropertyDescriptors();
@@ -106,7 +111,7 @@ public class TestInvokeGroovy extends BaseScriptTest {
@Test
public void testScriptDefinedRelationship() throws Exception {
InvokeScriptedProcessor processor = new InvokeScriptedProcessor();
- MockProcessContext context = new MockProcessContext(processor);
+ MockProcessContext context = new MockProcessContext(processor, variableRegistry);
MockProcessorInitializationContext initContext = new MockProcessorInitializationContext(processor, context);
processor.initialize(initContext);
@@ -114,7 +119,7 @@ public class TestInvokeGroovy extends BaseScriptTest {
context.setProperty(InvokeScriptedProcessor.SCRIPT_ENGINE, "Groovy");
context.setProperty(InvokeScriptedProcessor.SCRIPT_FILE, "target/test/resources/groovy/test_reader.groovy");
// State Manger is unused, and a null reference is specified
- processor.customValidate(new MockValidationContext(context, null));
+ processor.customValidate(new MockValidationContext(context, null, variableRegistry));
processor.setup(context);
Set<Relationship> relationships = processor.getRelationships();
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/test/java/org/apache/nifi/reporting/TestSiteToSiteProvenanceReportingTask.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/test/java/org/apache/nifi/reporting/TestSiteToSiteProvenanceReportingTask.java b/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/test/java/org/apache/nifi/reporting/TestSiteToSiteProvenanceReportingTask.java
index 265bdd0..12a5e4c 100644
--- a/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/test/java/org/apache/nifi/reporting/TestSiteToSiteProvenanceReportingTask.java
+++ b/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/test/java/org/apache/nifi/reporting/TestSiteToSiteProvenanceReportingTask.java
@@ -39,6 +39,7 @@ import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventRepository;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.provenance.StandardProvenanceEventRecord;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.remote.Transaction;
import org.apache.nifi.remote.TransferDirection;
import org.apache.nifi.remote.client.SiteToSiteClient;
@@ -126,7 +127,7 @@ public class TestSiteToSiteProvenanceReportingTask {
@Override
public PropertyValue answer(final InvocationOnMock invocation) throws Throwable {
final PropertyDescriptor descriptor = invocation.getArgumentAt(0, PropertyDescriptor.class);
- return new MockPropertyValue(properties.get(descriptor), null);
+ return new MockPropertyValue(properties.get(descriptor), null, VariableRegistryUtils.createVariableRegistry());
}
}).when(context).getProperty(Mockito.any(PropertyDescriptor.class));
@@ -181,4 +182,6 @@ public class TestSiteToSiteProvenanceReportingTask {
mockFlowFile.putAttributes(attributes);
return mockFlowFile;
}
+
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestRouteOnAttribute.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestRouteOnAttribute.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestRouteOnAttribute.java
index 66dc854..7c6435d 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestRouteOnAttribute.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestRouteOnAttribute.java
@@ -26,6 +26,8 @@ import java.util.Map;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.state.MockStateManager;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.MockProcessContext;
@@ -36,10 +38,12 @@ import org.junit.Test;
public class TestRouteOnAttribute {
+ private VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+
@Test
public void testInvalidOnMisconfiguredProperty() {
final RouteOnAttribute proc = new RouteOnAttribute();
- final MockProcessContext ctx = new MockProcessContext(proc, new MockStateManager(proc));
+ final MockProcessContext ctx = new MockProcessContext(proc, new MockStateManager(proc), variableRegistry);
final ValidationResult validationResult = ctx.setProperty("RouteA", "${a:equals('b')"); // Missing closing brace
assertFalse(validationResult.isValid());
}
@@ -47,7 +51,7 @@ public class TestRouteOnAttribute {
@Test
public void testInvalidOnNonBooleanProperty() {
final RouteOnAttribute proc = new RouteOnAttribute();
- final MockProcessContext ctx = new MockProcessContext(proc, new MockStateManager(proc));
+ final MockProcessContext ctx = new MockProcessContext(proc, new MockStateManager(proc), variableRegistry);
final ValidationResult validationResult = ctx.setProperty("RouteA", "${a:length()"); // Should be boolean
assertFalse(validationResult.isValid());
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/TestServerAndClient.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/TestServerAndClient.java b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/TestServerAndClient.java
index 82e4a99..ccd1dd5 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/TestServerAndClient.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/TestServerAndClient.java
@@ -41,6 +41,8 @@ import org.apache.nifi.distributed.cache.client.exception.DeserializationExcepti
import org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockConfigurationContext;
import org.apache.nifi.util.MockControllerServiceInitializationContext;
@@ -55,6 +57,7 @@ import org.slf4j.LoggerFactory;
public class TestServerAndClient {
private static Logger LOGGER;
+ private static VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
static {
System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "info");
@@ -329,7 +332,7 @@ public class TestServerAndClient {
clientProperties.put(DistributedMapCacheClientService.HOSTNAME, "localhost");
clientProperties.put(DistributedMapCacheClientService.PORT, String.valueOf(server.getPort()));
clientProperties.put(DistributedMapCacheClientService.COMMUNICATIONS_TIMEOUT, "360 secs");
- MockConfigurationContext clientContext = new MockConfigurationContext(clientProperties, clientInitContext.getControllerServiceLookup());
+ MockConfigurationContext clientContext = new MockConfigurationContext(clientProperties, clientInitContext.getControllerServiceLookup(), variableRegistry);
client.cacheConfig(clientContext);
final Serializer<String> valueSerializer = new StringSerializer();
final Serializer<String> keySerializer = new StringSerializer();
@@ -375,7 +378,7 @@ public class TestServerAndClient {
client2.initialize(clientInitContext2);
MockConfigurationContext clientContext2 = new MockConfigurationContext(clientProperties,
- clientInitContext2.getControllerServiceLookup());
+ clientInitContext2.getControllerServiceLookup(), variableRegistry);
client2.cacheConfig(clientContext2);
assertFalse(client2.putIfAbsent("testKey", "test", keySerializer, valueSerializer));
assertTrue(client2.containsKey("testKey", keySerializer));
@@ -408,7 +411,7 @@ public class TestServerAndClient {
server.initialize(serverInitContext);
final Map<PropertyDescriptor, String> serverProperties = new HashMap<>();
- final MockConfigurationContext serverContext = new MockConfigurationContext(serverProperties, serverInitContext.getControllerServiceLookup());
+ final MockConfigurationContext serverContext = new MockConfigurationContext(serverProperties, serverInitContext.getControllerServiceLookup(), variableRegistry);
server.startServer(serverContext);
DistributedMapCacheClientService client = new DistributedMapCacheClientService();
@@ -418,7 +421,7 @@ public class TestServerAndClient {
final Map<PropertyDescriptor, String> clientProperties = new HashMap<>();
clientProperties.put(DistributedMapCacheClientService.HOSTNAME, "localhost");
clientProperties.put(DistributedMapCacheClientService.COMMUNICATIONS_TIMEOUT, "360 secs");
- MockConfigurationContext clientContext = new MockConfigurationContext(clientProperties, clientInitContext.getControllerServiceLookup());
+ MockConfigurationContext clientContext = new MockConfigurationContext(clientProperties, clientInitContext.getControllerServiceLookup(), variableRegistry);
client.cacheConfig(clientContext);
final Serializer<String> valueSerializer = new StringSerializer();
final Serializer<String> keySerializer = new StringSerializer();
@@ -465,7 +468,7 @@ public class TestServerAndClient {
final Map<PropertyDescriptor, String> clientProperties = new HashMap<>();
clientProperties.put(DistributedSetCacheClientService.HOSTNAME, "localhost");
clientProperties.put(DistributedSetCacheClientService.PORT, String.valueOf(port));
- final MockConfigurationContext clientContext = new MockConfigurationContext(clientProperties, clientInitContext.getControllerServiceLookup());
+ final MockConfigurationContext clientContext = new MockConfigurationContext(clientProperties, clientInitContext.getControllerServiceLookup(), variableRegistry);
client.onConfigured(clientContext);
return client;
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-processor/src/test/java/org/apache/nifi/update/attributes/TestUpdateAttribute.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-processor/src/test/java/org/apache/nifi/update/attributes/TestUpdateAttribute.java b/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-processor/src/test/java/org/apache/nifi/update/attributes/TestUpdateAttribute.java
index 90b51bd..d73b308 100644
--- a/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-processor/src/test/java/org/apache/nifi/update/attributes/TestUpdateAttribute.java
+++ b/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-processor/src/test/java/org/apache/nifi/update/attributes/TestUpdateAttribute.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.update.attributes;
+
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collection;
@@ -32,6 +33,7 @@ import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
+
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -43,6 +45,7 @@ public class TestUpdateAttribute {
final static private String TEST_CONTENT = "THIS IS TEST CONTENT";
+
private Map<String, String> getMap(String... keyValues) {
final Map<String, String> map = new HashMap<>();
if (keyValues != null) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/UpdateAttributeModelFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/UpdateAttributeModelFactory.java b/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/UpdateAttributeModelFactory.java
index 0182d61..6102e82 100644
--- a/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/UpdateAttributeModelFactory.java
+++ b/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/UpdateAttributeModelFactory.java
@@ -23,6 +23,7 @@ import org.apache.nifi.attribute.expression.language.Query;
import org.apache.nifi.attribute.expression.language.StandardExpressionLanguageCompiler;
import org.apache.nifi.attribute.expression.language.exception.AttributeExpressionLanguageParsingException;
import org.apache.nifi.expression.AttributeExpression.ResultType;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.update.attributes.dto.ActionDTO;
import org.apache.nifi.update.attributes.dto.ConditionDTO;
import org.apache.nifi.update.attributes.dto.RuleDTO;
@@ -32,6 +33,12 @@ import org.apache.nifi.update.attributes.dto.RuleDTO;
*/
public class UpdateAttributeModelFactory {
+ private final VariableRegistry variableRegistry;
+
+ public UpdateAttributeModelFactory(VariableRegistry variableRegistry) {
+ this.variableRegistry = variableRegistry;
+ }
+
public Rule createRule(final RuleDTO dto) {
if (dto == null) {
throw new IllegalArgumentException("Rule must be specified.");
@@ -72,7 +79,7 @@ public class UpdateAttributeModelFactory {
}
// validate the condition's expression
- final StandardExpressionLanguageCompiler elCompiler = new StandardExpressionLanguageCompiler();
+ final StandardExpressionLanguageCompiler elCompiler = new StandardExpressionLanguageCompiler(variableRegistry);
final String syntaxError = elCompiler.validateExpression(dto.getExpression(), false);
if (syntaxError != null) {
throw new IllegalArgumentException(syntaxError);
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/api/RuleResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/api/RuleResource.java b/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/api/RuleResource.java
index b211b72..f099b55 100644
--- a/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/api/RuleResource.java
+++ b/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-ui/src/main/java/org/apache/nifi/update/attributes/api/RuleResource.java
@@ -46,6 +46,7 @@ import javax.ws.rs.core.Response.ResponseBuilder;
import javax.ws.rs.core.UriBuilder;
import javax.ws.rs.core.UriInfo;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.update.attributes.Action;
import org.apache.nifi.update.attributes.Condition;
import org.apache.nifi.update.attributes.Criteria;
@@ -187,6 +188,7 @@ public class RuleResource {
// get the web context
final NiFiWebConfigurationContext configurationContext = (NiFiWebConfigurationContext) servletContext.getAttribute("nifi-web-configuration-context");
+ final VariableRegistry variableRegistry = configurationContext.getVariableRegistry();
// ensure the rule has been specified
if (requestEntity == null || requestEntity.getRule() == null) {
@@ -219,7 +221,7 @@ public class RuleResource {
// load the criteria
final Criteria criteria = getCriteria(configurationContext, requestContext);
- final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory();
+ final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory(variableRegistry);
// create the new rule
final Rule rule;
@@ -261,10 +263,14 @@ public class RuleResource {
// generate a new id
final String uuid = UUID.randomUUID().toString();
+ // get the variable registry
+ final NiFiWebConfigurationContext configurationContext = (NiFiWebConfigurationContext) servletContext.getAttribute("nifi-web-configuration-context");
+ final VariableRegistry variableRegistry = configurationContext.getVariableRegistry();
+
final Condition condition;
try {
// create the condition object
- final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory();
+ final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory(variableRegistry);
condition = factory.createCondition(requestEntity.getCondition());
condition.setId(uuid);
} catch (final IllegalArgumentException iae) {
@@ -295,10 +301,14 @@ public class RuleResource {
// generate a new id
final String uuid = UUID.randomUUID().toString();
+ // get the variable registry
+ final NiFiWebConfigurationContext configurationContext = (NiFiWebConfigurationContext) servletContext.getAttribute("nifi-web-configuration-context");
+ final VariableRegistry variableRegistry = configurationContext.getVariableRegistry();
+
final Action action;
try {
// create the condition object
- final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory();
+ final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory(variableRegistry);
action = factory.createAction(requestEntity.getAction());
action.setId(uuid);
} catch (final IllegalArgumentException iae) {
@@ -461,6 +471,8 @@ public class RuleResource {
// get the web context
final NiFiWebConfigurationContext nifiWebContext = (NiFiWebConfigurationContext) servletContext.getAttribute("nifi-web-configuration-context");
+ // get the variable registry
+ final VariableRegistry variableRegistry = nifiWebContext.getVariableRegistry();
// ensure the rule has been specified
if (requestEntity == null || requestEntity.getRule() == null) {
@@ -497,7 +509,7 @@ public class RuleResource {
requestEntity.getProcessorId(), requestEntity.getRevision(), requestEntity.getClientId());
// load the criteria
- final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory();
+ final UpdateAttributeModelFactory factory = new UpdateAttributeModelFactory(variableRegistry);
final Criteria criteria = getCriteria(nifiWebContext, requestContext);
// attempt to locate the rule
[2/5] nifi git commit: NIFI-1974 - initial commit variable registry
for custom property support
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
index 11d1b51..a0a614c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
@@ -34,6 +34,7 @@ import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.events.BulletinFactory;
import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinRepository;
import org.apache.nifi.reporting.EventAccess;
@@ -50,16 +51,18 @@ public class StandardReportingContext implements ReportingContext, ControllerSer
private final ControllerServiceProvider serviceProvider;
private final Map<PropertyDescriptor, String> properties;
private final Map<PropertyDescriptor, PreparedQuery> preparedQueries;
+ private final VariableRegistry variableRegistry;
public StandardReportingContext(final FlowController flowController, final BulletinRepository bulletinRepository,
- final Map<PropertyDescriptor, String> properties, final ControllerServiceProvider serviceProvider, final ReportingTask reportingTask) {
+ final Map<PropertyDescriptor, String> properties, final ControllerServiceProvider serviceProvider, final ReportingTask reportingTask,
+ final VariableRegistry variableRegistry) {
this.flowController = flowController;
this.eventAccess = flowController;
this.bulletinRepository = bulletinRepository;
this.properties = Collections.unmodifiableMap(properties);
this.serviceProvider = serviceProvider;
this.reportingTask = reportingTask;
-
+ this.variableRegistry = variableRegistry;
preparedQueries = new HashMap<>();
for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) {
final PropertyDescriptor desc = entry.getKey();
@@ -106,7 +109,7 @@ public class StandardReportingContext implements ReportingContext, ControllerSer
@Override
public PropertyValue getProperty(final PropertyDescriptor property) {
final String configuredValue = properties.get(property);
- return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this, preparedQueries.get(property));
+ return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this, preparedQueries.get(property), variableRegistry);
}
@Override
@@ -148,4 +151,5 @@ public class StandardReportingContext implements ReportingContext, ControllerSer
public StateManager getStateManager() {
return flowController.getStateManagerProvider().getStateManager(reportingTask.getIdentifier());
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
index 0131a95..312d99c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
@@ -112,4 +112,5 @@ public class StandardReportingInitializationContext implements ReportingInitiali
public ComponentLog getLogger() {
return logger;
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java
index 1a40e8a..c5ce875 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java
@@ -20,6 +20,7 @@ import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ReportingTaskNode;
import org.apache.nifi.controller.ValidationContextFactory;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.ReportingContext;
import org.apache.nifi.reporting.ReportingTask;
@@ -28,13 +29,14 @@ public class StandardReportingTaskNode extends AbstractReportingTaskNode impleme
private final FlowController flowController;
public StandardReportingTaskNode(final ReportingTask reportingTask, final String id, final FlowController controller,
- final ProcessScheduler processScheduler, final ValidationContextFactory validationContextFactory) {
- super(reportingTask, id, controller, processScheduler, validationContextFactory);
+ final ProcessScheduler processScheduler, final ValidationContextFactory validationContextFactory,
+ final VariableRegistry variableRegistry) {
+ super(reportingTask, id, controller, processScheduler, validationContextFactory, variableRegistry);
this.flowController = controller;
}
@Override
public ReportingContext getReportingContext() {
- return new StandardReportingContext(flowController, flowController.getBulletinRepository(), getProperties(), flowController, getReportingTask());
+ return new StandardReportingContext(flowController, flowController.getBulletinRepository(), getProperties(), flowController, getReportingTask(), variableRegistry);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
index 228af70..af31722 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
@@ -45,6 +45,7 @@ import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.util.Connectables;
import org.apache.nifi.util.FormatUtils;
import org.apache.nifi.util.ReflectionUtils;
@@ -61,6 +62,7 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent {
private final ProcessContextFactory contextFactory;
private final AtomicInteger maxThreadCount;
private final StringEncryptor encryptor;
+ private final VariableRegistry variableRegistry;
private volatile String adminYieldDuration = "1 sec";
@@ -68,7 +70,8 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent {
private final ConcurrentMap<Connectable, ScheduleState> scheduleStates = new ConcurrentHashMap<>();
public EventDrivenSchedulingAgent(final FlowEngine flowEngine, final ControllerServiceProvider serviceProvider, final StateManagerProvider stateManagerProvider,
- final EventDrivenWorkerQueue workerQueue, final ProcessContextFactory contextFactory, final int maxThreadCount, final StringEncryptor encryptor) {
+ final EventDrivenWorkerQueue workerQueue, final ProcessContextFactory contextFactory, final int maxThreadCount, final StringEncryptor encryptor,
+ final VariableRegistry variableRegistry) {
super(flowEngine);
this.serviceProvider = serviceProvider;
this.stateManagerProvider = stateManagerProvider;
@@ -76,6 +79,7 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent {
this.contextFactory = contextFactory;
this.maxThreadCount = new AtomicInteger(maxThreadCount);
this.encryptor = encryptor;
+ this.variableRegistry = variableRegistry;
for (int i = 0; i < maxThreadCount; i++) {
final Runnable eventDrivenTask = new EventDrivenTask(workerQueue);
@@ -185,7 +189,7 @@ public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent {
if (connectable instanceof ProcessorNode) {
final ProcessorNode procNode = (ProcessorNode) connectable;
final StandardProcessContext standardProcessContext = new StandardProcessContext(procNode, serviceProvider,
- encryptor, getStateManager(connectable.getIdentifier()));
+ encryptor, getStateManager(connectable.getIdentifier()), variableRegistry);
final long runNanos = procNode.getRunDuration(TimeUnit.NANOSECONDS);
final ProcessSessionFactory sessionFactory;
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
index 3f19d28..34e7989 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
@@ -38,6 +38,7 @@ import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.processor.StandardProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.util.FormatUtils;
import org.quartz.CronExpression;
import org.slf4j.Logger;
@@ -50,15 +51,18 @@ public class QuartzSchedulingAgent extends AbstractSchedulingAgent {
private final FlowController flowController;
private final ProcessContextFactory contextFactory;
private final StringEncryptor encryptor;
+ private final VariableRegistry variableRegistry;
private volatile String adminYieldDuration = "1 sec";
private final Map<Object, List<AtomicBoolean>> canceledTriggers = new HashMap<>();
- public QuartzSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor enryptor) {
+ public QuartzSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor enryptor,
+ final VariableRegistry variableRegistry) {
super(flowEngine);
this.flowController = flowController;
this.contextFactory = contextFactory;
this.encryptor = enryptor;
+ this.variableRegistry = variableRegistry;
}
private StateManager getStateManager(final String componentId) {
@@ -141,7 +145,7 @@ public class QuartzSchedulingAgent extends AbstractSchedulingAgent {
if (connectable.getConnectableType() == ConnectableType.PROCESSOR) {
final ProcessorNode procNode = (ProcessorNode) connectable;
- final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier()));
+ final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier()), variableRegistry);
ContinuallyRunProcessorTask runnableTask = new ContinuallyRunProcessorTask(this, procNode, flowController, contextFactory, scheduleState, standardProcContext);
continuallyRunTask = runnableTask;
} else {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
index e8b2146..86dac7b 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
@@ -53,6 +53,7 @@ import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardProcessContext;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.ReportingTask;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.FormatUtils;
@@ -83,13 +84,15 @@ public final class StandardProcessScheduler implements ProcessScheduler {
private final ScheduledExecutorService componentMonitoringThreadPool = new FlowEngine(8, "StandardProcessScheduler", true);
private final StringEncryptor encryptor;
+ private final VariableRegistry variableRegistry;
public StandardProcessScheduler(final Heartbeater heartbeater, final ControllerServiceProvider controllerServiceProvider, final StringEncryptor encryptor,
- final StateManagerProvider stateManagerProvider) {
+ final StateManagerProvider stateManagerProvider, final VariableRegistry variableRegistry) {
this.heartbeater = heartbeater;
this.controllerServiceProvider = controllerServiceProvider;
this.encryptor = encryptor;
this.stateManagerProvider = stateManagerProvider;
+ this.variableRegistry = variableRegistry;
administrativeYieldDuration = NiFiProperties.getInstance().getAdministrativeYieldDuration();
administrativeYieldMillis = FormatUtils.getTimeDuration(administrativeYieldDuration, TimeUnit.MILLISECONDS);
@@ -295,7 +298,7 @@ public final class StandardProcessScheduler implements ProcessScheduler {
@Override
public synchronized void startProcessor(final ProcessorNode procNode) {
StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider,
- this.encryptor, getStateManager(procNode.getIdentifier()));
+ this.encryptor, getStateManager(procNode.getIdentifier()), variableRegistry);
final ScheduleState scheduleState = getScheduleState(requireNonNull(procNode));
SchedulingAgentCallback callback = new SchedulingAgentCallback() {
@@ -330,7 +333,7 @@ public final class StandardProcessScheduler implements ProcessScheduler {
@Override
public synchronized void stopProcessor(final ProcessorNode procNode) {
StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider,
- this.encryptor, getStateManager(procNode.getIdentifier()));
+ this.encryptor, getStateManager(procNode.getIdentifier()), variableRegistry);
final ScheduleState state = getScheduleState(procNode);
procNode.stop(this.componentLifeCycleThreadPool, processContext, new Callable<Boolean>() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
index 0436e21..f94beff 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
@@ -37,6 +37,7 @@ import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.StandardProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.util.FormatUtils;
import org.apache.nifi.util.NiFiProperties;
import org.slf4j.Logger;
@@ -50,14 +51,17 @@ public class TimerDrivenSchedulingAgent extends AbstractSchedulingAgent {
private final FlowController flowController;
private final ProcessContextFactory contextFactory;
private final StringEncryptor encryptor;
+ private final VariableRegistry variableRegistry;
private volatile String adminYieldDuration = "1 sec";
- public TimerDrivenSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor encryptor) {
+ public TimerDrivenSchedulingAgent(final FlowController flowController, final FlowEngine flowEngine, final ProcessContextFactory contextFactory, final StringEncryptor encryptor,
+ final VariableRegistry variableRegistry) {
super(flowEngine);
this.flowController = flowController;
this.contextFactory = contextFactory;
this.encryptor = encryptor;
+ this.variableRegistry = variableRegistry;
final String boredYieldDuration = NiFiProperties.getInstance().getBoredYieldDuration();
try {
@@ -100,7 +104,7 @@ public class TimerDrivenSchedulingAgent extends AbstractSchedulingAgent {
// Determine the task to run and create it.
if (connectable.getConnectableType() == ConnectableType.PROCESSOR) {
final ProcessorNode procNode = (ProcessorNode) connectable;
- final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier()));
+ final StandardProcessContext standardProcContext = new StandardProcessContext(procNode, flowController, encryptor, getStateManager(connectable.getIdentifier()), variableRegistry);
final ContinuallyRunProcessorTask runnableTask = new ContinuallyRunProcessorTask(this, procNode, flowController,
contextFactory, scheduleState, standardProcContext);
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java
index d57e61f..61db819 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardConfigurationContext.java
@@ -28,6 +28,7 @@ import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ConfiguredComponent;
import org.apache.nifi.controller.ControllerServiceLookup;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.util.FormatUtils;
public class StandardConfigurationContext implements ConfigurationContext {
@@ -35,13 +36,17 @@ public class StandardConfigurationContext implements ConfigurationContext {
private final ConfiguredComponent component;
private final ControllerServiceLookup serviceLookup;
private final Map<PropertyDescriptor, PreparedQuery> preparedQueries;
+ private final VariableRegistry variableRegistry;
private final String schedulingPeriod;
private final Long schedulingNanos;
- public StandardConfigurationContext(final ConfiguredComponent component, final ControllerServiceLookup serviceLookup, final String schedulingPeriod) {
+ public StandardConfigurationContext(final ConfiguredComponent component, final ControllerServiceLookup serviceLookup, final String schedulingPeriod,
+ final VariableRegistry variableRegistry) {
this.component = component;
this.serviceLookup = serviceLookup;
this.schedulingPeriod = schedulingPeriod;
+ this.variableRegistry = variableRegistry;
+
if (schedulingPeriod == null) {
schedulingNanos = null;
} else {
@@ -68,7 +73,7 @@ public class StandardConfigurationContext implements ConfigurationContext {
@Override
public PropertyValue getProperty(final PropertyDescriptor property) {
final String configuredValue = component.getProperty(property);
- return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceLookup, preparedQueries.get(property));
+ return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceLookup, preparedQueries.get(property), variableRegistry);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
index 482dabf..f861f6b 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
@@ -87,4 +87,5 @@ public class StandardControllerServiceInitializationContext implements Controlle
public StateManager getStateManager() {
return stateManager;
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
index 3f24ff1..83c35ad 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
@@ -43,6 +43,7 @@ import org.apache.nifi.controller.exception.ComponentLifeCycleException;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.SimpleProcessLogger;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +55,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
private final ControllerService proxedControllerService;
private final ControllerService implementation;
private final ControllerServiceProvider serviceProvider;
-
+ private final VariableRegistry variableRegistry;
private final AtomicReference<ControllerServiceState> stateRef = new AtomicReference<>(ControllerServiceState.DISABLED);
private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
@@ -67,12 +68,15 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
private final AtomicBoolean active;
public StandardControllerServiceNode(final ControllerService proxiedControllerService, final ControllerService implementation, final String id,
- final ValidationContextFactory validationContextFactory, final ControllerServiceProvider serviceProvider) {
+ final ValidationContextFactory validationContextFactory, final ControllerServiceProvider serviceProvider,
+ final VariableRegistry variableRegistry) {
super(implementation, id, validationContextFactory, serviceProvider);
this.proxedControllerService = proxiedControllerService;
this.implementation = implementation;
this.serviceProvider = serviceProvider;
this.active = new AtomicBoolean();
+ this.variableRegistry = variableRegistry;
+
}
@Override
@@ -141,7 +145,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
@SuppressWarnings("deprecation")
private void onConfigured() {
try (final NarCloseable x = NarCloseable.withNarLoader()) {
- final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceProvider, null);
+ final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceProvider, null, variableRegistry);
ReflectionUtils.invokeMethodsWithAnnotation(OnConfigured.class, implementation, configContext);
} catch (final Exception e) {
throw new ComponentLifeCycleException("Failed to invoke On-Configured Lifecycle methods of " + implementation, e);
@@ -277,7 +281,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
final Heartbeater heartbeater) {
if (this.stateRef.compareAndSet(ControllerServiceState.DISABLED, ControllerServiceState.ENABLING)) {
this.active.set(true);
- final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null);
+ final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null, variableRegistry);
scheduler.execute(new Runnable() {
@Override
public void run() {
@@ -342,7 +346,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
}
if (this.stateRef.compareAndSet(ControllerServiceState.ENABLED, ControllerServiceState.DISABLING)) {
- final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null);
+ final ConfigurationContext configContext = new StandardConfigurationContext(this, this.serviceProvider, null, variableRegistry);
scheduler.execute(new Runnable() {
@Override
public void run() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index 77dc87e..d0ad127 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -56,6 +56,7 @@ import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardValidationContextFactory;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.BulletinRepository;
import org.apache.nifi.reporting.Severity;
import org.apache.nifi.util.ObjectHolder;
@@ -72,6 +73,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
private static final Set<Method> validDisabledMethods;
private final BulletinRepository bulletinRepo;
private final StateManagerProvider stateManagerProvider;
+ private final VariableRegistry variableRegistry;
static {
// methods that are okay to be called when the service is disabled.
@@ -85,13 +87,15 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
validDisabledMethods = Collections.unmodifiableSet(validMethods);
}
- public StandardControllerServiceProvider(final ProcessScheduler scheduler, final BulletinRepository bulletinRepo, final StateManagerProvider stateManagerProvider) {
+ public StandardControllerServiceProvider(final ProcessScheduler scheduler, final BulletinRepository bulletinRepo, final StateManagerProvider stateManagerProvider,
+ final VariableRegistry variableRegistry) {
// the following 2 maps must be updated atomically, but we do not lock around them because they are modified
// only in the createControllerService method, and both are modified before the method returns
this.controllerServices = new ConcurrentHashMap<>();
this.processScheduler = scheduler;
this.bulletinRepo = bulletinRepo;
this.stateManagerProvider = stateManagerProvider;
+ this.variableRegistry = variableRegistry;
}
private Class<?>[] getInterfaces(final Class<?> cls) {
@@ -181,9 +185,9 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
final ComponentLog serviceLogger = new SimpleProcessLogger(id, originalService);
originalService.initialize(new StandardControllerServiceInitializationContext(id, serviceLogger, this, getStateManager(id)));
- final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this);
+ final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this, variableRegistry);
- final ControllerServiceNode serviceNode = new StandardControllerServiceNode(proxiedService, originalService, id, validationContextFactory, this);
+ final ControllerServiceNode serviceNode = new StandardControllerServiceNode(proxiedService, originalService, id, validationContextFactory, this, variableRegistry);
serviceNodeHolder.set(serviceNode);
serviceNode.setName(rawClass.getSimpleName());
@@ -481,7 +485,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
serviceNode.verifyCanDelete();
try (final NarCloseable x = NarCloseable.withNarLoader()) {
- final ConfigurationContext configurationContext = new StandardConfigurationContext(serviceNode, this, null);
+ final ConfigurationContext configurationContext = new StandardConfigurationContext(serviceNode, this, null, variableRegistry);
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnRemoved.class, serviceNode.getControllerServiceImplementation(), configurationContext);
}
@@ -615,4 +619,5 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
public void verifyCanStopReferencingComponents(final ControllerServiceNode serviceNode) {
// we can always stop referencing components
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java
index ef46f55..385f93a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/manager/StandardStateManagerProvider.java
@@ -46,11 +46,12 @@ import org.apache.nifi.controller.state.config.StateProviderConfiguration;
import org.apache.nifi.framework.security.util.SslContextFactory;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.processor.StandardValidationContext;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.util.NiFiProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class StandardStateManagerProvider implements StateManagerProvider {
+public class StandardStateManagerProvider implements StateManagerProvider{
private static final Logger logger = LoggerFactory.getLogger(StandardStateManagerProvider.class);
private final ConcurrentMap<String, StateManager> stateManagers = new ConcurrentHashMap<>();
@@ -62,12 +63,12 @@ public class StandardStateManagerProvider implements StateManagerProvider {
this.clusterStateProvider = clusterStateProvider;
}
- public static StateManagerProvider create(final NiFiProperties properties) throws ConfigParseException, IOException {
- final StateProvider localProvider = createLocalStateProvider(properties);
+ public static StateManagerProvider create(final NiFiProperties properties, final VariableRegistry variableRegistry) throws ConfigParseException, IOException {
+ final StateProvider localProvider = createLocalStateProvider(properties,variableRegistry);
final StateProvider clusterProvider;
if (properties.isNode()) {
- clusterProvider = createClusteredStateProvider(properties);
+ clusterProvider = createClusteredStateProvider(properties,variableRegistry);
} else {
clusterProvider = null;
}
@@ -75,25 +76,27 @@ public class StandardStateManagerProvider implements StateManagerProvider {
return new StandardStateManagerProvider(localProvider, clusterProvider);
}
- private static StateProvider createLocalStateProvider(final NiFiProperties properties) throws IOException, ConfigParseException {
+ private static StateProvider createLocalStateProvider(final NiFiProperties properties, final VariableRegistry variableRegistry) throws IOException, ConfigParseException {
final File configFile = properties.getStateManagementConfigFile();
- return createStateProvider(configFile, Scope.LOCAL, properties);
+ return createStateProvider(configFile, Scope.LOCAL, properties, variableRegistry);
}
- private static StateProvider createClusteredStateProvider(final NiFiProperties properties) throws IOException, ConfigParseException {
+ private static StateProvider createClusteredStateProvider(final NiFiProperties properties, final VariableRegistry variableRegistry) throws IOException, ConfigParseException {
final File configFile = properties.getStateManagementConfigFile();
- return createStateProvider(configFile, Scope.CLUSTER, properties);
+ return createStateProvider(configFile, Scope.CLUSTER, properties, variableRegistry);
}
- private static StateProvider createStateProvider(final File configFile, final Scope scope, final NiFiProperties properties) throws ConfigParseException, IOException {
+ private static StateProvider createStateProvider(final File configFile, final Scope scope, final NiFiProperties properties,
+ final VariableRegistry variableRegistry) throws ConfigParseException, IOException {
final String providerId;
final String providerIdPropertyName;
final String providerDescription;
final String providerXmlElementName;
final String oppositeScopeXmlElementName;
+
switch (scope) {
case CLUSTER:
providerId = properties.getClusterStateProviderId();
@@ -166,17 +169,18 @@ public class StandardStateManagerProvider implements StateManagerProvider {
+ " is configured to use scope " + scope);
}
+ //create variable registry
final Map<PropertyDescriptor, PropertyValue> propertyMap = new HashMap<>();
final Map<PropertyDescriptor, String> propertyStringMap = new HashMap<>();
for (final PropertyDescriptor descriptor : provider.getPropertyDescriptors()) {
- propertyMap.put(descriptor, new StandardPropertyValue(descriptor.getDefaultValue(), null));
+ propertyMap.put(descriptor, new StandardPropertyValue(descriptor.getDefaultValue(),null, variableRegistry));
propertyStringMap.put(descriptor, descriptor.getDefaultValue());
}
for (final Map.Entry<String, String> entry : providerConfig.getProperties().entrySet()) {
final PropertyDescriptor descriptor = provider.getPropertyDescriptor(entry.getKey());
propertyStringMap.put(descriptor, entry.getValue());
- propertyMap.put(descriptor, new StandardPropertyValue(entry.getValue(), null));
+ propertyMap.put(descriptor, new StandardPropertyValue(entry.getValue(),null, variableRegistry));
}
final SSLContext sslContext = SslContextFactory.createSslContext(properties, false);
@@ -186,7 +190,7 @@ public class StandardStateManagerProvider implements StateManagerProvider {
provider.initialize(initContext);
}
- final ValidationContext validationContext = new StandardValidationContext(null, propertyStringMap, null);
+ final ValidationContext validationContext = new StandardValidationContext(null, propertyStringMap, null, variableRegistry);
final Collection<ValidationResult> results = provider.validate(validationContext);
final StringBuilder validationFailures = new StringBuilder();
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index 4646d55..e9032eb 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -59,6 +59,7 @@ import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.logging.LogRepositoryFactory;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.StandardProcessContext;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.remote.RootGroupPort;
import org.apache.nifi.util.NiFiProperties;
@@ -87,6 +88,7 @@ public final class StandardProcessGroup implements ProcessGroup {
private final Map<String, ProcessorNode> processors = new HashMap<>();
private final Map<String, Funnel> funnels = new HashMap<>();
private final StringEncryptor encryptor;
+ private final VariableRegistry variableRegistry;
private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
private final Lock readLock = rwLock.readLock();
@@ -95,7 +97,8 @@ public final class StandardProcessGroup implements ProcessGroup {
private static final Logger LOG = LoggerFactory.getLogger(StandardProcessGroup.class);
public StandardProcessGroup(final String id, final ControllerServiceProvider serviceProvider, final StandardProcessScheduler scheduler,
- final NiFiProperties nifiProps, final StringEncryptor encryptor, final FlowController flowController) {
+ final NiFiProperties nifiProps, final StringEncryptor encryptor, final FlowController flowController,
+ final VariableRegistry variableRegistry) {
this.id = id;
this.controllerServiceProvider = serviceProvider;
this.parent = new AtomicReference<>();
@@ -103,6 +106,7 @@ public final class StandardProcessGroup implements ProcessGroup {
this.comments = new AtomicReference<>("");
this.encryptor = encryptor;
this.flowController = flowController;
+ this.variableRegistry = variableRegistry;
name = new AtomicReference<>();
position = new AtomicReference<>(new Position(0D, 0D));
@@ -342,7 +346,7 @@ public final class StandardProcessGroup implements ProcessGroup {
private void shutdown(final ProcessGroup procGroup) {
for (final ProcessorNode node : procGroup.getProcessors()) {
try (final NarCloseable x = NarCloseable.withNarLoader()) {
- final StandardProcessContext processContext = new StandardProcessContext(node, controllerServiceProvider, encryptor, getStateManager(node.getIdentifier()));
+ final StandardProcessContext processContext = new StandardProcessContext(node, controllerServiceProvider, encryptor, getStateManager(node.getIdentifier()), variableRegistry);
ReflectionUtils.quietlyInvokeMethodsWithAnnotations(OnShutdown.class, org.apache.nifi.processor.annotation.OnShutdown.class, node.getProcessor(), processContext);
}
}
@@ -708,7 +712,7 @@ public final class StandardProcessGroup implements ProcessGroup {
}
try (final NarCloseable x = NarCloseable.withNarLoader()) {
- final StandardProcessContext processContext = new StandardProcessContext(processor, controllerServiceProvider, encryptor, getStateManager(processor.getIdentifier()));
+ final StandardProcessContext processContext = new StandardProcessContext(processor, controllerServiceProvider, encryptor, getStateManager(processor.getIdentifier()), variableRegistry);
ReflectionUtils.quietlyInvokeMethodsWithAnnotations(OnRemoved.class, org.apache.nifi.processor.annotation.OnRemoved.class, processor.getProcessor(), processContext);
} catch (final Exception e) {
throw new ComponentLifeCycleException("Failed to invoke 'OnRemoved' methods of " + processor, e);
@@ -2097,4 +2101,5 @@ public final class StandardProcessGroup implements ProcessGroup {
readLock.unlock();
}
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
index dae7d0c..09c68c2 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
@@ -36,6 +36,7 @@ import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.encrypt.StringEncryptor;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.util.Connectables;
public class StandardProcessContext implements ProcessContext, ControllerServiceLookup {
@@ -45,12 +46,15 @@ public class StandardProcessContext implements ProcessContext, ControllerService
private final Map<PropertyDescriptor, PreparedQuery> preparedQueries;
private final StringEncryptor encryptor;
private final StateManager stateManager;
+ private final VariableRegistry variableRegistry;
- public StandardProcessContext(final ProcessorNode processorNode, final ControllerServiceProvider controllerServiceProvider, final StringEncryptor encryptor, final StateManager stateManager) {
+ public StandardProcessContext(final ProcessorNode processorNode, final ControllerServiceProvider controllerServiceProvider, final StringEncryptor encryptor, final StateManager stateManager,
+ final VariableRegistry variableRegistry) {
this.procNode = processorNode;
this.controllerServiceProvider = controllerServiceProvider;
this.encryptor = encryptor;
this.stateManager = stateManager;
+ this.variableRegistry = variableRegistry;
preparedQueries = new HashMap<>();
for (final Map.Entry<PropertyDescriptor, String> entry : procNode.getProperties().entrySet()) {
@@ -86,12 +90,12 @@ public class StandardProcessContext implements ProcessContext, ControllerService
final String setPropertyValue = procNode.getProperty(descriptor);
final String propValue = (setPropertyValue == null) ? descriptor.getDefaultValue() : setPropertyValue;
- return new StandardPropertyValue(propValue, this, preparedQueries.get(descriptor));
+ return new StandardPropertyValue(propValue, this, preparedQueries.get(descriptor), variableRegistry);
}
@Override
public PropertyValue newPropertyValue(final String rawValue) {
- return new StandardPropertyValue(rawValue, this, Query.prepare(rawValue));
+ return new StandardPropertyValue(rawValue, this, Query.prepare(rawValue), variableRegistry);
}
@Override
@@ -221,4 +225,5 @@ public class StandardProcessContext implements ProcessContext, ControllerService
public String getName() {
return procNode.getName();
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java
index c24e146..5489221 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContext.java
@@ -35,6 +35,7 @@ import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.expression.ExpressionLanguageCompiler;
+import org.apache.nifi.registry.VariableRegistry;
public class StandardValidationContext implements ValidationContext {
@@ -44,20 +45,23 @@ public class StandardValidationContext implements ValidationContext {
private final Map<String, Boolean> expressionLanguageSupported;
private final String annotationData;
private final Set<String> serviceIdentifiersToNotValidate;
+ private final VariableRegistry variableRegistry;
- public StandardValidationContext(final ControllerServiceProvider controllerServiceProvider, final Map<PropertyDescriptor, String> properties, final String annotationData) {
- this(controllerServiceProvider, Collections.<String>emptySet(), properties, annotationData);
+ public StandardValidationContext(final ControllerServiceProvider controllerServiceProvider, final Map<PropertyDescriptor, String> properties, final String annotationData,
+ final VariableRegistry variableRegistry) {
+ this(controllerServiceProvider, Collections.<String>emptySet(), properties, annotationData, variableRegistry);
}
public StandardValidationContext(
final ControllerServiceProvider controllerServiceProvider,
final Set<String> serviceIdentifiersToNotValidate,
final Map<PropertyDescriptor, String> properties,
- final String annotationData) {
+ final String annotationData, VariableRegistry variableRegistry) {
this.controllerServiceProvider = controllerServiceProvider;
this.properties = new HashMap<>(properties);
this.annotationData = annotationData;
this.serviceIdentifiersToNotValidate = serviceIdentifiersToNotValidate;
+ this.variableRegistry = variableRegistry;
preparedQueries = new HashMap<>(properties.size());
for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) {
@@ -79,24 +83,24 @@ public class StandardValidationContext implements ValidationContext {
@Override
public PropertyValue newPropertyValue(final String rawValue) {
- return new StandardPropertyValue(rawValue, controllerServiceProvider, Query.prepare(rawValue));
+ return new StandardPropertyValue(rawValue, controllerServiceProvider, Query.prepare(rawValue), variableRegistry);
}
@Override
public ExpressionLanguageCompiler newExpressionLanguageCompiler() {
- return new StandardExpressionLanguageCompiler();
+ return new StandardExpressionLanguageCompiler(variableRegistry);
}
@Override
public ValidationContext getControllerServiceValidationContext(final ControllerService controllerService) {
final ControllerServiceNode serviceNode = controllerServiceProvider.getControllerServiceNode(controllerService.getIdentifier());
- return new StandardValidationContext(controllerServiceProvider, serviceNode.getProperties(), serviceNode.getAnnotationData());
+ return new StandardValidationContext(controllerServiceProvider, serviceNode.getProperties(), serviceNode.getAnnotationData(), variableRegistry);
}
@Override
public PropertyValue getProperty(final PropertyDescriptor property) {
final String configuredValue = properties.get(property);
- return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, controllerServiceProvider, preparedQueries.get(property));
+ return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, controllerServiceProvider, preparedQueries.get(property), variableRegistry);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java
index c3df987..e6b21e0 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardValidationContextFactory.java
@@ -23,22 +23,25 @@ import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.controller.ValidationContextFactory;
import org.apache.nifi.controller.service.ControllerServiceProvider;
+import org.apache.nifi.registry.VariableRegistry;
public class StandardValidationContextFactory implements ValidationContextFactory {
private final ControllerServiceProvider serviceProvider;
+ private final VariableRegistry variableRegistry;
- public StandardValidationContextFactory(final ControllerServiceProvider serviceProvider) {
+ public StandardValidationContextFactory(final ControllerServiceProvider serviceProvider, final VariableRegistry variableRegistry) {
this.serviceProvider = serviceProvider;
+ this.variableRegistry = variableRegistry;
}
@Override
public ValidationContext newValidationContext(final Map<PropertyDescriptor, String> properties, final String annotationData) {
- return new StandardValidationContext(serviceProvider, properties, annotationData);
+ return new StandardValidationContext(serviceProvider, properties, annotationData, variableRegistry);
}
@Override
public ValidationContext newValidationContext(final Set<String> serviceIdentifiersToNotValidate, final Map<PropertyDescriptor, String> properties, final String annotationData) {
- return new StandardValidationContext(serviceProvider, serviceIdentifiersToNotValidate, properties, annotationData);
+ return new StandardValidationContext(serviceProvider, serviceIdentifiersToNotValidate, properties, annotationData, variableRegistry);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
index 3b33478..c2d5cac 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
@@ -59,6 +59,8 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.StandardValidationContextFactory;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.reporting.AbstractReportingTask;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.reporting.ReportingContext;
@@ -74,12 +76,13 @@ public class TestStandardProcessScheduler {
private ReportingTaskNode taskNode = null;
private TestReportingTask reportingTask = null;
private StateManagerProvider stateMgrProvider = Mockito.mock(StateManagerProvider.class);
+ private VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
@Before
public void setup() throws InitializationException {
System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties");
this.refreshNiFiProperties();
- scheduler = new StandardProcessScheduler(Mockito.mock(Heartbeater.class), Mockito.mock(ControllerServiceProvider.class), null, stateMgrProvider);
+ scheduler = new StandardProcessScheduler(Mockito.mock(Heartbeater.class), Mockito.mock(ControllerServiceProvider.class), null, stateMgrProvider, variableRegistry);
scheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, Mockito.mock(SchedulingAgent.class));
reportingTask = new TestReportingTask();
@@ -87,8 +90,8 @@ public class TestStandardProcessScheduler {
Mockito.mock(ComponentLog.class), null);
reportingTask.initialize(config);
- final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(null);
- taskNode = new StandardReportingTaskNode(reportingTask, UUID.randomUUID().toString(), null, scheduler, validationContextFactory);
+ final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(null, variableRegistry);
+ taskNode = new StandardReportingTaskNode(reportingTask, UUID.randomUUID().toString(), null, scheduler, validationContextFactory, variableRegistry);
}
/**
@@ -119,10 +122,10 @@ public class TestStandardProcessScheduler {
public void testDisableControllerServiceWithProcessorTryingToStartUsingIt() throws InterruptedException {
final Processor proc = new ServiceReferencingProcessor();
- final ControllerServiceProvider serviceProvider = new StandardControllerServiceProvider(scheduler, null, Mockito.mock(StateManagerProvider.class));
+ final ControllerServiceProvider serviceProvider = new StandardControllerServiceProvider(scheduler, null, Mockito.mock(StateManagerProvider.class), variableRegistry);
final ControllerServiceNode service = serviceProvider.createControllerService(NoStartServiceImpl.class.getName(), "service", true);
final ProcessorNode procNode = new StandardProcessorNode(proc, UUID.randomUUID().toString(),
- new StandardValidationContextFactory(serviceProvider), scheduler, serviceProvider);
+ new StandardValidationContextFactory(serviceProvider, VariableRegistryUtils.createVariableRegistry()), scheduler, serviceProvider);
procNode.setProperty(ServiceReferencingProcessor.SERVICE_DESC.getName(), service.getIdentifier());
@@ -200,7 +203,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateServiceEnablementLogicHappensOnlyOnce() throws Exception {
final ProcessScheduler scheduler = createScheduler();
- StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
+ StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(),
"1", false);
assertFalse(serviceNode.isActive());
@@ -239,7 +242,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateDisabledServiceCantBeDisabled() throws Exception {
final ProcessScheduler scheduler = createScheduler();
- StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
+ StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(),
"1", false);
SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
@@ -277,7 +280,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateEnabledServiceCanOnlyBeDisabledOnce() throws Exception {
final ProcessScheduler scheduler = createScheduler();
- StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
+ StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
final ControllerServiceNode serviceNode = provider.createControllerService(SimpleTestService.class.getName(),
"1", false);
SimpleTestService ts = (SimpleTestService) serviceNode.getControllerServiceImplementation();
@@ -311,7 +314,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateDisablingOfTheFailedService() throws Exception {
final ProcessScheduler scheduler = createScheduler();
- StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
+ StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
final ControllerServiceNode serviceNode = provider.createControllerService(FailingService.class.getName(),
"1", false);
scheduler.enableControllerService(serviceNode);
@@ -342,7 +345,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateEnabledDisableMultiThread() throws Exception {
final ProcessScheduler scheduler = createScheduler();
- StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
+ StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 200; i++) {
final ControllerServiceNode serviceNode = provider
@@ -385,7 +388,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateNeverEnablingServiceCanStillBeDisabled() throws Exception {
final ProcessScheduler scheduler = createScheduler();
- StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
+ StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(),
"1", false);
LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation();
@@ -410,7 +413,7 @@ public class TestStandardProcessScheduler {
@Test
public void validateLongEnablingServiceCanStillBeDisabled() throws Exception {
final ProcessScheduler scheduler = createScheduler();
- StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider);
+ StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
final ControllerServiceNode serviceNode = provider.createControllerService(LongEnablingService.class.getName(),
"1", false);
LongEnablingService ts = (LongEnablingService) serviceNode.getControllerServiceImplementation();
@@ -507,6 +510,6 @@ public class TestStandardProcessScheduler {
}
private ProcessScheduler createScheduler() {
- return new StandardProcessScheduler(mock(Heartbeater.class), null, null, stateMgrProvider);
+ return new StandardProcessScheduler(mock(Heartbeater.class), null, null, stateMgrProvider, VariableRegistryUtils.createVariableRegistry());
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
index b9c0f7f..2390c79 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/StandardControllerServiceProviderTest.java
@@ -22,6 +22,9 @@ import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.StandardFlowServiceTest;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarClassLoaders;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryFactory;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.NiFiProperties;
import org.junit.Before;
@@ -33,6 +36,7 @@ public class StandardControllerServiceProviderTest {
private ControllerService proxied;
private ControllerService implementation;
+ private static VariableRegistry variableRegistry;
@BeforeClass
public static void setupSuite() throws Exception {
@@ -40,6 +44,10 @@ public class StandardControllerServiceProviderTest {
NiFiProperties properties = NiFiProperties.getInstance();
NarClassLoaders.load(properties);
ExtensionManager.discoverExtensions();
+ VariableRegistry propRegistry = VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths());
+ variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ variableRegistry.addRegistry(propRegistry);
+
}
@Before
@@ -67,7 +75,7 @@ public class StandardControllerServiceProviderTest {
@Override
public void onComponentRemoved(String componentId) {
}
- });
+ }, variableRegistry);
ControllerServiceNode node = provider.createControllerService(clazz, id, true);
proxied = node.getProxiedControllerService();
implementation = node.getControllerServiceImplementation();
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
index 5abefda..a11610d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
@@ -41,6 +41,8 @@ import org.apache.nifi.controller.service.mock.ServiceB;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.StandardProcessGroup;
import org.apache.nifi.processor.StandardValidationContextFactory;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -70,6 +72,8 @@ public class TestStandardControllerServiceProvider {
}
};
+ private static VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+
@BeforeClass
public static void setNiFiProps() {
System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties");
@@ -77,13 +81,13 @@ public class TestStandardControllerServiceProvider {
private StandardProcessScheduler createScheduler() {
final Heartbeater heartbeater = Mockito.mock(Heartbeater.class);
- return new StandardProcessScheduler(heartbeater, null, null, stateManagerProvider);
+ return new StandardProcessScheduler(heartbeater, null, null, stateManagerProvider, variableRegistry);
}
@Test
public void testDisableControllerService() {
final ProcessScheduler scheduler = createScheduler();
- final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider);
+ final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, variableRegistry);
final ControllerServiceNode serviceNode = provider.createControllerService(ServiceB.class.getName(), "B", false);
provider.enableControllerService(serviceNode);
@@ -93,7 +97,7 @@ public class TestStandardControllerServiceProvider {
@Test(timeout=10000)
public void testEnableDisableWithReference() {
final ProcessScheduler scheduler = createScheduler();
- final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider);
+ final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, variableRegistry);
final ControllerServiceNode serviceNodeB = provider.createControllerService(ServiceB.class.getName(), "B", false);
final ControllerServiceNode serviceNodeA = provider.createControllerService(ServiceA.class.getName(), "A", false);
@@ -146,7 +150,7 @@ public class TestStandardControllerServiceProvider {
}
public void testEnableReferencingServicesGraph(ProcessScheduler scheduler) {
- final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider);
+ final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider, variableRegistry);
// build a graph of controller services with dependencies as such:
//
@@ -189,7 +193,7 @@ public class TestStandardControllerServiceProvider {
@Test
public void testOrderingOfServices() {
- final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider);
+ final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider, variableRegistry);
final ControllerServiceNode serviceNode1 = provider.createControllerService(ServiceA.class.getName(), "1", false);
final ControllerServiceNode serviceNode2 = provider.createControllerService(ServiceB.class.getName(), "2", false);
@@ -333,9 +337,9 @@ public class TestStandardControllerServiceProvider {
private ProcessorNode createProcessor(final StandardProcessScheduler scheduler, final ControllerServiceProvider serviceProvider) {
final ProcessorNode procNode = new StandardProcessorNode(new DummyProcessor(), UUID.randomUUID().toString(),
- new StandardValidationContextFactory(serviceProvider), scheduler, serviceProvider);
+ new StandardValidationContextFactory(serviceProvider, null), scheduler, serviceProvider);
- final ProcessGroup group = new StandardProcessGroup(UUID.randomUUID().toString(), serviceProvider, scheduler, null, null, null);
+ final ProcessGroup group = new StandardProcessGroup(UUID.randomUUID().toString(), serviceProvider, scheduler, null, null, null, variableRegistry);
group.addProcessor(procNode);
procNode.setProcessGroup(group);
@@ -345,7 +349,7 @@ public class TestStandardControllerServiceProvider {
@Test
public void testEnableReferencingComponents() {
final StandardProcessScheduler scheduler = createScheduler();
- final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider);
+ final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(null, null, stateManagerProvider, variableRegistry);
final ControllerServiceNode serviceNode = provider.createControllerService(ServiceA.class.getName(), "1", false);
final ProcessorNode procNode = createProcessor(scheduler, provider);
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
index 9f7c4c9..d24441a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
@@ -32,6 +32,8 @@ import org.apache.nifi.components.state.StateProvider;
import org.apache.nifi.components.state.StateProviderInitializationContext;
import org.apache.nifi.controller.state.StateMapUpdate;
import org.apache.nifi.controller.state.providers.AbstractTestStateProvider;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.junit.After;
import org.junit.Before;
import org.wali.WriteAheadRepository;
@@ -43,9 +45,9 @@ public class TestWriteAheadLocalStateProvider extends AbstractTestStateProvider
@Before
public void setup() throws IOException {
provider = new WriteAheadLocalStateProvider();
-
+ final VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
final Map<PropertyDescriptor, PropertyValue> properties = new HashMap<>();
- properties.put(WriteAheadLocalStateProvider.PATH, new StandardPropertyValue("target/local-state-provider/" + UUID.randomUUID().toString(), null));
+ properties.put(WriteAheadLocalStateProvider.PATH, new StandardPropertyValue("target/local-state-provider/" + UUID.randomUUID().toString(), null, variableRegistry));
provider.initialize(new StateProviderInitializationContext() {
@Override
@@ -62,7 +64,7 @@ public class TestWriteAheadLocalStateProvider extends AbstractTestStateProvider
public PropertyValue getProperty(final PropertyDescriptor property) {
final PropertyValue prop = properties.get(property);
if (prop == null) {
- return new StandardPropertyValue(null, null);
+ return new StandardPropertyValue(null, null, variableRegistry);
}
return prop;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java
index 3542a09..dd7c871 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/zookeeper/TestZooKeeperStateProvider.java
@@ -36,6 +36,8 @@ import org.apache.nifi.components.state.StateProvider;
import org.apache.nifi.components.state.StateProviderInitializationContext;
import org.apache.nifi.components.state.exception.StateTooLargeException;
import org.apache.nifi.controller.state.providers.AbstractTestStateProvider;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.ZooDefs.Perms;
@@ -50,6 +52,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider {
private StateProvider provider;
private TestingServer zkServer;
+ private VariableRegistry variableRegistry;
private static final Map<PropertyDescriptor, String> defaultProperties = new HashMap<>();
@@ -68,6 +71,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider {
final Map<PropertyDescriptor, String> properties = new HashMap<>(defaultProperties);
properties.put(ZooKeeperStateProvider.CONNECTION_STRING, zkServer.getConnectString());
this.provider = createProvider(properties);
+ variableRegistry = VariableRegistryUtils.createVariableRegistry();
}
private void initializeProvider(final ZooKeeperStateProvider provider, final Map<PropertyDescriptor, String> properties) throws IOException {
@@ -81,7 +85,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider {
public Map<PropertyDescriptor, PropertyValue> getProperties() {
final Map<PropertyDescriptor, PropertyValue> propValueMap = new HashMap<>();
for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) {
- propValueMap.put(entry.getKey(), new StandardPropertyValue(entry.getValue(), null));
+ propValueMap.put(entry.getKey(), new StandardPropertyValue(entry.getValue(), null, variableRegistry));
}
return propValueMap;
}
@@ -89,7 +93,7 @@ public class TestZooKeeperStateProvider extends AbstractTestStateProvider {
@Override
public PropertyValue getProperty(final PropertyDescriptor property) {
final String prop = properties.get(property);
- return new StandardPropertyValue(prop, null);
+ return new StandardPropertyValue(prop, null, variableRegistry);
}
@Override
[5/5] nifi git commit: NIFI-197: Trim filenames and updated log
message. This closes #501
Posted by ma...@apache.org.
NIFI-197: Trim filenames and updated log message. This closes #501
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/2cdb1c19
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/2cdb1c19
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/2cdb1c19
Branch: refs/heads/0.x
Commit: 2cdb1c1900feb7cb92c97a888213846536c238e6
Parents: c600f15
Author: Mark Payne <ma...@hotmail.com>
Authored: Sat Jun 18 20:47:15 2016 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Sat Jun 18 20:47:15 2016 -0400
----------------------------------------------------------------------
.../org/apache/nifi/registry/FileVariableRegistry.java | 2 +-
.../nifi/registry/PropertiesVariableRegistry.java | 8 +-------
.../main/java/org/apache/nifi/util/NiFiProperties.java | 12 +++++++-----
.../java/org/apache/nifi/controller/FlowController.java | 7 +++++--
4 files changed, 14 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/2cdb1c19/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java b/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java
index 539f24e..bb977cb 100644
--- a/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java
@@ -49,7 +49,7 @@ abstract class FileVariableRegistry extends MultiMapVariableRegistry {
private void addVariables(Path ...paths) throws IOException{
- if(paths != null) {
+ if (paths != null) {
for (final Path path : paths) {
registry.addMap(convertFile(path.toFile()));
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2cdb1c19/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java b/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java
index bc5b15b..8a39f79 100644
--- a/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java
@@ -26,13 +26,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
class PropertiesVariableRegistry extends FileVariableRegistry {
- private static final Logger LOG = LoggerFactory.getLogger(PropertiesVariableRegistry.class);
-
PropertiesVariableRegistry(File... files) throws IOException{
super(files);
}
@@ -56,8 +51,7 @@ class PropertiesVariableRegistry extends FileVariableRegistry {
@Override
protected Map<String,String> convertFile(File file) throws IOException{
-
- try(final InputStream inStream = new BufferedInputStream(new FileInputStream(file))) {
+ try (final InputStream inStream = new BufferedInputStream(new FileInputStream(file))) {
Properties properties = new Properties();
properties.load(inStream);
return convertToMap(properties);
http://git-wip-us.apache.org/repos/asf/nifi/blob/2cdb1c19/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index ecef74b..e7f1661 100644
--- a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++ b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -1085,17 +1085,19 @@ public class NiFiProperties extends Properties {
public Path[] getVariableRegistryPropertiesPaths() {
final List<Path> vrPropertiesPaths = new ArrayList<>();
-
final String vrPropertiesFiles = getVariableRegistryProperties();
- if(!StringUtils.isEmpty(vrPropertiesFiles)) {
+ if(!StringUtils.isEmpty(vrPropertiesFiles)) {
final List<String> vrPropertiesFileList = Arrays.asList(vrPropertiesFiles.split(","));
- for(String propertiesFile : vrPropertiesFileList){
- vrPropertiesPaths.add(Paths.get(propertiesFile));
+ for (final String propertiesFile : vrPropertiesFileList) {
+ final String trimmed = propertiesFile.trim();
+ if (!trimmed.isEmpty()) {
+ vrPropertiesPaths.add(Paths.get(trimmed));
+ }
}
- return vrPropertiesPaths.toArray( new Path[vrPropertiesPaths.size()]);
+ return vrPropertiesPaths.toArray(new Path[vrPropertiesPaths.size()]);
} else {
return null;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/2cdb1c19/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 7c8763b..ae8e50f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -193,6 +193,7 @@ import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.nio.file.Path;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
@@ -595,11 +596,13 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
private static VariableRegistry createVariableRegistry(final NiFiProperties properties){
VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+
+ final Path[] registryPropPaths = properties.getVariableRegistryPropertiesPaths();
try {
- VariableRegistry customRegistry = VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths());
+ VariableRegistry customRegistry = VariableRegistryFactory.getPropertiesInstance(registryPropPaths);
variableRegistry.addRegistry(customRegistry);
} catch (IOException ioe){
- LOG.error("Exception thrown while attempting to add properties to registry",ioe);
+ LOG.error("Failed to load Variable Registry from " + registryPropPaths, ioe);
}
return variableRegistry;
[4/5] nifi git commit: NIFI-1974 - initial commit variable registry
for custom property support
Posted by ma...@apache.org.
NIFI-1974 - initial commit variable registry for custom property support
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/c600f150
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/c600f150
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/c600f150
Branch: refs/heads/0.x
Commit: c600f150aa78bf51b1bb163c83d90c9e47869b2e
Parents: 90c749d
Author: Yolanda M. Davis <yo...@gmail.com>
Authored: Sun Jun 5 21:17:37 2016 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Sat Jun 18 19:49:20 2016 -0400
----------------------------------------------------------------------
nifi-api/pom.xml | 4 +-
.../nifi/registry/FileVariableRegistry.java | 61 ++++++
.../java/org/apache/nifi/registry/MultiMap.java | 145 ++++++++++++++
.../nifi/registry/MultiMapVariableRegistry.java | 70 +++++++
.../registry/PropertiesVariableRegistry.java | 76 ++++++++
.../apache/nifi/registry/VariableRegistry.java | 52 +++++
.../nifi/registry/VariableRegistryFactory.java | 48 +++++
.../nifi/registry/VariableRegistryProvider.java | 23 +++
.../nifi/registry/VariableRegistryUtils.java | 63 ++++++
.../nifi/web/NiFiWebConfigurationContext.java | 5 +-
.../nifi/registry/TestVariableRegistry.java | 137 +++++++++++++
.../registry/TestVariableRegistryUtils.java | 123 ++++++++++++
.../TestVariableRegistry/foobar.properties | 16 ++
.../TestVariableRegistry/test.properties | 17 ++
.../bootstrap/NotificationServiceManager.java | 4 +-
.../NotificationValidationContext.java | 5 +-
.../expression/language/EmptyPreparedQuery.java | 35 +---
.../language/InvalidPreparedQuery.java | 33 +---
.../expression/language/PreparedQuery.java | 16 +-
.../attribute/expression/language/Query.java | 190 +------------------
.../language/StandardAttributeExpression.java | 9 +-
.../StandardExpressionLanguageCompiler.java | 9 +-
.../language/StandardPreparedQuery.java | 37 +---
.../language/StandardPropertyValue.java | 24 ++-
.../expression/language/QueryGroovyTest.groovy | 17 +-
.../expression/language/TestQuery.java | 112 +++++++----
.../language/TestStandardPreparedQuery.java | 7 +-
.../org/apache/nifi/util/NiFiProperties.java | 29 +++
.../nifi/util/MockConfigurationContext.java | 13 +-
...kControllerServiceInitializationContext.java | 2 +
.../nifi/util/MockControllerServiceLookup.java | 1 +
.../apache/nifi/util/MockProcessContext.java | 23 ++-
.../MockProcessorInitializationContext.java | 1 +
.../org/apache/nifi/util/MockPropertyValue.java | 18 +-
.../apache/nifi/util/MockReportingContext.java | 7 +-
.../apache/nifi/util/MockValidationContext.java | 15 +-
.../nifi/util/StandardProcessorTestRunner.java | 15 +-
.../java/org/apache/nifi/util/TestRunners.java | 3 +-
.../CurrentTestStandardProcessorTestRunner.java | 3 +-
.../nifi/util/TestMockProcessContext.java | 3 +-
.../util/TestStandardProcessorTestRunner.java | 9 +-
.../ambari/TestAmbariReportingTask.java | 15 +-
.../elasticsearch/TestFetchElasticsearch.java | 6 +-
.../manager/impl/ClusteredReportingContext.java | 7 +-
.../cluster/manager/impl/WebClusterManager.java | 33 +++-
.../reporting/ClusteredReportingTaskNode.java | 11 +-
.../apache/nifi/controller/FlowController.java | 46 +++--
.../reporting/AbstractReportingTaskNode.java | 13 +-
.../reporting/StandardReportingContext.java | 10 +-
.../StandardReportingInitializationContext.java | 1 +
.../reporting/StandardReportingTaskNode.java | 8 +-
.../scheduling/EventDrivenSchedulingAgent.java | 8 +-
.../scheduling/QuartzSchedulingAgent.java | 8 +-
.../scheduling/StandardProcessScheduler.java | 9 +-
.../scheduling/TimerDrivenSchedulingAgent.java | 8 +-
.../service/StandardConfigurationContext.java | 9 +-
...dControllerServiceInitializationContext.java | 1 +
.../service/StandardControllerServiceNode.java | 14 +-
.../StandardControllerServiceProvider.java | 13 +-
.../manager/StandardStateManagerProvider.java | 28 +--
.../nifi/groups/StandardProcessGroup.java | 11 +-
.../nifi/processor/StandardProcessContext.java | 11 +-
.../processor/StandardValidationContext.java | 18 +-
.../StandardValidationContextFactory.java | 9 +-
.../TestStandardProcessScheduler.java | 29 +--
.../StandardControllerServiceProviderTest.java | 10 +-
.../TestStandardControllerServiceProvider.java | 20 +-
.../local/TestWriteAheadLocalStateProvider.java | 8 +-
.../zookeeper/TestZooKeeperStateProvider.java | 8 +-
.../processor/TestStandardPropertyValue.java | 27 +--
.../src/main/resources/conf/nifi.properties | 6 +-
.../StandardNiFiWebConfigurationContext.java | 17 +-
.../web/controller/StandardSearchContext.java | 2 +-
.../processors/script/TestInvokeGroovy.java | 13 +-
.../TestSiteToSiteProvenanceReportingTask.java | 5 +-
.../standard/TestRouteOnAttribute.java | 8 +-
.../cache/server/TestServerAndClient.java | 13 +-
.../update/attributes/TestUpdateAttribute.java | 3 +
.../attributes/UpdateAttributeModelFactory.java | 9 +-
.../update/attributes/api/RuleResource.java | 20 +-
80 files changed, 1434 insertions(+), 541 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-api/pom.xml b/nifi-api/pom.xml
index e8a8222..8bdfc5c 100644
--- a/nifi-api/pom.xml
+++ b/nifi-api/pom.xml
@@ -21,5 +21,5 @@
<version>0.7.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-api</artifactId>
- <packaging>jar</packaging>
-</project>
+ <packaging>jar</packaging>
+ </project>
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java b/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java
new file mode 100644
index 0000000..539f24e
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/FileVariableRegistry.java
@@ -0,0 +1,61 @@
+/*
+ * 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.registry;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Map;
+
+
+abstract class FileVariableRegistry extends MultiMapVariableRegistry {
+
+ FileVariableRegistry() {
+ super();
+ }
+
+ FileVariableRegistry(File... files) throws IOException{
+ super();
+ addVariables(files);
+ }
+
+ FileVariableRegistry(Path... paths) throws IOException{
+ super();
+ addVariables(paths);
+ }
+
+ private void addVariables(File ...files) throws IOException{
+ if(files != null) {
+ for (final File file : files) {
+ registry.addMap(convertFile(file));
+ }
+ }
+ }
+
+
+ private void addVariables(Path ...paths) throws IOException{
+ if(paths != null) {
+ for (final Path path : paths) {
+ registry.addMap(convertFile(path.toFile()));
+ }
+ }
+ }
+
+ protected abstract Map<String,String> convertFile(File file) throws IOException;
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/MultiMap.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/MultiMap.java b/nifi-api/src/main/java/org/apache/nifi/registry/MultiMap.java
new file mode 100644
index 0000000..91e27c0
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/MultiMap.java
@@ -0,0 +1,145 @@
+/*
+ * 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.registry;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class MultiMap<V> implements Map<String,V> {
+
+ private final List<Map<String,V>> maps;
+
+ MultiMap() {
+ this.maps = new ArrayList<>();
+ }
+
+ @Override
+ public int size() {
+ return keySet().size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ for (final Map<String,V> map : maps) {
+ if (!map.isEmpty()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public boolean containsKey(final Object key) {
+ if (key == null) {
+ return false;
+ }
+
+ for (final Map<String,V> map : maps) {
+ if (map.containsKey(key)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public boolean containsValue(final Object value) {
+ for (final Map<String,V> map : maps) {
+ if (map.containsValue(value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ @SuppressWarnings("rawtypes")
+ public V get(final Object key) {
+ if (key == null) {
+ throw new IllegalArgumentException("Null Keys are not allowed");
+ }
+
+ for (final Map<String,V> map : maps) {
+ final V val = map.get(key);
+ if (val != null) {
+ return val;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public V put(String key, V value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public V remove(Object key) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void putAll(Map<? extends String, ? extends V> m) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void clear() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public Set<String> keySet() {
+ final Set<String> keySet = new HashSet<>();
+ for (final Map map : maps) {
+ keySet.addAll(map.keySet());
+ }
+ return keySet;
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public Collection<V> values() {
+ final Set<V> values = new HashSet<>();
+ for (final Map map : maps) {
+ values.addAll(map.values());
+ }
+ return values;
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public Set<java.util.Map.Entry<String, V>> entrySet() {
+ final Set<java.util.Map.Entry<String, V>> entrySet = new HashSet<>();
+ for (final Map map : maps) {
+ entrySet.addAll(map.entrySet());
+ }
+ return entrySet;
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ void addMap(Map<String,V> map){
+ this.maps.add(map);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/MultiMapVariableRegistry.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/MultiMapVariableRegistry.java b/nifi-api/src/main/java/org/apache/nifi/registry/MultiMapVariableRegistry.java
new file mode 100644
index 0000000..f9d04d0
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/MultiMapVariableRegistry.java
@@ -0,0 +1,70 @@
+/*
+ * 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.registry;
+
+import java.util.Map;
+import java.util.Set;
+
+/***
+ * This implementation of variable registry uses the MultiMap which stores one or more
+ * registries that can be searched, accessed and appended. NOTE: Duplicate values within
+ * or between added registries will be stored however on retrieval the first value encountered will be returned.
+ * */
+public class MultiMapVariableRegistry implements VariableRegistry {
+
+ protected final MultiMap<String> registry;
+
+ MultiMapVariableRegistry() {
+ this.registry = new MultiMap<>();
+ }
+
+ @SafeVarargs
+ MultiMapVariableRegistry(Map<String,String>...maps){
+ this();
+ if(maps != null) {
+ for (Map<String, String> map : maps) {
+ addVariables(map);
+ }
+ }
+ }
+
+ void addVariables(Map<String, String> map) {
+ this.registry.addMap(map);
+ }
+
+ @Override
+ public void addRegistry(VariableRegistry variableRegistry) {
+ if(!variableRegistry.getVariables().isEmpty()) {
+ this.registry.addMap(variableRegistry.getVariables());
+ }
+ }
+
+ @Override
+ public Map<String, String> getVariables() {
+ return registry;
+ }
+
+ @Override
+ public String getVariableValue(String variable) {
+ return registry.get(variable);
+ }
+
+ @Override
+ public Set<String> getVariableNames() {
+ return this.registry.keySet();
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java b/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java
new file mode 100644
index 0000000..bc5b15b
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/PropertiesVariableRegistry.java
@@ -0,0 +1,76 @@
+/*
+ * 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.registry;
+
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class PropertiesVariableRegistry extends FileVariableRegistry {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PropertiesVariableRegistry.class);
+
+ PropertiesVariableRegistry(File... files) throws IOException{
+ super(files);
+ }
+
+ PropertiesVariableRegistry(Path... paths) throws IOException {
+ super(paths);
+ }
+
+ PropertiesVariableRegistry(Properties...properties){
+ super();
+ addVariables(properties);
+ }
+
+ private void addVariables(Properties... properties){
+ if(properties != null) {
+ for (Properties props : properties) {
+ addVariables(convertToMap(props));
+ }
+ }
+ }
+
+ @Override
+ protected Map<String,String> convertFile(File file) throws IOException{
+
+ try(final InputStream inStream = new BufferedInputStream(new FileInputStream(file))) {
+ Properties properties = new Properties();
+ properties.load(inStream);
+ return convertToMap(properties);
+ }
+ }
+
+ private Map<String,String> convertToMap(Properties properties){
+ HashMap<String,String> propertiesMap = new HashMap<>(properties.keySet().size());
+ for(Object key: properties.keySet()){
+ propertiesMap.put((String)key,(String) properties.get(key));
+ }
+ return propertiesMap;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistry.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistry.java b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistry.java
new file mode 100644
index 0000000..9f54f23
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistry.java
@@ -0,0 +1,52 @@
+/*
+ * 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.registry;
+
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Access key/value pairs throughout the application.
+ */
+public interface VariableRegistry {
+
+ /**
+ * Returns a map of key/value pairs stored in the registry
+ * @return variables
+ **/
+ Map<String, String> getVariables();
+
+ /**
+ * Return a value for a given variable
+ * @param variable variable
+ * @return value
+ **/
+ String getVariableValue(String variable);
+
+ /**
+ * Concatenate a variable registry
+ * @param variableRegistry variableRegistry
+ * */
+ void addRegistry(VariableRegistry variableRegistry);
+
+ /**
+ * Returns a set variable names in the registry
+ * @return variableNames
+ **/
+ Set<String> getVariableNames();
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryFactory.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryFactory.java b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryFactory.java
new file mode 100644
index 0000000..1852ad4
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryFactory.java
@@ -0,0 +1,48 @@
+/*
+ * 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.registry;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.Properties;
+
+public class VariableRegistryFactory {
+
+ public static VariableRegistry getPropertiesInstance(final Properties...properties){
+ return new PropertiesVariableRegistry(properties);
+ }
+
+ public static VariableRegistry getPropertiesInstance(final Path... paths) throws IOException{
+ return new PropertiesVariableRegistry(paths);
+ }
+
+ public static VariableRegistry getPropertiesInstance(final File ...files) throws IOException{
+ return new PropertiesVariableRegistry(files);
+ }
+
+ @SafeVarargs
+ public static VariableRegistry getInstance(final Map<String,String> ...maps){
+ return new MultiMapVariableRegistry(maps);
+ }
+
+ public static VariableRegistry getInstance(){
+ return new MultiMapVariableRegistry();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryProvider.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryProvider.java b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryProvider.java
new file mode 100644
index 0000000..af7ab38
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryProvider.java
@@ -0,0 +1,23 @@
+/*
+ * 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.registry;
+
+public interface VariableRegistryProvider {
+
+ VariableRegistry getVariableRegistry();
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryUtils.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryUtils.java b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryUtils.java
new file mode 100644
index 0000000..c312dc7
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/registry/VariableRegistryUtils.java
@@ -0,0 +1,63 @@
+/*
+ * 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.registry;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.nifi.flowfile.FlowFile;
+
+public class VariableRegistryUtils {
+
+
+ public static VariableRegistry createVariableRegistry(){
+ VariableRegistry variableRegistry = VariableRegistryFactory.getInstance();
+ VariableRegistry envRegistry = VariableRegistryFactory.getInstance(System.getenv());
+ VariableRegistry propRegistry = VariableRegistryFactory.getPropertiesInstance(System.getProperties());
+ variableRegistry.addRegistry(envRegistry);
+ variableRegistry.addRegistry(propRegistry);
+ return variableRegistry;
+ }
+
+ public static VariableRegistry populateRegistry(VariableRegistry variableRegistry, final FlowFile flowFile, final Map<String, String> additionalAttributes){
+ final Map<String, String> flowFileAttributes = flowFile == null ? null : flowFile.getAttributes();
+ final Map<String, String> additionalMap = additionalAttributes == null ? null : additionalAttributes;
+
+ Map<String, String> flowFileProps = null;
+ if (flowFile != null) {
+ flowFileProps = new HashMap<>();
+ flowFileProps.put("flowFileId", String.valueOf(flowFile.getId()));
+ flowFileProps.put("fileSize", String.valueOf(flowFile.getSize()));
+ flowFileProps.put("entryDate", String.valueOf(flowFile.getEntryDate()));
+ flowFileProps.put("lineageStartDate", String.valueOf(flowFile.getLineageStartDate()));
+ }
+ VariableRegistry newRegistry = VariableRegistryFactory.getInstance();
+ newRegistry.addRegistry(variableRegistry);
+
+ if(flowFileAttributes != null) {
+ newRegistry.addRegistry(VariableRegistryFactory.getInstance(flowFileAttributes));
+ }
+ if(additionalMap != null) {
+ newRegistry.addRegistry(VariableRegistryFactory.getInstance(additionalMap));
+ }
+ if(flowFileProps != null) {
+ newRegistry.addRegistry(VariableRegistryFactory.getInstance(flowFileProps));
+ }
+ return newRegistry;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java b/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java
index 021d243..4420a56 100644
--- a/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java
+++ b/nifi-api/src/main/java/org/apache/nifi/web/NiFiWebConfigurationContext.java
@@ -20,12 +20,13 @@ import java.util.Collection;
import java.util.Map;
import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.registry.VariableRegistryProvider;
/**
* NiFi web context providing limited access to dataflow configuration for
* component custom UIs.
*/
-public interface NiFiWebConfigurationContext {
+public interface NiFiWebConfigurationContext extends VariableRegistryProvider {
/**
* @param serviceIdentifier of the controller service
@@ -117,4 +118,6 @@ public interface NiFiWebConfigurationContext {
* populated or isn't appropriate for the given request
*/
ComponentDetails getComponentDetails(NiFiWebRequestContext requestContext) throws ResourceNotFoundException, ClusterRequestException;
+
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistry.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistry.java b/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistry.java
new file mode 100644
index 0000000..0167e6d
--- /dev/null
+++ b/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistry.java
@@ -0,0 +1,137 @@
+/*
+ * 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.registry;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+public class TestVariableRegistry {
+
+ @Test
+ public void testReadMap(){
+ Map<String,String> variables1 = new HashMap<>();
+ variables1.put("fake.property.1","fake test value");
+
+ Map<String,String> variables2 = new HashMap<>();
+ variables1.put("fake.property.2","fake test value");
+
+ VariableRegistry registry = VariableRegistryFactory.getInstance(variables1,variables2);
+
+ Map<String,String> variables = registry.getVariables();
+ assertTrue(variables.size() == 2);
+ assertTrue(variables.get("fake.property.1").equals("fake test value"));
+ assertTrue(registry.getVariableValue("fake.property.2").equals("fake test value"));
+ }
+
+ @Test
+ public void testReadProperties(){
+ Properties properties = new Properties();
+ properties.setProperty("fake.property.1","fake test value");
+ VariableRegistry registry = VariableRegistryFactory.getPropertiesInstance(properties);
+ Map<String,String> variables = registry.getVariables();
+ assertTrue(variables.get("fake.property.1").equals("fake test value"));
+ }
+
+ @Test
+ public void testReadFiles() throws IOException{
+ final Path fooPath = Paths.get("src/test/resources/TestVariableRegistry/foobar.properties");
+ final Path testPath = Paths.get("src/test/resources/TestVariableRegistry/test.properties");
+ VariableRegistry registry = VariableRegistryFactory.getPropertiesInstance(fooPath.toFile(),testPath.toFile());
+ Map<String,String> variables = registry.getVariables();
+ assertTrue(variables.size() == 3);
+ assertTrue(variables.get("fake.property.1").equals("test me out 1"));
+ assertTrue(variables.get("fake.property.3").equals("test me out 3, test me out 4"));
+ }
+
+ @Test
+ public void testReadPaths() throws IOException{
+ final Path fooPath = Paths.get("src/test/resources/TestVariableRegistry/foobar.properties");
+ final Path testPath = Paths.get("src/test/resources/TestVariableRegistry/test.properties");
+ VariableRegistry registry = VariableRegistryFactory.getPropertiesInstance(fooPath,testPath);
+ Map<String,String> variables = registry.getVariables();
+ assertTrue(variables.size() == 3);
+ assertTrue(variables.get("fake.property.1").equals("test me out 1"));
+ assertTrue(variables.get("fake.property.3").equals("test me out 3, test me out 4"));
+ }
+
+ @Test
+ public void testAddRegistry() throws IOException{
+
+ final Path fooPath = Paths.get("src/test/resources/TestVariableRegistry/foobar.properties");
+ VariableRegistry pathRegistry = VariableRegistryFactory.getPropertiesInstance(fooPath);
+
+ final Path testPath = Paths.get("src/test/resources/TestVariableRegistry/test.properties");
+ VariableRegistry fileRegistry = VariableRegistryFactory.getPropertiesInstance(testPath.toFile());
+
+ Properties properties = new Properties();
+ properties.setProperty("fake.property.5","test me out 5");
+ VariableRegistry propRegistry = VariableRegistryFactory.getPropertiesInstance(properties);
+
+ propRegistry.addRegistry(pathRegistry);
+ propRegistry.addRegistry(fileRegistry);
+
+ Map<String,String> variables = propRegistry.getVariables();
+ assertTrue(variables.size() == 4);
+ }
+
+ @Test
+ public void testNoOverwriteRegistry()throws IOException{
+ final Path fooPath = Paths.get("src/test/resources/TestVariableRegistry/foobar.properties");
+ VariableRegistry pathRegistry = VariableRegistryFactory.getPropertiesInstance(fooPath);
+
+ final Path testPath = Paths.get("src/test/resources/TestVariableRegistry/test.properties");
+ VariableRegistry fileRegistry = VariableRegistryFactory.getPropertiesInstance(testPath.toFile());
+
+ Properties properties = new Properties();
+ properties.setProperty("fake.property.3","test me out 5");
+ VariableRegistry propRegistry = VariableRegistryFactory.getPropertiesInstance(properties);
+
+ propRegistry.addRegistry(pathRegistry);
+ propRegistry.addRegistry(fileRegistry);
+
+ Map<String,String> variables = propRegistry.getVariables();
+ String testDuplicate = propRegistry.getVariableValue("fake.property.3");
+ assertTrue(variables.size() == 3);
+ assertTrue(testDuplicate.equals("test me out 5"));
+ }
+
+ @Test
+ public void testGetVariableNames() throws IOException{
+ final Path fooPath = Paths.get("src/test/resources/TestVariableRegistry/foobar.properties");
+ final Path testPath = Paths.get("src/test/resources/TestVariableRegistry/test.properties");
+ VariableRegistry registry = VariableRegistryFactory.getPropertiesInstance(fooPath,testPath);
+ Set<String> variableNames= registry.getVariableNames();
+ assertTrue(variableNames.size() == 3);
+ assertTrue(variableNames.contains("fake.property.1"));
+ assertTrue(variableNames.contains("fake.property.2"));
+ assertTrue(variableNames.contains("fake.property.3"));
+ assertTrue(!variableNames.contains("fake.property.4"));
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistryUtils.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistryUtils.java b/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistryUtils.java
new file mode 100644
index 0000000..a1d5ea0
--- /dev/null
+++ b/nifi-api/src/test/java/org/apache/nifi/registry/TestVariableRegistryUtils.java
@@ -0,0 +1,123 @@
+/*
+ * 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.registry;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+public class TestVariableRegistryUtils {
+
+ @Test
+ public void testCreateVariableRegistry(){
+ System.setProperty("fake","test");
+ VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ Map<String,String> variables = variableRegistry.getVariables();
+ assertTrue(variables.containsKey("PATH"));
+ assertTrue(variables.get("fake").equals("test"));
+ }
+
+ @Test
+ public void testPopulateRegistry(){
+ System.setProperty("fake","test");
+ FlowFile flowFile = createFlowFile();
+
+ VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ VariableRegistry populatedRegistry = VariableRegistryUtils.populateRegistry(variableRegistry,flowFile,null);
+ Map<String,String> variables = populatedRegistry.getVariables();
+ assertTrue(variables.containsKey("PATH"));
+ assertTrue(variables.get("fake").equals("test"));
+ assertTrue(variables.get("flowFileId").equals("1"));
+ assertTrue(variables.get("fileSize").equals("50"));
+ assertTrue(variables.get("entryDate").equals("1000"));
+ assertTrue(variables.get("lineageStartDate").equals("10000"));
+ assertTrue(variables.get("filename").equals("fakefile.txt"));
+ }
+
+ @Test
+ public void testPopulateRegistryWithEmptyFlowFileAndAttributes(){
+ System.setProperty("fake","test");
+ VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ VariableRegistry populatedRegistry = VariableRegistryUtils.populateRegistry(variableRegistry,null,null);
+ Map<String,String> variables = populatedRegistry.getVariables();
+ assertTrue( variables.containsKey("PATH"));
+ assertTrue( variables.get("fake").equals("test"));
+ }
+
+
+ private FlowFile createFlowFile(){
+ return new FlowFile() {
+ @Override
+ public long getId() {
+ return 1;
+ }
+
+ @Override
+ public long getEntryDate() {
+ return 1000;
+ }
+
+ @Override
+ public long getLineageStartDate() {
+ return 10000;
+ }
+
+ @Override
+ public Long getLastQueueDate() {
+ return null;
+ }
+
+ @Override
+ public Set<String> getLineageIdentifiers() {
+ return null;
+ }
+
+ @Override
+ public boolean isPenalized() {
+ return false;
+ }
+
+ @Override
+ public String getAttribute(String key) {
+ return null;
+ }
+
+ @Override
+ public long getSize() {
+ return 50;
+ }
+
+ @Override
+ public Map<String, String> getAttributes() {
+ Map<String,String> attributes = new HashMap<>();
+ attributes.put("filename","fakefile.txt");
+ return attributes;
+ }
+
+ @Override
+ public int compareTo(FlowFile o) {
+ return 0;
+ }
+ };
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/test/resources/TestVariableRegistry/foobar.properties
----------------------------------------------------------------------
diff --git a/nifi-api/src/test/resources/TestVariableRegistry/foobar.properties b/nifi-api/src/test/resources/TestVariableRegistry/foobar.properties
new file mode 100644
index 0000000..1094e1b
--- /dev/null
+++ b/nifi-api/src/test/resources/TestVariableRegistry/foobar.properties
@@ -0,0 +1,16 @@
+# 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.
+
+fake.property.3=test me out 3, test me out 4
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-api/src/test/resources/TestVariableRegistry/test.properties
----------------------------------------------------------------------
diff --git a/nifi-api/src/test/resources/TestVariableRegistry/test.properties b/nifi-api/src/test/resources/TestVariableRegistry/test.properties
new file mode 100644
index 0000000..6191449
--- /dev/null
+++ b/nifi-api/src/test/resources/TestVariableRegistry/test.properties
@@ -0,0 +1,17 @@
+# 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.
+
+fake.property.1=test me out 1
+fake.property.2=test me out 2
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NotificationServiceManager.java
----------------------------------------------------------------------
diff --git a/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NotificationServiceManager.java b/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NotificationServiceManager.java
index 21d8e82..e80f2b1 100644
--- a/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NotificationServiceManager.java
+++ b/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/NotificationServiceManager.java
@@ -247,7 +247,7 @@ public class NotificationServiceManager {
configuredValue = fullPropDescriptor.getDefaultValue();
}
- return new StandardPropertyValue(configuredValue, null);
+ return new StandardPropertyValue(configuredValue, null, null);
}
@Override
@@ -364,7 +364,7 @@ public class NotificationServiceManager {
value = descriptor.getDefaultValue();
}
- return new StandardPropertyValue(value, null);
+ return new StandardPropertyValue(value, null, null);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/notification/NotificationValidationContext.java
----------------------------------------------------------------------
diff --git a/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/notification/NotificationValidationContext.java b/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/notification/NotificationValidationContext.java
index 49afc16..3ca1003 100644
--- a/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/notification/NotificationValidationContext.java
+++ b/nifi-bootstrap/src/main/java/org/apache/nifi/bootstrap/notification/NotificationValidationContext.java
@@ -48,12 +48,13 @@ public class NotificationValidationContext implements ValidationContext {
@Override
public PropertyValue newPropertyValue(final String rawValue) {
- return new StandardPropertyValue(rawValue, null);
+ return new StandardPropertyValue(rawValue, null, null);
}
@Override
public ExpressionLanguageCompiler newExpressionLanguageCompiler() {
- return new StandardExpressionLanguageCompiler();
+
+ return new StandardExpressionLanguageCompiler(null);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/EmptyPreparedQuery.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/EmptyPreparedQuery.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/EmptyPreparedQuery.java
index d85c9ef..5dec2fa 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/EmptyPreparedQuery.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/EmptyPreparedQuery.java
@@ -16,11 +16,10 @@
*/
package org.apache.nifi.attribute.expression.language;
-import java.util.Map;
import org.apache.nifi.expression.AttributeValueDecorator;
-import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
public class EmptyPreparedQuery implements PreparedQuery {
@@ -31,37 +30,7 @@ public class EmptyPreparedQuery implements PreparedQuery {
}
@Override
- public String evaluateExpressions(final FlowFile flowFile, final AttributeValueDecorator decorator) throws ProcessException {
- return value;
- }
-
- @Override
- public String evaluateExpressions() throws ProcessException {
- return value;
- }
-
- @Override
- public String evaluateExpressions(final AttributeValueDecorator decorator) throws ProcessException {
- return value;
- }
-
- @Override
- public String evaluateExpressions(final FlowFile flowFile) throws ProcessException {
- return value;
- }
-
- @Override
- public String evaluateExpressions(Map<String, String> attributes) throws ProcessException {
- return value;
- }
-
- @Override
- public String evaluateExpressions(Map<String, String> attributes, AttributeValueDecorator decorator) throws ProcessException {
- return value;
- }
-
- @Override
- public String evaluateExpressions(FlowFile flowFile, Map<String, String> additionalAttributes, AttributeValueDecorator decorator) throws ProcessException {
+ public String evaluateExpressions(VariableRegistry variableRegistry, AttributeValueDecorator decorator) throws ProcessException {
return value;
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/InvalidPreparedQuery.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/InvalidPreparedQuery.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/InvalidPreparedQuery.java
index aa2428d..0ca9a8f 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/InvalidPreparedQuery.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/InvalidPreparedQuery.java
@@ -16,12 +16,11 @@
*/
package org.apache.nifi.attribute.expression.language;
-import java.util.Map;
import org.apache.nifi.attribute.expression.language.exception.AttributeExpressionLanguageException;
import org.apache.nifi.expression.AttributeValueDecorator;
-import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
/**
* An implementation of PreparedQuery that throws an
@@ -40,37 +39,9 @@ public class InvalidPreparedQuery implements PreparedQuery {
}
@Override
- public String evaluateExpressions(final FlowFile flowFile, final AttributeValueDecorator decorator) throws ProcessException {
+ public String evaluateExpressions(final VariableRegistry variableRegistry, final AttributeValueDecorator decorator) throws ProcessException {
throw new AttributeExpressionLanguageException("Invalid Expression: " + query + " due to " + explanation);
}
- @Override
- public String evaluateExpressions() throws ProcessException {
- throw new AttributeExpressionLanguageException("Invalid Expression: " + query + " due to " + explanation);
- }
-
- @Override
- public String evaluateExpressions(final AttributeValueDecorator decorator) throws ProcessException {
- throw new AttributeExpressionLanguageException("Invalid Expression: " + query + " due to " + explanation);
- }
-
- @Override
- public String evaluateExpressions(final FlowFile flowFile) throws ProcessException {
- throw new AttributeExpressionLanguageException("Invalid Expression: " + query + " due to " + explanation);
- }
- @Override
- public String evaluateExpressions(final Map<String, String> attributes) throws ProcessException {
- throw new AttributeExpressionLanguageException("Invalid Expression: " + query + " due to " + explanation);
- }
-
- @Override
- public String evaluateExpressions(final Map<String, String> attributes, final AttributeValueDecorator decorator) throws ProcessException {
- throw new AttributeExpressionLanguageException("Invalid Expression: " + query + " due to " + explanation);
- }
-
- @Override
- public String evaluateExpressions(FlowFile flowFile, Map<String, String> additionalAttributes, AttributeValueDecorator decorator) throws ProcessException {
- throw new AttributeExpressionLanguageException("Invalid Expression: " + query + " due to " + explanation);
- }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/PreparedQuery.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/PreparedQuery.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/PreparedQuery.java
index ad9225d..37d8b86 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/PreparedQuery.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/PreparedQuery.java
@@ -16,25 +16,13 @@
*/
package org.apache.nifi.attribute.expression.language;
-import java.util.Map;
import org.apache.nifi.expression.AttributeValueDecorator;
-import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
public interface PreparedQuery {
- String evaluateExpressions(FlowFile flowFile, AttributeValueDecorator decorator) throws ProcessException;
+ String evaluateExpressions(VariableRegistry registry, AttributeValueDecorator decorator) throws ProcessException;
- String evaluateExpressions() throws ProcessException;
-
- String evaluateExpressions(AttributeValueDecorator decorator) throws ProcessException;
-
- String evaluateExpressions(FlowFile flowFile) throws ProcessException;
-
- String evaluateExpressions(Map<String, String> attributes) throws ProcessException;
-
- String evaluateExpressions(Map<String, String> attributes, AttributeValueDecorator decorator) throws ProcessException;
-
- String evaluateExpressions(FlowFile flowFile, Map<String, String> additionalAttributes, AttributeValueDecorator decorator) throws ProcessException;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/Query.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/Query.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/Query.java
index d39f0f8..f2d3915 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/Query.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/Query.java
@@ -18,13 +18,9 @@ package org.apache.nifi.attribute.expression.language;
import java.net.UnknownHostException;
import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionLexer;
@@ -198,6 +194,7 @@ import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpre
import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.UUID;
import org.apache.nifi.attribute.expression.language.evaluation.selection.MappingEvaluator;
+import org.apache.nifi.registry.VariableRegistry;
/**
* Class used for creating and evaluating NiFi Expression Language. Once a Query
@@ -367,8 +364,8 @@ public class Query {
return -1;
}
- static String evaluateExpression(final Tree tree, final String queryText, final Map<String, String> expressionMap, final AttributeValueDecorator decorator) throws ProcessException {
- final Object evaluated = Query.fromTree(tree, queryText).evaluate(expressionMap).getValue();
+ static String evaluateExpression(final Tree tree, final String queryText, final VariableRegistry registry, final AttributeValueDecorator decorator) throws ProcessException {
+ final Object evaluated = Query.fromTree(tree, queryText).evaluate(registry).getValue();
if (evaluated == null) {
return null;
}
@@ -378,29 +375,12 @@ public class Query {
return decorator == null ? escaped : decorator.decorate(escaped);
}
- static String evaluateExpressions(final String rawValue, Map<String, String> expressionMap) throws ProcessException {
- return evaluateExpressions(rawValue, expressionMap, null);
+ static String evaluateExpressions(final String rawValue, VariableRegistry registry) throws ProcessException {
+ return evaluateExpressions(rawValue, registry, null);
}
- static String evaluateExpressions(final String rawValue) throws ProcessException {
- return evaluateExpressions(rawValue, createExpressionMap(null), null);
- }
-
- static String evaluateExpressions(final String rawValue, final FlowFile flowFile) throws ProcessException {
- return evaluateExpressions(rawValue, createExpressionMap(flowFile), null);
- }
-
- static String evaluateExpressions(final String rawValue, Map<String, String> expressionMap, final AttributeValueDecorator decorator) throws ProcessException {
- return Query.prepare(rawValue).evaluateExpressions(expressionMap, decorator);
- }
-
- public static String evaluateExpressions(final String rawValue, final FlowFile flowFile, final AttributeValueDecorator decorator) throws ProcessException {
- if (rawValue == null) {
- return null;
- }
-
- final Map<String, String> expressionMap = createExpressionMap(flowFile);
- return evaluateExpressions(rawValue, expressionMap, decorator);
+ static String evaluateExpressions(final String rawValue, VariableRegistry registry, final AttributeValueDecorator decorator) throws ProcessException {
+ return Query.prepare(rawValue).evaluateExpressions(registry, decorator);
}
private static Evaluator<?> getRootSubjectEvaluator(final Evaluator<?> evaluator) {
@@ -426,150 +406,6 @@ public class Query {
return value.replaceAll("\\$\\$(?=\\$*\\{.*?\\})", "\\$");
}
- static Map<String, String> createExpressionMap(final FlowFile flowFile) {
- return createExpressionMap(flowFile, null);
- }
-
- static Map<String, String> createExpressionMap(final FlowFile flowFile, final Map<String, String> additionalAttributes) {
- final Map<String, String> attributeMap = flowFile == null ? Collections.<String, String> emptyMap() : flowFile.getAttributes();
- final Map<String, String> additionalOrEmpty = additionalAttributes == null ? Collections.<String, String> emptyMap() : additionalAttributes;
- final Map<String, String> envMap = System.getenv();
- final Map<?, ?> sysProps = System.getProperties();
-
- final Map<String, String> flowFileProps = new HashMap<>();
- if (flowFile != null) {
- flowFileProps.put("flowFileId", String.valueOf(flowFile.getId()));
- flowFileProps.put("fileSize", String.valueOf(flowFile.getSize()));
- flowFileProps.put("entryDate", String.valueOf(flowFile.getEntryDate()));
- flowFileProps.put("lineageStartDate", String.valueOf(flowFile.getLineageStartDate()));
- }
-
- return wrap(additionalOrEmpty, attributeMap, flowFileProps, envMap, sysProps);
- }
-
- private static Map<String, String> wrap(final Map<String, String> additional, final Map<String, String> attributes, final Map<String, String> flowFileProps,
- final Map<String, String> env, final Map<?, ?> sysProps) {
- @SuppressWarnings("rawtypes")
- final Map[] maps = new Map[] {additional, attributes, flowFileProps, env, sysProps};
-
- return new Map<String, String>() {
- @Override
- public int size() {
- int size = 0;
- for (final Map<?, ?> map : maps) {
- size += map.size();
- }
- return size;
- }
-
- @Override
- public boolean isEmpty() {
- for (final Map<?, ?> map : maps) {
- if (!map.isEmpty()) {
- return false;
- }
- }
- return true;
- }
-
- @Override
- public boolean containsKey(final Object key) {
- if (key == null) {
- return false;
- }
- if (!(key instanceof String)) {
- return false;
- }
-
- for (final Map<?, ?> map : maps) {
- if (map.containsKey(key)) {
- return true;
- }
- }
- return false;
- }
-
- @Override
- public boolean containsValue(final Object value) {
- for (final Map<?, ?> map : maps) {
- if (map.containsValue(value)) {
- return true;
- }
- }
- return false;
- }
-
- @Override
- @SuppressWarnings("rawtypes")
- public String get(final Object key) {
- if (key == null) {
- throw new IllegalArgumentException("Null Keys are not allowed");
- }
- if (!(key instanceof String)) {
- return null;
- }
-
- for (final Map map : maps) {
- final Object val = map.get(key);
- if (val != null) {
- return String.valueOf(val);
- }
- }
- return null;
- }
-
- @Override
- public String put(String key, String value) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public String remove(final Object key) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void putAll(final Map<? extends String, ? extends String> m) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void clear() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- @SuppressWarnings({"unchecked", "rawtypes"})
- public Set<String> keySet() {
- final Set<String> keySet = new HashSet<>();
- for (final Map map : maps) {
- keySet.addAll(map.keySet());
- }
- return keySet;
- }
-
- @Override
- @SuppressWarnings({"unchecked", "rawtypes"})
- public Collection<String> values() {
- final Set<String> values = new HashSet<>();
- for (final Map map : maps) {
- values.addAll(map.values());
- }
- return values;
- }
-
- @Override
- @SuppressWarnings({"unchecked", "rawtypes"})
- public Set<java.util.Map.Entry<String, String>> entrySet() {
- final Set<java.util.Map.Entry<String, String>> entrySet = new HashSet<>();
- for (final Map map : maps) {
- entrySet.addAll(map.entrySet());
- }
- return entrySet;
- }
-
- };
- }
public static Query fromTree(final Tree tree, final String text) {
return new Query(text, tree, buildEvaluator(tree));
@@ -706,20 +542,12 @@ public class Query {
return evaluator.getResultType();
}
- QueryResult<?> evaluate() {
- return evaluate(createExpressionMap(null));
- }
-
- QueryResult<?> evaluate(final FlowFile flowFile) {
- return evaluate(createExpressionMap(flowFile));
- }
-
- QueryResult<?> evaluate(final Map<String, String> attributes) {
+ QueryResult<?> evaluate(final VariableRegistry registry) {
if (evaluated.getAndSet(true)) {
throw new IllegalStateException("A Query cannot be evaluated more than once");
}
- return evaluator.evaluate(attributes);
+ return evaluator.evaluate(registry.getVariables());
}
Tree getTree() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardAttributeExpression.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardAttributeExpression.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardAttributeExpression.java
index 49ef6ef..cbfb511 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardAttributeExpression.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardAttributeExpression.java
@@ -20,13 +20,17 @@ import org.apache.nifi.expression.AttributeExpression;
import org.apache.nifi.expression.AttributeValueDecorator;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
public class StandardAttributeExpression implements AttributeExpression {
private final Query query;
+ private final VariableRegistry variableRegistry;
- public StandardAttributeExpression(final Query query) {
+ public StandardAttributeExpression(final Query query, final VariableRegistry variableRegistry) {
this.query = query;
+ this.variableRegistry = variableRegistry;
}
@Override
@@ -51,7 +55,8 @@ public class StandardAttributeExpression implements AttributeExpression {
@Override
public String evaluate(final FlowFile flowFile, final AttributeValueDecorator decorator) throws ProcessException {
- final Object evaluationResult = query.evaluate(flowFile).getValue();
+ VariableRegistry flowFileRegistry = VariableRegistryUtils.populateRegistry(variableRegistry,flowFile,null);
+ final Object evaluationResult = query.evaluate(flowFileRegistry).getValue();
if (evaluationResult == null) {
return "";
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardExpressionLanguageCompiler.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardExpressionLanguageCompiler.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardExpressionLanguageCompiler.java
index cec73d1..e85853f 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardExpressionLanguageCompiler.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardExpressionLanguageCompiler.java
@@ -20,13 +20,20 @@ import org.apache.nifi.attribute.expression.language.exception.AttributeExpressi
import org.apache.nifi.expression.AttributeExpression;
import org.apache.nifi.expression.ExpressionLanguageCompiler;
import org.apache.nifi.expression.AttributeExpression.ResultType;
+import org.apache.nifi.registry.VariableRegistry;
public class StandardExpressionLanguageCompiler implements ExpressionLanguageCompiler {
+ private final VariableRegistry variableRegistry;
+
+ public StandardExpressionLanguageCompiler(final VariableRegistry variableRegistry) {
+ this.variableRegistry = variableRegistry;
+ }
+
@Override
public AttributeExpression compile(final String expression) throws IllegalArgumentException {
try {
- return new StandardAttributeExpression(Query.compile(expression));
+ return new StandardAttributeExpression(Query.compile(expression),variableRegistry);
} catch (final AttributeExpressionLanguageParsingException e) {
throw new IllegalArgumentException(e.getMessage());
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPreparedQuery.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPreparedQuery.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPreparedQuery.java
index b81a583..4ee3e02 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPreparedQuery.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPreparedQuery.java
@@ -22,10 +22,10 @@ import java.util.List;
import java.util.Map;
import org.apache.nifi.expression.AttributeValueDecorator;
-import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.exception.ProcessException;
import org.antlr.runtime.tree.Tree;
+import org.apache.nifi.registry.VariableRegistry;
public class StandardPreparedQuery implements PreparedQuery {
@@ -37,20 +37,16 @@ public class StandardPreparedQuery implements PreparedQuery {
this.trees = new HashMap<>(trees);
}
- @Override
- public String evaluateExpressions(Map<String, String> attributes) throws ProcessException {
- return evaluateExpressions(attributes, null);
- }
@Override
- public String evaluateExpressions(final Map<String, String> attributes, final AttributeValueDecorator decorator) throws ProcessException {
+ public String evaluateExpressions(final VariableRegistry registry, final AttributeValueDecorator decorator) throws ProcessException {
final StringBuilder sb = new StringBuilder();
for (final String val : queryStrings) {
final Tree tree = trees.get(val);
if (tree == null) {
sb.append(val);
} else {
- final String evaluated = Query.evaluateExpression(tree, val, attributes, decorator);
+ final String evaluated = Query.evaluateExpression(tree, val, registry, decorator);
if (evaluated != null) {
sb.append(evaluated);
}
@@ -59,31 +55,4 @@ public class StandardPreparedQuery implements PreparedQuery {
return sb.toString();
}
- @Override
- public String evaluateExpressions(final FlowFile flowFile, final Map<String, String> additionalAttributes, final AttributeValueDecorator decorator) throws ProcessException {
- final Map<String, String> expressionMap = Query.createExpressionMap(flowFile, additionalAttributes);
- return evaluateExpressions(expressionMap, decorator);
- }
-
- @Override
- public String evaluateExpressions(final FlowFile flowFile, final AttributeValueDecorator decorator) throws ProcessException {
- final Map<String, String> expressionMap = Query.createExpressionMap(flowFile);
- return evaluateExpressions(expressionMap, decorator);
- }
-
- @Override
- public String evaluateExpressions() throws ProcessException {
- return evaluateExpressions((FlowFile) null, null);
- }
-
- @Override
- public String evaluateExpressions(final AttributeValueDecorator decorator) throws ProcessException {
- return evaluateExpressions((FlowFile) null, decorator);
- }
-
- @Override
- public String evaluateExpressions(final FlowFile flowFile) throws ProcessException {
- return evaluateExpressions(flowFile, null);
- }
-
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java
index a5c336a..c3cff49 100644
--- a/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java
+++ b/nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/StandardPropertyValue.java
@@ -26,6 +26,8 @@ import org.apache.nifi.expression.AttributeValueDecorator;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.util.FormatUtils;
public class StandardPropertyValue implements PropertyValue {
@@ -33,24 +35,27 @@ public class StandardPropertyValue implements PropertyValue {
private final String rawValue;
private final ControllerServiceLookup serviceLookup;
private final PreparedQuery preparedQuery;
+ private final VariableRegistry variableRegistry;
- public StandardPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup) {
- this(rawValue, serviceLookup, Query.prepare(rawValue));
+ public StandardPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final VariableRegistry variableRegistry) {
+ this(rawValue, serviceLookup, Query.prepare(rawValue), variableRegistry);
}
/**
* Constructs a new StandardPropertyValue with the given value & service lookup and indicates whether or not the rawValue contains any NiFi Expressions. If it is unknown whether or not the value
- * contains any NiFi Expressions, the {@link #StandardPropertyValue(String, ControllerServiceLookup)} constructor should be used or <code>true</code> should be passed. However, if it is known that
- * the value contains no NiFi Expression, that information should be provided so that calls to {@link #evaluateAttributeExpressions()} are much more efficient
- *
+ * contains any NiFi Expressions, the {@link #StandardPropertyValue(String, ControllerServiceLookup, VariableRegistry)} constructor should be used or <code>true</code> should be passed.
+ * However, if it is known that the value contains no NiFi Expression, that information should be provided so that calls to {@link #evaluateAttributeExpressions()} are much more efficient
* @param rawValue value
* @param serviceLookup lookup
* @param preparedQuery query
+ * @param variableRegistry variableRegistry
*/
- public StandardPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PreparedQuery preparedQuery) {
+ public StandardPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PreparedQuery preparedQuery,
+ final VariableRegistry variableRegistry) {
this.rawValue = rawValue;
this.serviceLookup = serviceLookup;
this.preparedQuery = preparedQuery;
+ this.variableRegistry = variableRegistry;
}
@Override
@@ -134,9 +139,14 @@ public class StandardPropertyValue implements PropertyValue {
return this;
}
- return new StandardPropertyValue(preparedQuery.evaluateExpressions(flowFile, additionalAttributes, decorator), serviceLookup, null);
+ VariableRegistry sysRegistry = variableRegistry != null ? variableRegistry : VariableRegistryUtils.createVariableRegistry();
+ VariableRegistry registry = VariableRegistryUtils.populateRegistry(sysRegistry,flowFile,additionalAttributes);
+
+ return new StandardPropertyValue(preparedQuery.evaluateExpressions(registry,decorator), serviceLookup,null,registry);
}
+
+
@Override
public String toString() {
return rawValue;
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/test/groovy/org/apache/nifi/attribute/expression/language/QueryGroovyTest.groovy
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/test/groovy/org/apache/nifi/attribute/expression/language/QueryGroovyTest.groovy b/nifi-commons/nifi-expression-language/src/test/groovy/org/apache/nifi/attribute/expression/language/QueryGroovyTest.groovy
index bdd704d..3a5f587 100644
--- a/nifi-commons/nifi-expression-language/src/test/groovy/org/apache/nifi/attribute/expression/language/QueryGroovyTest.groovy
+++ b/nifi-commons/nifi-expression-language/src/test/groovy/org/apache/nifi/attribute/expression/language/QueryGroovyTest.groovy
@@ -18,6 +18,8 @@ package org.apache.nifi.attribute.expression.language
import org.apache.nifi.attribute.expression.language.evaluation.QueryResult
import org.apache.nifi.expression.AttributeExpression
+import org.apache.nifi.registry.VariableRegistry
+import org.apache.nifi.registry.VariableRegistryFactory
import org.junit.After
import org.junit.Before
import org.junit.BeforeClass
@@ -57,6 +59,7 @@ public class QueryGroovyTest extends GroovyTestCase {
final Map<String, String> attributes = [
single : ORIGINAL_VALUE,
repeating: [ORIGINAL_VALUE].multiply(n).join(" ")]
+ final VariableRegistry variableRegistry = VariableRegistryFactory.getInstance(attributes)
logger.info("Attributes: ${attributes}")
final String REPLACEMENT_VALUE = "Goodbye Planet"
@@ -75,10 +78,10 @@ public class QueryGroovyTest extends GroovyTestCase {
Query replaceRepeatingQuery = Query.compile(REPLACE_REPEATING_EXPRESSION)
// Act
- QueryResult<?> replaceSingleResult = replaceSingleQuery.evaluate(attributes)
+ QueryResult<?> replaceSingleResult = replaceSingleQuery.evaluate(variableRegistry)
logger.info("Replace single result: ${replaceSingleResult.value}")
- QueryResult<?> replaceRepeatingResult = replaceRepeatingQuery.evaluate(attributes)
+ QueryResult<?> replaceRepeatingResult = replaceRepeatingQuery.evaluate(variableRegistry)
logger.info("Replace repeating result: ${replaceRepeatingResult.value}")
// Assert
@@ -97,6 +100,7 @@ public class QueryGroovyTest extends GroovyTestCase {
final Map<String, String> attributes = [
single : ORIGINAL_VALUE,
repeating: [ORIGINAL_VALUE].multiply(n).join(" ")]
+ final VariableRegistry variableRegistry = VariableRegistryFactory.getInstance(attributes)
logger.info("Attributes: ${attributes}")
final String REPLACEMENT_VALUE = "Goodbye Planet"
@@ -115,10 +119,10 @@ public class QueryGroovyTest extends GroovyTestCase {
Query replaceRepeatingQuery = Query.compile(REPLACE_REPEATING_EXPRESSION)
// Act
- QueryResult<?> replaceSingleResult = replaceSingleQuery.evaluate(attributes)
+ QueryResult<?> replaceSingleResult = replaceSingleQuery.evaluate(variableRegistry)
logger.info("Replace single result: ${replaceSingleResult.value}")
- QueryResult<?> replaceRepeatingResult = replaceRepeatingQuery.evaluate(attributes)
+ QueryResult<?> replaceRepeatingResult = replaceRepeatingQuery.evaluate(variableRegistry)
logger.info("Replace repeating result: ${replaceRepeatingResult.value}")
// Assert
@@ -137,6 +141,7 @@ public class QueryGroovyTest extends GroovyTestCase {
final Map<String, String> attributes = [
single : ORIGINAL_VALUE,
repeating: [ORIGINAL_VALUE].multiply(n).join(" ")]
+ final VariableRegistry variableRegistry = VariableRegistryFactory.getInstance(attributes)
logger.info("Attributes: ${attributes}")
final String REPLACEMENT_VALUE = "Goodbye Planet"
@@ -155,10 +160,10 @@ public class QueryGroovyTest extends GroovyTestCase {
Query replaceRepeatingQuery = Query.compile(REPLACE_REPEATING_EXPRESSION)
// Act
- QueryResult<?> replaceSingleResult = replaceSingleQuery.evaluate(attributes)
+ QueryResult<?> replaceSingleResult = replaceSingleQuery.evaluate(variableRegistry)
logger.info("Replace single result: ${replaceSingleResult.value}")
- QueryResult<?> replaceRepeatingResult = replaceRepeatingQuery.evaluate(attributes)
+ QueryResult<?> replaceRepeatingResult = replaceRepeatingQuery.evaluate(variableRegistry)
logger.info("Replace repeating result: ${replaceRepeatingResult.value}")
// Assert
[3/5] nifi git commit: NIFI-1974 - initial commit variable registry
for custom property support
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java
index 05bd017..9f4bf4d 100644
--- a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java
+++ b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java
@@ -35,6 +35,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Properties;
import org.apache.nifi.attribute.expression.language.Query.Range;
import org.apache.nifi.attribute.expression.language.evaluation.QueryResult;
@@ -43,7 +44,12 @@ import org.apache.nifi.attribute.expression.language.exception.AttributeExpressi
import org.apache.nifi.expression.AttributeExpression.ResultType;
import org.apache.nifi.flowfile.FlowFile;
import org.antlr.runtime.tree.Tree;
+
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryFactory;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.junit.Assert;
+
import org.junit.Ignore;
import org.junit.Test;
@@ -51,6 +57,7 @@ import org.mockito.Mockito;
public class TestQuery {
+
@Test
public void testCompilation() {
assertInvalid("${attr:uuid()}");
@@ -114,7 +121,7 @@ public class TestQuery {
final Map<String, String> attributes = new HashMap<>();
attributes.put("x", "x");
attributes.put("y", "x");
- final String result = Query.evaluateExpressions(expression, attributes, null);
+ final String result = Query.evaluateExpressions(expression,VariableRegistryFactory.getInstance(attributes), null);
assertEquals("true", result);
Query.validateExpression(expression, false);
@@ -174,14 +181,14 @@ public class TestQuery {
public void testWithTicksOutside() {
final Map<String, String> attributes = new HashMap<>();
attributes.put("attr", "My Value");
-
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
assertEquals(1, Query.extractExpressionRanges("\"${attr}").size());
assertEquals(1, Query.extractExpressionRanges("'${attr}").size());
assertEquals(1, Query.extractExpressionRanges("'${attr}'").size());
assertEquals(1, Query.extractExpressionRanges("${attr}").size());
- assertEquals("'My Value'", Query.evaluateExpressions("'${attr}'", attributes, null));
- assertEquals("'My Value", Query.evaluateExpressions("'${attr}", attributes, null));
+ assertEquals("'My Value'", Query.evaluateExpressions("'${attr}'", registry, null));
+ assertEquals("'My Value", Query.evaluateExpressions("'${attr}", registry, null));
}
@Test
@@ -191,7 +198,7 @@ public class TestQuery {
final Map<String, String> attributes = new HashMap<>();
attributes.put("dateTime", "2013/11/18 10:22:27.678");
- final QueryResult<?> result = query.evaluate(attributes);
+ final QueryResult<?> result = query.evaluate(VariableRegistryFactory.getInstance(attributes));
assertEquals(ResultType.NUMBER, result.getResultType());
assertEquals(1384788147678L, result.getValue());
}
@@ -220,7 +227,7 @@ public class TestQuery {
final Date roundedToNearestSecond = new Date(date.getTime() - millis);
final String formatted = sdf.format(roundedToNearestSecond);
- final QueryResult<?> result = query.evaluate(attributes);
+ final QueryResult<?> result = query.evaluate(VariableRegistryFactory.getInstance(attributes));
assertEquals(ResultType.STRING, result.getResultType());
assertEquals(formatted, result.getValue());
}
@@ -230,14 +237,31 @@ public class TestQuery {
final Map<String, String> attributes = new HashMap<>();
attributes.put("x", "abc");
attributes.put("a", "abc");
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
verifyEquals("${x:equals(${a})}", attributes, true);
Query.validateExpression("${x:equals('${a}')}", false);
- assertEquals("true", Query.evaluateExpressions("${x:equals('${a}')}", attributes, null));
+ assertEquals("true", Query.evaluateExpressions("${x:equals('${a}')}", registry, null));
+
+ Query.validateExpression("${x:equals(\"${a}\")}", false);
+ assertEquals("true", Query.evaluateExpressions("${x:equals(\"${a}\")}", registry, null));
+ }
+
+ @Test
+ public void testEmbeddedExpressionsAndQuotesWithProperties() {
+ final Properties attributes = new Properties();
+ attributes.put("x", "abc");
+ attributes.put("a", "abc");
+ VariableRegistry registry = VariableRegistryFactory.getPropertiesInstance(attributes);
+
+ verifyEquals("${x:equals(${a})}",registry,true);
+
+ Query.validateExpression("${x:equals('${a}')}", false);
+ assertEquals("true", Query.evaluateExpressions("${x:equals('${a}')}", registry, null));
Query.validateExpression("${x:equals(\"${a}\")}", false);
- assertEquals("true", Query.evaluateExpressions("${x:equals(\"${a}\")}", attributes, null));
+ assertEquals("true", Query.evaluateExpressions("${x:equals(\"${a}\")}", registry, null));
}
@Test
@@ -343,7 +367,9 @@ public class TestQuery {
Mockito.when(mockFlowFile.getSize()).thenReturn(1L);
Mockito.when(mockFlowFile.getLineageIdentifiers()).thenReturn(new HashSet<String>());
Mockito.when(mockFlowFile.getLineageStartDate()).thenReturn(System.currentTimeMillis());
- return Query.evaluateExpressions(queryString, mockFlowFile);
+
+ final VariableRegistry variableRegistry = VariableRegistryUtils.populateRegistry(VariableRegistryUtils.createVariableRegistry(),mockFlowFile,null);
+ return Query.evaluateExpressions(queryString,variableRegistry);
}
@Test
@@ -499,7 +525,7 @@ public class TestQuery {
verifyEquals("${x:toNumber():gt( ${y:toNumber():plus( ${z:toNumber()} )} )}", attributes, true);
attributes.put("y", "88");
- assertEquals("true", Query.evaluateExpressions("${x:equals( '${y}' )}", attributes, null));
+ assertEquals("true", Query.evaluateExpressions("${x:equals( '${y}' )}", VariableRegistryFactory.getInstance(attributes), null));
}
@Test
@@ -547,7 +573,7 @@ public class TestQuery {
final String format = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
final String query = "startDateTime=\"${date:toNumber():toDate():format(\"" + format + "\")}\"";
- final String result = Query.evaluateExpressions(query, attributes, null);
+ final String result = Query.evaluateExpressions(query, VariableRegistryFactory.getInstance(attributes), null);
final String expectedTime = new SimpleDateFormat(format, Locale.US).format(timestamp);
assertEquals("startDateTime=\"" + expectedTime + "\"", result);
@@ -616,7 +642,7 @@ public class TestQuery {
final String query = "${ abc:equals('abc'):or( \n\t${xx:isNull()}\n) }";
assertEquals(ResultType.BOOLEAN, Query.getResultType(query));
Query.validateExpression(query, false);
- assertEquals("true", Query.evaluateExpressions(query));
+ assertEquals("true", Query.evaluateExpressions(query,VariableRegistryUtils.createVariableRegistry()));
}
@Test
@@ -632,7 +658,7 @@ public class TestQuery {
public void testComments() {
final Map<String, String> attributes = new HashMap<>();
attributes.put("abc", "xyz");
-
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
final String expression
= "# hello, world\n"
+ "${# ref attr\n"
@@ -643,12 +669,12 @@ public class TestQuery {
+ "}";
Query query = Query.compile(expression);
- QueryResult<?> result = query.evaluate(attributes);
+ QueryResult<?> result = query.evaluate(registry);
assertEquals(ResultType.STRING, result.getResultType());
assertEquals("xyz", result.getValue());
query = Query.compile("${abc:append('# hello') #good-bye \n}");
- result = query.evaluate(attributes);
+ result = query.evaluate(registry);
assertEquals(ResultType.STRING, result.getResultType());
assertEquals("xyz# hello", result.getValue());
}
@@ -777,14 +803,15 @@ public class TestQuery {
final Map<String, String> attributes = new HashMap<>();
attributes.put("entryDate", String.valueOf(now.getTimeInMillis()));
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
verifyEquals("${entryDate:toNumber():toDate():format('yyyy')}", attributes, String.valueOf(year));
attributes.clear();
attributes.put("month", "3");
attributes.put("day", "4");
attributes.put("year", "2013");
- assertEquals("63", Query.evaluateExpressions("${year:append('/'):append(${month}):append('/'):append(${day}):toDate('yyyy/MM/dd'):format('D')}", attributes, null));
- assertEquals("63", Query.evaluateExpressions("${year:append('/'):append('${month}'):append('/'):append('${day}'):toDate('yyyy/MM/dd'):format('D')}", attributes, null));
+ assertEquals("63", Query.evaluateExpressions("${year:append('/'):append(${month}):append('/'):append(${day}):toDate('yyyy/MM/dd'):format('D')}", registry, null));
+ assertEquals("63", Query.evaluateExpressions("${year:append('/'):append('${month}'):append('/'):append('${day}'):toDate('yyyy/MM/dd'):format('D')}", registry, null));
verifyEquals("${year:append('/'):append(${month}):append('/'):append(${day}):toDate('yyyy/MM/dd'):format('D')}", attributes, "63");
}
@@ -792,8 +819,9 @@ public class TestQuery {
@Test
public void testSystemProperty() {
System.setProperty("hello", "good-bye");
- assertEquals("good-bye", Query.evaluateExpressions("${hello}"));
- assertEquals("good-bye", Query.compile("${hello}").evaluate().getValue());
+ VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ assertEquals("good-bye", Query.evaluateExpressions("${hello}",VariableRegistryUtils.createVariableRegistry()));
+ assertEquals("good-bye", Query.compile("${hello}").evaluate(variableRegistry).getValue());
}
@Test
@@ -833,14 +861,15 @@ public class TestQuery {
final Map<String, String> attributes = new HashMap<>();
attributes.put("abc", "a,b,c");
attributes.put("xyz", "abc");
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
final String query = "${anyDelineatedValue('${abc}', ','):equals('b')}";
assertEquals(ResultType.BOOLEAN, Query.getResultType(query));
- assertEquals("true", Query.evaluateExpressions(query, attributes, null));
- assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('a')}", attributes, null));
- assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('c')}", attributes, null));
- assertEquals("false", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('d')}", attributes, null));
+ assertEquals("true", Query.evaluateExpressions(query, registry, null));
+ assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('a')}", registry, null));
+ assertEquals("true", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('c')}", registry, null));
+ assertEquals("false", Query.evaluateExpressions("${anyDelineatedValue('${abc}', ','):equals('d')}", registry, null));
verifyEquals("${anyDelineatedValue(${abc}, ','):equals('b')}", attributes, true);
verifyEquals("${anyDelineatedValue(${abc}, ','):equals('a')}", attributes, true);
@@ -854,13 +883,15 @@ public class TestQuery {
attributes.put("abc", "a,b,c");
attributes.put("xyz", "abc");
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
+
final String query = "${allDelineatedValues('${abc}', ','):matches('[abc]')}";
assertEquals(ResultType.BOOLEAN, Query.getResultType(query));
- assertEquals("true", Query.evaluateExpressions(query, attributes, null));
- assertEquals("true", Query.evaluateExpressions(query, attributes, null));
- assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):matches('[abd]')}", attributes, null));
- assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):equals('a'):not()}", attributes, null));
+ assertEquals("true", Query.evaluateExpressions(query, registry, null));
+ assertEquals("true", Query.evaluateExpressions(query, registry, null));
+ assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):matches('[abd]')}",registry, null));
+ assertEquals("false", Query.evaluateExpressions("${allDelineatedValues('${abc}', ','):equals('a'):not()}", registry, null));
verifyEquals("${allDelineatedValues(${abc}, ','):matches('[abc]')}", attributes, true);
verifyEquals("${allDelineatedValues(${abc}, ','):matches('[abd]')}", attributes, false);
@@ -926,12 +957,13 @@ public class TestQuery {
attributes.put("xyz", "4132");
attributes.put("hello", "world!");
attributes.put("dotted", "abc.xyz");
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
- final String evaluated = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", attributes, null);
+ final String evaluated = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", registry, null);
assertEquals("true", evaluated);
attributes.put("end", "888");
- final String secondEvaluation = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", attributes, null);
+ final String secondEvaluation = Query.evaluateExpressions("${abc:matches('1234${end}4321')}", registry, null);
assertEquals("false", secondEvaluation);
verifyEquals("${dotted:matches('abc\\.xyz')}", attributes, true);
@@ -946,11 +978,13 @@ public class TestQuery {
attributes.put("hello", "world!");
attributes.put("dotted", "abc.xyz");
- final String evaluated = Query.evaluateExpressions("${abc:find('1234${end}4321')}", attributes, null);
+ final String evaluated = Query.evaluateExpressions("${abc:find('1234${end}4321')}", VariableRegistryFactory.getInstance(attributes), null);
assertEquals("true", evaluated);
attributes.put("end", "888");
- final String secondEvaluation = Query.evaluateExpressions("${abc:find('${end}4321')}", attributes, null);
+
+
+ final String secondEvaluation = Query.evaluateExpressions("${abc:find('${end}4321')}",VariableRegistryFactory.getInstance(attributes), null);
assertEquals("false", secondEvaluation);
verifyEquals("${dotted:find('\\.')}", attributes, true);
@@ -1131,7 +1165,7 @@ public class TestQuery {
attributes.put("b", "x");
attributes.put("abcxcba", "hello");
- final String evaluated = Query.evaluateExpressions("${ 'abc${b}cba':substring(0, 1) }", attributes, null);
+ final String evaluated = Query.evaluateExpressions("${ 'abc${b}cba':substring(0, 1) }", VariableRegistryFactory.getInstance(attributes), null);
assertEquals("h", evaluated);
}
@@ -1165,7 +1199,7 @@ public class TestQuery {
final List<String> expressions = Query.extractExpressions(query);
assertEquals(1, expressions.size());
assertEquals("${abc}", expressions.get(0));
- assertEquals("{ xyz }", Query.evaluateExpressions(query, attributes));
+ assertEquals("{ xyz }", Query.evaluateExpressions(query, VariableRegistryFactory.getInstance(attributes)));
}
@Test
@@ -1189,7 +1223,7 @@ public class TestQuery {
QueryResult<?> getResult(String expr, Map<String, String> attrs) {
final Query query = Query.compile(expr);
- final QueryResult<?> result = query.evaluate(attrs);
+ final QueryResult<?> result = query.evaluate(VariableRegistryFactory.getInstance(attrs));
return result;
}
@@ -1298,11 +1332,17 @@ public class TestQuery {
}
private void verifyEquals(final String expression, final Map<String, String> attributes, final Object expectedResult) {
+
+ VariableRegistry registry = VariableRegistryFactory.getInstance(attributes);
+ verifyEquals(expression,registry,expectedResult);
+ }
+
+ private void verifyEquals(final String expression, final VariableRegistry registry, final Object expectedResult) {
Query.validateExpression(expression, false);
- assertEquals(String.valueOf(expectedResult), Query.evaluateExpressions(expression, attributes, null));
+ assertEquals(String.valueOf(expectedResult), Query.evaluateExpressions(expression, registry, null));
final Query query = Query.compile(expression);
- final QueryResult<?> result = query.evaluate(attributes);
+ final QueryResult<?> result = query.evaluate(registry);
if (expectedResult instanceof Number) {
assertEquals(ResultType.NUMBER, result.getResultType());
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java
index 5acba8d..dbee665 100644
--- a/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java
+++ b/nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestStandardPreparedQuery.java
@@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import org.apache.nifi.registry.VariableRegistryFactory;
import org.junit.Ignore;
import org.junit.Test;
@@ -53,7 +54,7 @@ public class TestStandardPreparedQuery {
final StandardPreparedQuery prepared = (StandardPreparedQuery) Query.prepare("${xx}");
final long start = System.nanoTime();
for (int i = 0; i < 10000000; i++) {
- assertEquals("world", prepared.evaluateExpressions(attrs, null));
+ assertEquals("world", prepared.evaluateExpressions(VariableRegistryFactory.getInstance(attrs), null));
}
final long nanos = System.nanoTime() - start;
System.out.println(TimeUnit.NANOSECONDS.toMillis(nanos));
@@ -67,7 +68,7 @@ public class TestStandardPreparedQuery {
final long start = System.nanoTime();
for (int i = 0; i < 10000000; i++) {
- assertEquals("world", Query.evaluateExpressions("${xx}", attrs));
+ assertEquals("world", Query.evaluateExpressions("${xx}", VariableRegistryFactory.getInstance(attrs)));
}
final long nanos = System.nanoTime() - start;
System.out.println(TimeUnit.NANOSECONDS.toMillis(nanos));
@@ -85,7 +86,7 @@ public class TestStandardPreparedQuery {
}
private String evaluate(final String query, final Map<String, String> attrs) {
- final String evaluated = ((StandardPreparedQuery) Query.prepare(query)).evaluateExpressions(attrs, null);
+ final String evaluated = ((StandardPreparedQuery) Query.prepare(query)).evaluateExpressions(VariableRegistryFactory.getInstance(attrs), null);
return evaluated;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index 8c98c0b..ecef74b 100644
--- a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++ b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -200,6 +200,9 @@ public class NiFiProperties extends Properties {
public static final String STATE_MANAGEMENT_START_EMBEDDED_ZOOKEEPER = "nifi.state.management.embedded.zookeeper.start";
public static final String STATE_MANAGEMENT_ZOOKEEPER_PROPERTIES = "nifi.state.management.embedded.zookeeper.properties";
+ // expression language properties
+ public static final String VARIABLE_REGISTRY_PROPERTIES = "nifi.variable.registry.properties";
+
// defaults
public static final String DEFAULT_TITLE = "NiFi";
public static final Boolean DEFAULT_AUTO_RESUME_STATE = true;
@@ -248,11 +251,13 @@ public class NiFiProperties extends Properties {
public static final String DEFAULT_CLUSTER_MANAGER_SAFEMODE_DURATION = "0 sec";
// state management defaults
+
public static final String DEFAULT_STATE_MANAGEMENT_CONFIG_FILE = "conf/state-management.xml";
// Kerberos defaults
public static final String DEFAULT_KERBEROS_AUTHENTICATION_EXPIRATION = "12 hours";
+
private NiFiProperties() {
super();
}
@@ -1073,4 +1078,28 @@ public class NiFiProperties extends Properties {
public boolean isStartEmbeddedZooKeeper() {
return Boolean.parseBoolean(getProperty(STATE_MANAGEMENT_START_EMBEDDED_ZOOKEEPER));
}
+
+ public String getVariableRegistryProperties(){
+ return getProperty(VARIABLE_REGISTRY_PROPERTIES);
+ }
+
+ public Path[] getVariableRegistryPropertiesPaths() {
+ final List<Path> vrPropertiesPaths = new ArrayList<>();
+
+ final String vrPropertiesFiles = getVariableRegistryProperties();
+ if(!StringUtils.isEmpty(vrPropertiesFiles)) {
+
+ final List<String> vrPropertiesFileList = Arrays.asList(vrPropertiesFiles.split(","));
+
+ for(String propertiesFile : vrPropertiesFileList){
+ vrPropertiesPaths.add(Paths.get(propertiesFile));
+ }
+
+ return vrPropertiesPaths.toArray( new Path[vrPropertiesPaths.size()]);
+ } else {
+ return null;
+ }
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java
index c90e722..4c3b399 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockConfigurationContext.java
@@ -25,21 +25,26 @@ import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
+import org.apache.nifi.registry.VariableRegistry;
public class MockConfigurationContext implements ConfigurationContext {
private final Map<PropertyDescriptor, String> properties;
private final ControllerServiceLookup serviceLookup;
private final ControllerService service;
+ private final VariableRegistry variableRegistry;
- public MockConfigurationContext(final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup) {
- this(null, properties, serviceLookup);
+ public MockConfigurationContext(final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup,
+ final VariableRegistry variableRegistry) {
+ this(null, properties, serviceLookup, variableRegistry);
}
- public MockConfigurationContext(final ControllerService service, final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup) {
+ public MockConfigurationContext(final ControllerService service, final Map<PropertyDescriptor, String> properties, final ControllerServiceLookup serviceLookup,
+ final VariableRegistry variableRegistry) {
this.service = service;
this.properties = properties;
this.serviceLookup = serviceLookup;
+ this.variableRegistry = variableRegistry;
}
@Override
@@ -48,7 +53,7 @@ public class MockConfigurationContext implements ConfigurationContext {
if (value == null) {
value = getActualDescriptor(property).getDefaultValue();
}
- return new MockPropertyValue(value, serviceLookup);
+ return new MockPropertyValue(value, serviceLookup, variableRegistry);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
index 754bec0..3134c19 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
@@ -68,4 +68,6 @@ public class MockControllerServiceInitializationContext extends MockControllerSe
public StateManager getStateManager() {
return stateManager;
}
+
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
index 219ee24..fdb40a8 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
@@ -97,4 +97,5 @@ public abstract class MockControllerServiceLookup implements ControllerServiceLo
final ControllerServiceConfiguration status = controllerServiceMap.get(serviceIdentifier);
return status == null ? null : serviceIdentifier;
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java
index d50d708..c88de5f 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessContext.java
@@ -40,6 +40,7 @@ import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.SchedulingContext;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.state.MockStateManager;
import org.junit.Assert;
@@ -48,6 +49,7 @@ public class MockProcessContext extends MockControllerServiceLookup implements S
private final ConfigurableComponent component;
private final Map<PropertyDescriptor, String> properties = new HashMap<>();
private final StateManager stateManager;
+ private final VariableRegistry variableRegistry;
private String annotationData = null;
private boolean yieldCalled = false;
@@ -60,22 +62,24 @@ public class MockProcessContext extends MockControllerServiceLookup implements S
private volatile Set<Relationship> connections = new HashSet<>();
private volatile Set<Relationship> unavailableRelationships = new HashSet<>();
- public MockProcessContext(final ConfigurableComponent component) {
- this(component, new MockStateManager(component));
+ public MockProcessContext(final ConfigurableComponent component, final VariableRegistry variableRegistry) {
+ this(component, new MockStateManager(component), variableRegistry);
}
/**
* Creates a new MockProcessContext for the given Processor
*
* @param component being mocked
+ * @param variableRegistry variableRegistry
*/
- public MockProcessContext(final ConfigurableComponent component, final StateManager stateManager) {
+ public MockProcessContext(final ConfigurableComponent component, final StateManager stateManager, final VariableRegistry variableRegistry) {
this.component = Objects.requireNonNull(component);
this.stateManager = stateManager;
+ this.variableRegistry = variableRegistry;
}
- public MockProcessContext(final ControllerService component, final MockProcessContext context, final StateManager stateManager) {
- this(component, stateManager);
+ public MockProcessContext(final ControllerService component, final MockProcessContext context, final StateManager stateManager, final VariableRegistry variableRegistry) {
+ this(component, stateManager, variableRegistry);
try {
annotationData = context.getControllerServiceAnnotationData(component);
@@ -102,12 +106,13 @@ public class MockProcessContext extends MockControllerServiceLookup implements S
final String setPropertyValue = properties.get(descriptor);
final String propValue = (setPropertyValue == null) ? descriptor.getDefaultValue() : setPropertyValue;
- return new MockPropertyValue(propValue, this, (enableExpressionValidation && allowExpressionValidation) ? descriptor : null);
+
+ return new MockPropertyValue(propValue, this, variableRegistry, (enableExpressionValidation && allowExpressionValidation) ? descriptor : null);
}
@Override
public PropertyValue newPropertyValue(final String rawValue) {
- return new MockPropertyValue(rawValue, this);
+ return new MockPropertyValue(rawValue, this, variableRegistry);
}
public ValidationResult setProperty(final String propertyName, final String propertyValue) {
@@ -130,7 +135,7 @@ public class MockProcessContext extends MockControllerServiceLookup implements S
requireNonNull(value, "Cannot set property to null value; if the intent is to remove the property, call removeProperty instead");
final PropertyDescriptor fullyPopulatedDescriptor = component.getPropertyDescriptor(descriptor.getName());
- final ValidationResult result = fullyPopulatedDescriptor.validate(value, new MockValidationContext(this, stateManager));
+ final ValidationResult result = fullyPopulatedDescriptor.validate(value, new MockValidationContext(this, stateManager, variableRegistry));
String oldValue = properties.put(fullyPopulatedDescriptor, value);
if (oldValue == null) {
oldValue = fullyPopulatedDescriptor.getDefaultValue();
@@ -213,7 +218,7 @@ public class MockProcessContext extends MockControllerServiceLookup implements S
* non-null
*/
public Collection<ValidationResult> validate() {
- return component.validate(new MockValidationContext(this, stateManager));
+ return component.validate(new MockValidationContext(this, stateManager, variableRegistry));
}
public boolean isValid() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
index 6e94943..9ef513c 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
@@ -80,4 +80,5 @@ public class MockProcessorInitializationContext implements ProcessorInitializati
public boolean isControllerServiceEnabling(String serviceIdentifier) {
return context.isControllerServiceEnabling(serviceIdentifier);
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java
index 090a8eb..0fb4f89 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockPropertyValue.java
@@ -28,6 +28,7 @@ import org.apache.nifi.expression.AttributeValueDecorator;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
public class MockPropertyValue implements PropertyValue {
private final String rawValue;
@@ -35,24 +36,27 @@ public class MockPropertyValue implements PropertyValue {
private final ControllerServiceLookup serviceLookup;
private final PropertyDescriptor propertyDescriptor;
private final PropertyValue stdPropValue;
+ private final VariableRegistry variableRegistry;
private boolean expressionsEvaluated = false;
- public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup) {
- this(rawValue, serviceLookup, null);
+ public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final VariableRegistry variableRegistry) {
+ this(rawValue, serviceLookup, variableRegistry, null);
}
- public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PropertyDescriptor propertyDescriptor) {
- this(rawValue, serviceLookup, propertyDescriptor, false);
+ public MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, VariableRegistry variableRegistry, final PropertyDescriptor propertyDescriptor) {
+ this(rawValue, serviceLookup, propertyDescriptor, false, variableRegistry);
}
- private MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PropertyDescriptor propertyDescriptor, final boolean alreadyEvaluated) {
- this.stdPropValue = new StandardPropertyValue(rawValue, serviceLookup);
+ private MockPropertyValue(final String rawValue, final ControllerServiceLookup serviceLookup, final PropertyDescriptor propertyDescriptor, final boolean alreadyEvaluated,
+ final VariableRegistry variableRegistry) {
+ this.stdPropValue = new StandardPropertyValue(rawValue, serviceLookup, variableRegistry);
this.rawValue = rawValue;
this.serviceLookup = serviceLookup;
this.expectExpressions = propertyDescriptor == null ? null : propertyDescriptor.isExpressionLanguageSupported();
this.propertyDescriptor = propertyDescriptor;
this.expressionsEvaluated = alreadyEvaluated;
+ this.variableRegistry = variableRegistry;
}
@@ -165,7 +169,7 @@ public class MockPropertyValue implements PropertyValue {
}
final PropertyValue newValue = stdPropValue.evaluateAttributeExpressions(flowFile, additionalAttributes, decorator);
- return new MockPropertyValue(newValue.getValue(), serviceLookup, propertyDescriptor, true);
+ return new MockPropertyValue(newValue.getValue(), serviceLookup, propertyDescriptor, true, variableRegistry);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java
index 33719ec..da43c62 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockReportingContext.java
@@ -27,6 +27,7 @@ import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinFactory;
import org.apache.nifi.reporting.BulletinRepository;
@@ -39,12 +40,14 @@ public class MockReportingContext extends MockControllerServiceLookup implements
private final MockEventAccess eventAccess = new MockEventAccess();
private final Map<PropertyDescriptor, String> properties = new HashMap<>();
private final StateManager stateManager;
+ private final VariableRegistry variableRegistry;
private final Map<String, List<Bulletin>> componentBulletinsCreated = new HashMap<>();
- public MockReportingContext(final Map<String, ControllerService> controllerServices, final StateManager stateManager) {
+ public MockReportingContext(final Map<String, ControllerService> controllerServices, final StateManager stateManager, final VariableRegistry variableRegistry) {
this.controllerServices = new HashMap<>();
this.stateManager = stateManager;
+ this.variableRegistry = variableRegistry;
for (final Map.Entry<String, ControllerService> entry : controllerServices.entrySet()) {
this.controllerServices.put(entry.getKey(), new ControllerServiceConfiguration(entry.getValue()));
}
@@ -58,7 +61,7 @@ public class MockReportingContext extends MockControllerServiceLookup implements
@Override
public PropertyValue getProperty(final PropertyDescriptor property) {
final String configuredValue = properties.get(property);
- return new MockPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this);
+ return new MockPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, this, variableRegistry);
}
public void setProperty(final String propertyName, final String value) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
index 6442778..322288b 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
@@ -31,16 +31,20 @@ import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.expression.ExpressionLanguageCompiler;
+import org.apache.nifi.registry.VariableRegistry;
+
public class MockValidationContext implements ValidationContext, ControllerServiceLookup {
private final MockProcessContext context;
private final Map<String, Boolean> expressionLanguageSupported;
private final StateManager stateManager;
+ private final VariableRegistry variableRegistry;
- public MockValidationContext(final MockProcessContext processContext, final StateManager stateManager) {
+ public MockValidationContext(final MockProcessContext processContext, final StateManager stateManager, final VariableRegistry variableRegistry) {
this.context = processContext;
this.stateManager = stateManager;
+ this.variableRegistry = variableRegistry;
final Map<PropertyDescriptor, String> properties = processContext.getProperties();
expressionLanguageSupported = new HashMap<>(properties.size());
@@ -56,18 +60,18 @@ public class MockValidationContext implements ValidationContext, ControllerServi
@Override
public PropertyValue newPropertyValue(final String rawValue) {
- return new MockPropertyValue(rawValue, this);
+ return new MockPropertyValue(rawValue, this, variableRegistry);
}
@Override
public ExpressionLanguageCompiler newExpressionLanguageCompiler() {
- return new StandardExpressionLanguageCompiler();
+ return new StandardExpressionLanguageCompiler(variableRegistry);
}
@Override
public ValidationContext getControllerServiceValidationContext(final ControllerService controllerService) {
- final MockProcessContext serviceProcessContext = new MockProcessContext(controllerService, context, stateManager);
- return new MockValidationContext(serviceProcessContext, stateManager);
+ final MockProcessContext serviceProcessContext = new MockProcessContext(controllerService, context, stateManager, variableRegistry);
+ return new MockValidationContext(serviceProcessContext, stateManager, variableRegistry);
}
@Override
@@ -136,4 +140,5 @@ public class MockValidationContext implements ValidationContext, ControllerServi
final Boolean supported = expressionLanguageSupported.get(propertyName);
return Boolean.TRUE.equals(supported);
}
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
index 43621e0..6206c02 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
@@ -69,6 +69,7 @@ import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceReporter;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.state.MockStateManager;
import org.junit.Assert;
@@ -84,6 +85,7 @@ public class StandardProcessorTestRunner implements TestRunner {
private final boolean triggerSerially;
private final MockStateManager processorStateManager;
private final Map<String, MockStateManager> controllerServiceStateManagers = new HashMap<>();
+ private final VariableRegistry variableRegistry;
private int numThreads = 1;
private final AtomicInteger invocations = new AtomicInteger(0);
@@ -100,14 +102,15 @@ public class StandardProcessorTestRunner implements TestRunner {
populateDeprecatedMethods();
}
- StandardProcessorTestRunner(final Processor processor) {
+ StandardProcessorTestRunner(final Processor processor, final VariableRegistry variableRegistry) {
this.processor = processor;
this.idGenerator = new AtomicLong(0L);
this.sharedState = new SharedSessionState(processor, idGenerator);
this.flowFileQueue = sharedState.getFlowFileQueue();
this.sessionFactory = new MockSessionFactory(sharedState, processor);
this.processorStateManager = new MockStateManager(processor);
- this.context = new MockProcessContext(processor, processorStateManager);
+ this.variableRegistry = variableRegistry;
+ this.context = new MockProcessContext(processor, processorStateManager, variableRegistry);
detectDeprecatedAnnotations(processor);
@@ -670,7 +673,7 @@ public class StandardProcessorTestRunner implements TestRunner {
throw new IllegalStateException("Controller Service has not been added to this TestRunner via the #addControllerService method");
}
- final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager).getControllerServiceValidationContext(service);
+ final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager, variableRegistry).getControllerServiceValidationContext(service);
final Collection<ValidationResult> results = context.getControllerService(service.getIdentifier()).validate(validationContext);
for (final ValidationResult result : results) {
@@ -689,7 +692,7 @@ public class StandardProcessorTestRunner implements TestRunner {
throw new IllegalStateException("Controller Service has not been added to this TestRunner via the #addControllerService method");
}
- final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager).getControllerServiceValidationContext(service);
+ final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager, variableRegistry).getControllerServiceValidationContext(service);
final Collection<ValidationResult> results = context.getControllerService(service.getIdentifier()).validate(validationContext);
for (final ValidationResult result : results) {
@@ -732,7 +735,7 @@ public class StandardProcessorTestRunner implements TestRunner {
}
try {
- final ConfigurationContext configContext = new MockConfigurationContext(service, configuration.getProperties(), context);
+ final ConfigurationContext configContext = new MockConfigurationContext(service, configuration.getProperties(), context,variableRegistry);
ReflectionUtils.invokeMethodsWithAnnotation(OnEnabled.class, service, configContext);
} catch (final InvocationTargetException ite) {
ite.getCause().printStackTrace();
@@ -804,7 +807,7 @@ public class StandardProcessorTestRunner implements TestRunner {
final Map<PropertyDescriptor, String> curProps = configuration.getProperties();
final Map<PropertyDescriptor, String> updatedProps = new HashMap<>(curProps);
- final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager).getControllerServiceValidationContext(service);
+ final ValidationContext validationContext = new MockValidationContext(context, serviceStateManager, variableRegistry).getControllerServiceValidationContext(service);
final ValidationResult validationResult = property.validate(value, validationContext);
updatedProps.put(property, value);
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
index f2b0b23..f7153bf 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
@@ -17,11 +17,12 @@
package org.apache.nifi.util;
import org.apache.nifi.processor.Processor;
+import org.apache.nifi.registry.VariableRegistryUtils;
public class TestRunners {
public static TestRunner newTestRunner(final Processor processor) {
- return new StandardProcessorTestRunner(processor);
+ return new StandardProcessorTestRunner(processor, VariableRegistryUtils.createVariableRegistry());
}
public static TestRunner newTestRunner(final Class<? extends Processor> processorClass) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java b/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java
index 6b403af..24e9307 100644
--- a/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java
+++ b/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java
@@ -20,6 +20,7 @@ import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.junit.Assert;
import org.junit.Test;
@@ -31,7 +32,7 @@ public class CurrentTestStandardProcessorTestRunner {
@Test
public void testOnScheduledCalledAfterRunFinished() {
SlowRunProcessor processor = new SlowRunProcessor();
- StandardProcessorTestRunner runner = new StandardProcessorTestRunner(processor);
+ StandardProcessorTestRunner runner = new StandardProcessorTestRunner(processor, VariableRegistryUtils.createVariableRegistry());
final int iterations = 5;
runner.run(iterations);
// if the counter is not equal to iterations, the the processor must have been unscheduled
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java b/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java
index d48af63..e033451 100644
--- a/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java
+++ b/nifi-mock/src/test/java/org/apache/nifi/util/TestMockProcessContext.java
@@ -32,6 +32,7 @@ import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.junit.Test;
public class TestMockProcessContext {
@@ -39,7 +40,7 @@ public class TestMockProcessContext {
@Test
public void testRemoveProperty() {
final DummyProcessor proc = new DummyProcessor();
- final MockProcessContext context = new MockProcessContext(proc);
+ final MockProcessContext context = new MockProcessContext(proc, VariableRegistryUtils.createVariableRegistry());
context.setProperty(DummyProcessor.REQUIRED_PROP, "req-value");
context.setProperty(DummyProcessor.OPTIONAL_PROP, "opt-value");
context.setProperty(DummyProcessor.DEFAULTED_PROP, "custom-value");
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java
index c2ab654..89d2006 100644
--- a/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java
+++ b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java
@@ -30,6 +30,7 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.junit.Ignore;
import org.junit.Test;
@@ -106,25 +107,25 @@ public class TestStandardProcessorTestRunner {
@Test(expected = AssertionError.class)
@Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods")
public void testFailOnDeprecatedTypeAnnotation() {
- new StandardProcessorTestRunner(new DeprecatedAnnotation());
+ new StandardProcessorTestRunner(new DeprecatedAnnotation(), VariableRegistryUtils.createVariableRegistry());
}
@Test
@Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods")
public void testDoesNotFailOnNonDeprecatedTypeAnnotation() {
- new StandardProcessorTestRunner(new NewAnnotation());
+ new StandardProcessorTestRunner(new NewAnnotation(), VariableRegistryUtils.createVariableRegistry());
}
@Test(expected = AssertionError.class)
@Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods")
public void testFailOnDeprecatedMethodAnnotation() {
- new StandardProcessorTestRunner(new DeprecatedMethodAnnotation());
+ new StandardProcessorTestRunner(new DeprecatedMethodAnnotation(), VariableRegistryUtils.createVariableRegistry());
}
@Test
@Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods")
public void testDoesNotFailOnNonDeprecatedMethodAnnotation() {
- new StandardProcessorTestRunner(new NewMethodAnnotation());
+ new StandardProcessorTestRunner(new NewMethodAnnotation(), VariableRegistryUtils.createVariableRegistry());
}
@SuppressWarnings("deprecation")
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java b/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java
index ce5f8a6..3688b7c 100644
--- a/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java
+++ b/nifi-nar-bundles/nifi-ambari-bundle/nifi-ambari-reporting-task/src/test/java/org/apache/nifi/reporting/ambari/TestAmbariReportingTask.java
@@ -17,13 +17,17 @@
package org.apache.nifi.reporting.ambari;
import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.reporting.EventAccess;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.reporting.ReportingContext;
import org.apache.nifi.reporting.ReportingInitializationContext;
+import org.apache.nifi.util.MockControllerServiceLookup;
import org.apache.nifi.util.MockPropertyValue;
import org.junit.Before;
import org.junit.Test;
@@ -43,6 +47,7 @@ import java.util.UUID;
public class TestAmbariReportingTask {
private ProcessGroupStatus status;
+ private VariableRegistry variableRegistry;
@Before
public void setup() {
@@ -73,6 +78,7 @@ public class TestAmbariReportingTask {
Collection<ProcessGroupStatus> groupStatuses = new ArrayList<>();
groupStatuses.add(groupStatus);
status.setProcessGroupStatus(groupStatuses);
+ variableRegistry = VariableRegistryUtils.createVariableRegistry();
}
@Test
@@ -101,15 +107,17 @@ public class TestAmbariReportingTask {
// mock the ConfigurationContext for setup(...)
final ConfigurationContext configurationContext = Mockito.mock(ConfigurationContext.class);
+ final ControllerServiceLookup serviceLookup = new MockControllerServiceLookup() {};
// mock the ReportingContext for onTrigger(...)
final ReportingContext context = Mockito.mock(ReportingContext.class);
Mockito.when(context.getProperty(AmbariReportingTask.METRICS_COLLECTOR_URL))
- .thenReturn(new MockPropertyValue(metricsUrl, null));
+ .thenReturn(new MockPropertyValue(metricsUrl, serviceLookup, variableRegistry));
Mockito.when(context.getProperty(AmbariReportingTask.APPLICATION_ID))
- .thenReturn(new MockPropertyValue(applicationId, null));
+ .thenReturn(new MockPropertyValue(applicationId, serviceLookup, variableRegistry));
Mockito.when(context.getProperty(AmbariReportingTask.HOSTNAME))
- .thenReturn(new MockPropertyValue(hostName, null));
+ .thenReturn(new MockPropertyValue(hostName, serviceLookup, variableRegistry));
+
final EventAccess eventAccess = Mockito.mock(EventAccess.class);
Mockito.when(context.getEventAccess()).thenReturn(eventAccess);
@@ -121,7 +129,6 @@ public class TestAmbariReportingTask {
task.setup(configurationContext);
task.onTrigger(context);
}
-
// override the creation of the client to provide a mock
private class TestableAmbariReportingTask extends AmbariReportingTask {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java
index cb928fa..4bc8d2c 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearch.java
@@ -18,6 +18,8 @@ package org.apache.nifi.processors.elasticsearch;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.ssl.SSLContextService;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.MockProcessContext;
@@ -62,11 +64,13 @@ public class TestFetchElasticsearch {
private InputStream docExample;
private TestRunner runner;
+ private VariableRegistry variableRegistry;
@Before
public void setUp() throws IOException {
ClassLoader classloader = Thread.currentThread().getContextClassLoader();
docExample = classloader.getResourceAsStream("DocumentExample.json");
+ variableRegistry = VariableRegistryUtils.createVariableRegistry();
}
@@ -216,7 +220,7 @@ public class TestFetchElasticsearch {
}
};
- MockProcessContext context = new MockProcessContext(processor);
+ MockProcessContext context = new MockProcessContext(processor, variableRegistry);
processor.initialize(new MockProcessorInitializationContext(processor, context));
processor.callCreateElasticsearchClient(context);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java
index 7f176b0..1ff74d5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredReportingContext.java
@@ -34,6 +34,7 @@ import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
import org.apache.nifi.controller.status.RemoteProcessGroupStatus;
import org.apache.nifi.events.BulletinFactory;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinRepository;
import org.apache.nifi.reporting.ComponentType;
@@ -49,14 +50,16 @@ public class ClusteredReportingContext implements ReportingContext {
private final Map<PropertyDescriptor, String> properties;
private final Map<PropertyDescriptor, PreparedQuery> preparedQueries;
private final StateManager stateManager;
+ private final VariableRegistry variableRegistry;
public ClusteredReportingContext(final EventAccess eventAccess, final BulletinRepository bulletinRepository, final Map<PropertyDescriptor, String> properties,
- final ControllerServiceProvider serviceProvider, final StateManager stateManager) {
+ final ControllerServiceProvider serviceProvider, final StateManager stateManager, final VariableRegistry variableRegistry) {
this.eventAccess = eventAccess;
this.bulletinRepository = bulletinRepository;
this.properties = Collections.unmodifiableMap(properties);
this.serviceProvider = serviceProvider;
this.stateManager = stateManager;
+ this.variableRegistry = variableRegistry;
preparedQueries = new HashMap<>();
for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) {
@@ -104,7 +107,7 @@ public class ClusteredReportingContext implements ReportingContext {
@Override
public PropertyValue getProperty(final PropertyDescriptor property) {
final String configuredValue = properties.get(property);
- return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceProvider, preparedQueries.get(property));
+ return new StandardPropertyValue(configuredValue == null ? property.getDefaultValue() : configuredValue, serviceProvider, preparedQueries.get(property), variableRegistry);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
index 7bf8de3..b74ea50 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
@@ -181,6 +181,9 @@ import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.nar.NarThreadContextClassLoader;
import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardValidationContextFactory;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryFactory;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.remote.RemoteResourceManager;
import org.apache.nifi.remote.RemoteSiteListener;
import org.apache.nifi.remote.SocketRemoteSiteListener;
@@ -382,6 +385,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
private final StandardProcessScheduler processScheduler;
private final StateManagerProvider stateManagerProvider;
private final long componentStatusSnapshotMillis;
+ private final VariableRegistry variableRegistry;
public WebClusterManager(final HttpRequestReplicator httpRequestReplicator, final HttpResponseMapper httpResponseMapper,
@@ -412,6 +416,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
this.optimisticLockingManager = optimisticLockingManager;
senderListener.addHandler(this);
senderListener.setBulletinRepository(bulletinRepository);
+ this.variableRegistry = createVariableRegistry(properties);
final String snapshotFrequency = properties.getProperty(NiFiProperties.COMPONENT_STATUS_SNAPSHOT_FREQUENCY, NiFiProperties.DEFAULT_COMPONENT_STATUS_SNAPSHOT_FREQUENCY);
long snapshotMillis;
@@ -446,7 +451,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
reportingTaskEngine = new FlowEngine(8, "Reporting Task Thread");
try {
- this.stateManagerProvider = StandardStateManagerProvider.create(properties);
+ this.stateManagerProvider = StandardStateManagerProvider.create(properties,variableRegistry);
} catch (final IOException e) {
throw new RuntimeException(e);
}
@@ -455,17 +460,19 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
@Override
public void heartbeat() {
}
- }, this, encryptor, stateManagerProvider);
+ }, this, encryptor, stateManagerProvider, variableRegistry);
// When we construct the scheduling agents, we can pass null for a lot of the arguments because we are only
// going to be scheduling Reporting Tasks. Otherwise, it would not be okay.
- processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, new TimerDrivenSchedulingAgent(null, reportingTaskEngine, null, encryptor));
- processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, new QuartzSchedulingAgent(null, reportingTaskEngine, null, encryptor));
+ processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, new TimerDrivenSchedulingAgent(null, reportingTaskEngine, null, encryptor, variableRegistry));
+ processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, new QuartzSchedulingAgent(null, reportingTaskEngine, null, encryptor, variableRegistry));
processScheduler.setMaxThreadCount(SchedulingStrategy.TIMER_DRIVEN, 10);
processScheduler.setMaxThreadCount(SchedulingStrategy.CRON_DRIVEN, 10);
processScheduler.scheduleFrameworkTask(new CaptureComponentMetrics(), "Capture Component Metrics", componentStatusSnapshotMillis, componentStatusSnapshotMillis, TimeUnit.MILLISECONDS);
- controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider);
+ controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider, variableRegistry);
+
+
}
public void start() throws IOException {
@@ -956,6 +963,17 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
reconnectionThread.start();
}
+ private VariableRegistry createVariableRegistry(NiFiProperties properties){
+ VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ try {
+ VariableRegistry customRegistry = VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths());
+ variableRegistry.addRegistry(customRegistry);
+ } catch (IOException ioe){
+ logger.error("Exception thrown while attempting to add properties to registry",ioe);
+ }
+ return variableRegistry;
+ }
+
private Map<String, ReportingTaskNode> loadReportingTasks(final byte[] serialized) {
final Map<String, ReportingTaskNode> tasks = new HashMap<>();
@@ -1105,10 +1123,10 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
}
}
- final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this);
+ final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this, variableRegistry);
final ReportingTaskNode taskNode = new ClusteredReportingTaskNode(task, id, processScheduler,
new ClusteredEventAccess(this, auditService), bulletinRepository, controllerServiceProvider,
- validationContextFactory, stateManagerProvider.getStateManager(id));
+ validationContextFactory, stateManagerProvider.getStateManager(id), null);
taskNode.setName(task.getClass().getSimpleName());
reportingTasks.put(id, taskNode);
@@ -4601,6 +4619,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return controllerServiceProvider.getControllerServiceIdentifiers(serviceType);
}
+
/**
* Captures snapshots of components' metrics
*/
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java
index a23cfdd..0adf41a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/controller/reporting/ClusteredReportingTaskNode.java
@@ -21,6 +21,7 @@ import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ValidationContextFactory;
import org.apache.nifi.controller.service.ControllerServiceProvider;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.BulletinRepository;
import org.apache.nifi.reporting.EventAccess;
import org.apache.nifi.reporting.ReportingContext;
@@ -33,10 +34,12 @@ public class ClusteredReportingTaskNode extends AbstractReportingTaskNode {
private final ControllerServiceProvider serviceProvider;
private final StateManager stateManager;
+
public ClusteredReportingTaskNode(final ReportingTask reportingTask, final String id, final ProcessScheduler scheduler,
- final EventAccess eventAccess, final BulletinRepository bulletinRepository, final ControllerServiceProvider serviceProvider,
- final ValidationContextFactory validationContextFactory, final StateManager stateManager) {
- super(reportingTask, id, serviceProvider, scheduler, validationContextFactory);
+ final EventAccess eventAccess, final BulletinRepository bulletinRepository, final ControllerServiceProvider serviceProvider,
+ final ValidationContextFactory validationContextFactory, final StateManager stateManager,
+ final VariableRegistry variableRegistry) {
+ super(reportingTask, id, serviceProvider, scheduler, validationContextFactory, variableRegistry);
this.eventAccess = eventAccess;
this.bulletinRepository = bulletinRepository;
@@ -46,7 +49,7 @@ public class ClusteredReportingTaskNode extends AbstractReportingTaskNode {
@Override
public ReportingContext getReportingContext() {
- return new ClusteredReportingContext(eventAccess, bulletinRepository, getProperties(), serviceProvider, stateManager);
+ return new ClusteredReportingContext(eventAccess, bulletinRepository, getProperties(), serviceProvider, stateManager, variableRegistry);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index a5f2ee0..7c8763b 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -140,6 +140,9 @@ import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventRepository;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.provenance.StandardProvenanceEventRecord;
+import org.apache.nifi.registry.VariableRegistry;
+import org.apache.nifi.registry.VariableRegistryFactory;
+import org.apache.nifi.registry.VariableRegistryUtils;
import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.remote.RemoteResourceManager;
import org.apache.nifi.remote.RemoteSiteListener;
@@ -263,6 +266,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
private final StateManagerProvider stateManagerProvider;
private final long systemStartTime = System.currentTimeMillis(); // time at which the node was started
private final ConcurrentMap<String, ReportingTaskNode> reportingTasks = new ConcurrentHashMap<>();
+ private final VariableRegistry variableRegistry;
private volatile ZooKeeperStateServer zooKeeperStateServer;
@@ -422,6 +426,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
bulletinRepository = new VolatileBulletinRepository();
nodeBulletinSubscriber = new AtomicReference<>();
+ variableRegistry = createVariableRegistry(properties);
+
try {
this.provenanceEventRepository = createProvenanceRepository(properties);
this.provenanceEventRepository.initialize(createEventReporter(bulletinRepository));
@@ -436,21 +442,21 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
}
try {
- this.stateManagerProvider = StandardStateManagerProvider.create(properties);
+ this.stateManagerProvider = StandardStateManagerProvider.create(properties,variableRegistry);
} catch (final IOException e) {
throw new RuntimeException(e);
}
- processScheduler = new StandardProcessScheduler(this, this, encryptor, stateManagerProvider);
+ processScheduler = new StandardProcessScheduler(this, this, encryptor, stateManagerProvider, variableRegistry);
eventDrivenWorkerQueue = new EventDrivenWorkerQueue(false, false, processScheduler);
- controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider);
+ controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider, variableRegistry);
final ProcessContextFactory contextFactory = new ProcessContextFactory(contentRepository, flowFileRepository, flowFileEventRepository, counterRepositoryRef.get(), provenanceEventRepository);
processScheduler.setSchedulingAgent(SchedulingStrategy.EVENT_DRIVEN, new EventDrivenSchedulingAgent(
- eventDrivenEngineRef.get(), this, stateManagerProvider, eventDrivenWorkerQueue, contextFactory, maxEventDrivenThreads.get(), encryptor));
+ eventDrivenEngineRef.get(), this, stateManagerProvider, eventDrivenWorkerQueue, contextFactory, maxEventDrivenThreads.get(), encryptor, variableRegistry));
- final QuartzSchedulingAgent quartzSchedulingAgent = new QuartzSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor);
- final TimerDrivenSchedulingAgent timerDrivenAgent = new TimerDrivenSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor);
+ final QuartzSchedulingAgent quartzSchedulingAgent = new QuartzSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor, variableRegistry);
+ final TimerDrivenSchedulingAgent timerDrivenAgent = new TimerDrivenSchedulingAgent(this, timerDrivenEngineRef.get(), contextFactory, encryptor, variableRegistry);
processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, timerDrivenAgent);
processScheduler.setSchedulingAgent(SchedulingStrategy.PRIMARY_NODE_ONLY, timerDrivenAgent);
processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, quartzSchedulingAgent);
@@ -491,7 +497,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
this.snippetManager = new SnippetManager();
- rootGroup = new StandardProcessGroup(UUID.randomUUID().toString(), this, processScheduler, properties, encryptor, this);
+ rootGroup = new StandardProcessGroup(UUID.randomUUID().toString(), this, processScheduler, properties, encryptor, this, variableRegistry);
rootGroup.setName(DEFAULT_ROOT_GROUP_NAME);
instanceId = UUID.randomUUID().toString();
@@ -538,8 +544,12 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
}, snapshotMillis, snapshotMillis, TimeUnit.MILLISECONDS);
heartbeatBeanRef.set(new HeartbeatBean(rootGroup, false, false));
+
+
}
+
+
private static FlowFileRepository createFlowFileRepository(final NiFiProperties properties, final ResourceClaimManager contentClaimManager) {
final String implementationClassName = properties.getProperty(NiFiProperties.FLOWFILE_REPOSITORY_IMPLEMENTATION, DEFAULT_FLOWFILE_REPO_IMPLEMENTATION);
if (implementationClassName == null) {
@@ -583,6 +593,18 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
};
}
+ private static VariableRegistry createVariableRegistry(final NiFiProperties properties){
+ VariableRegistry variableRegistry = VariableRegistryUtils.createVariableRegistry();
+ try {
+ VariableRegistry customRegistry = VariableRegistryFactory.getPropertiesInstance(properties.getVariableRegistryPropertiesPaths());
+ variableRegistry.addRegistry(customRegistry);
+ } catch (IOException ioe){
+ LOG.error("Exception thrown while attempting to add properties to registry",ioe);
+ }
+
+ return variableRegistry;
+ }
+
public void initializeFlow() throws IOException {
writeLock.lock();
try {
@@ -903,7 +925,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
* @throws NullPointerException if the argument is null
*/
public ProcessGroup createProcessGroup(final String id) {
- return new StandardProcessGroup(requireNonNull(id).intern(), this, processScheduler, properties, encryptor, this);
+ return new StandardProcessGroup(requireNonNull(id).intern(), this, processScheduler, properties, encryptor, this, variableRegistry);
}
/**
@@ -937,7 +959,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
public ProcessorNode createProcessor(final String type, String id, final boolean firstTimeAdded) throws ProcessorInstantiationException {
id = id.intern();
final Processor processor = instantiateProcessor(type, id);
- final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider);
+ final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider, variableRegistry);
final ProcessorNode procNode = new StandardProcessorNode(processor, id, validationContextFactory, processScheduler, controllerServiceProvider);
final LogRepository logRepository = LogRepositoryFactory.getRepository(id);
@@ -1195,7 +1217,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
// invoke any methods annotated with @OnShutdown on Controller Services
for (final ControllerServiceNode serviceNode : getAllControllerServices()) {
try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) {
- final ConfigurationContext configContext = new StandardConfigurationContext(serviceNode, controllerServiceProvider, null);
+ final ConfigurationContext configContext = new StandardConfigurationContext(serviceNode, controllerServiceProvider, null, variableRegistry);
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnShutdown.class, serviceNode.getControllerServiceImplementation(), configContext);
}
}
@@ -2666,8 +2688,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
}
}
- final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider);
- final ReportingTaskNode taskNode = new StandardReportingTaskNode(task, id, this, processScheduler, validationContextFactory);
+ final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(controllerServiceProvider, variableRegistry);
+ final ReportingTaskNode taskNode = new StandardReportingTaskNode(task, id, this, processScheduler, validationContextFactory, variableRegistry);
taskNode.setName(task.getClass().getSimpleName());
if (firstTimeAdded) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c600f150/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
index c3eb0a0..31c2242 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
@@ -36,6 +36,7 @@ import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.controller.service.StandardConfigurationContext;
import org.apache.nifi.nar.NarCloseable;
+import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.reporting.ReportingTask;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.FormatUtils;
@@ -47,19 +48,23 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon
private final ProcessScheduler processScheduler;
private final ControllerServiceLookup serviceLookup;
+
private final AtomicReference<SchedulingStrategy> schedulingStrategy = new AtomicReference<>(SchedulingStrategy.TIMER_DRIVEN);
private final AtomicReference<String> schedulingPeriod = new AtomicReference<>("5 mins");
private volatile String comment;
private volatile ScheduledState scheduledState = ScheduledState.STOPPED;
+ protected final VariableRegistry variableRegistry;
+
public AbstractReportingTaskNode(final ReportingTask reportingTask, final String id,
- final ControllerServiceProvider controllerServiceProvider, final ProcessScheduler processScheduler,
- final ValidationContextFactory validationContextFactory) {
+ final ControllerServiceProvider controllerServiceProvider, final ProcessScheduler processScheduler,
+ final ValidationContextFactory validationContextFactory, final VariableRegistry variableRegistry) {
super(reportingTask, id, validationContextFactory, controllerServiceProvider);
this.reportingTask = reportingTask;
this.processScheduler = processScheduler;
this.serviceLookup = controllerServiceProvider;
+ this.variableRegistry = variableRegistry;
}
@Override
@@ -104,7 +109,7 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon
@Override
public ConfigurationContext getConfigurationContext() {
- return new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod());
+ return new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod(), variableRegistry);
}
@Override
@@ -146,7 +151,7 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon
// We need to invoke any method annotation with the OnConfigured annotation in order to
// maintain backward compatibility. This will be removed when we remove the old, deprecated annotations.
try (final NarCloseable x = NarCloseable.withNarLoader()) {
- final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod());
+ final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceLookup, getSchedulingPeriod(), variableRegistry);
ReflectionUtils.invokeMethodsWithAnnotation(OnConfigured.class, reportingTask, configContext);
} catch (final Exception e) {
throw new ComponentLifeCycleException("Failed to invoke On-Configured Lifecycle methods of " + reportingTask, e);