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/08/04 18:23:38 UTC
nifi git commit: NIFI-2407 Implements EL support on some properties
of processors/services - Added EL support to "Directory" property of ListHDFS
and GetHDFS processors - Added EL support to "Destination" property for
ConsumeJMS and PublishJMS processo
Repository: nifi
Updated Branches:
refs/heads/master 0d730c5cc -> 1ea49ad8c
NIFI-2407 Implements EL support on some properties of processors/services
- Added EL support to "Directory" property of ListHDFS and GetHDFS processors
- Added EL support to "Destination" property for ConsumeJMS and PublishJMS processors
- Added EL support to "MQ ConnectionFactory Implementation", "MQ client library path", "Broker URI" properties for the JMS Connection Factory Provider
- Added EL support to "Database Connection URL", "Database Driver Class Name", "DB Driver jar url", "DB username", and "DB password" properties for the DBCP Connection Pool
- Removed NotificationServiceManagerSpec.groovy, previous PR from Joe Witt covers the needed testing for Variable Registry/Value Lookup at this time
This closes #778.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/1ea49ad8
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/1ea49ad8
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/1ea49ad8
Branch: refs/heads/master
Commit: 1ea49ad8cff63dd9e31fc4d1f6907222563306ea
Parents: 0d730c5
Author: Jeff Storck <jt...@gmail.com>
Authored: Wed Jul 27 11:51:15 2016 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Thu Aug 4 14:23:29 2016 -0400
----------------------------------------------------------------------
.../NotificationServiceManagerSpec.groovy | 50 --------------------
.../hadoop/AbstractHadoopProcessor.java | 15 ++++--
.../apache/nifi/processors/hadoop/GetHDFS.java | 7 +--
.../apache/nifi/processors/hadoop/ListHDFS.java | 3 +-
.../nifi/processors/hadoop/PutHDFSTest.java | 5 +-
.../jms/cf/JMSConnectionFactoryProvider.java | 8 ++--
.../JMSConnectionFactoryProviderDefinition.java | 3 ++
.../cf/JMSConnectionFactoryProviderTest.java | 43 ++++++++---------
.../jms/processors/AbstractJMSProcessor.java | 3 +-
.../apache/nifi/jms/processors/ConsumeJMS.java | 2 +-
.../apache/nifi/jms/processors/PublishJMS.java | 2 +-
.../nifi/jms/processors/ConsumeJMSTest.java | 10 ++--
.../nifi/jms/processors/PublishJMSTest.java | 25 +++++-----
.../apache/nifi/dbcp/DBCPConnectionPool.java | 39 ++++++++-------
.../org/apache/nifi/dbcp/DBCPServiceTest.java | 24 +++++-----
15 files changed, 102 insertions(+), 137 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-bootstrap/src/test/groovy/org/apache/nifi/bootstrap/NotificationServiceManagerSpec.groovy
----------------------------------------------------------------------
diff --git a/nifi-bootstrap/src/test/groovy/org/apache/nifi/bootstrap/NotificationServiceManagerSpec.groovy b/nifi-bootstrap/src/test/groovy/org/apache/nifi/bootstrap/NotificationServiceManagerSpec.groovy
deleted file mode 100644
index 7bd4c52..0000000
--- a/nifi-bootstrap/src/test/groovy/org/apache/nifi/bootstrap/NotificationServiceManagerSpec.groovy
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.bootstrap
-
-import org.apache.nifi.bootstrap.notification.NotificationType
-import org.apache.nifi.registry.VariableRegistry
-import spock.lang.Specification
-import java.nio.file.Paths
-
-class NotificationServiceManagerSpec extends Specification{
-
- def setupSpec(){
- }
-
- def "should acess variable registry to replace EL values"(){
-
- given:
- def mockRegistry = Mock(VariableRegistry.class)
- def notificationServiceManager = new NotificationServiceManager(mockRegistry);
- def file = Paths.get("src/test/resources/notification-services.xml").toFile()
- notificationServiceManager.loadNotificationServices(file)
- //testing with stopped becasue it will block until method is completed
- notificationServiceManager.registerNotificationService(NotificationType.NIFI_STOPPED,"custom-notification")
-
- when:
- notificationServiceManager.notify(NotificationType.NIFI_STOPPED,"NiFi Stopped","NiFi Stopped")
-
- then:
- 6 * mockRegistry.getVariables() >> ["test.server":"smtp://fakeserver.com","test.username":"user","test.password":"pass"]
-
-
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
index f96aa78..6288de3 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
@@ -191,9 +191,18 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
HdfsResources resources = hdfsResources.get();
if (resources.getConfiguration() == null) {
String configResources = context.getProperty(HADOOP_CONFIGURATION_RESOURCES).getValue();
- String dir = context.getProperty(DIRECTORY_PROP_NAME).getValue();
- dir = dir == null ? "/" : dir;
- resources = resetHDFSResources(configResources, dir, context);
+ final String dir;
+ final PropertyDescriptor directoryPropDescriptor = getPropertyDescriptor(DIRECTORY_PROP_NAME);
+ if (directoryPropDescriptor != null) {
+ if (directoryPropDescriptor.isExpressionLanguageSupported()) {
+ dir = context.getProperty(DIRECTORY_PROP_NAME).evaluateAttributeExpressions().getValue();
+ } else {
+ dir = context.getProperty(DIRECTORY_PROP_NAME).getValue();
+ }
+ } else {
+ dir = null;
+ }
+ resources = resetHDFSResources(configResources, dir == null ? "/" : dir, context);
hdfsResources.set(resources);
}
} catch (IOException ex) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
index d18c13f..24c0330 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
@@ -97,6 +97,7 @@ public class GetHDFS extends AbstractHadoopProcessor {
.description("The HDFS directory from which files should be read")
.required(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor RECURSE_SUBDIRS = new PropertyDescriptor.Builder()
@@ -241,7 +242,7 @@ public class GetHDFS extends AbstractHadoopProcessor {
// copy configuration values to pass them around cleanly
processorConfig = new ProcessorConfiguration(context);
final FileSystem fs = getFileSystem();
- final Path dir = new Path(context.getProperty(DIRECTORY).getValue());
+ final Path dir = new Path(context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue());
if (!fs.exists(dir)) {
throw new IOException("PropertyDescriptor " + DIRECTORY + " has invalid value " + dir + ". The directory does not exist.");
}
@@ -341,7 +342,7 @@ public class GetHDFS extends AbstractHadoopProcessor {
final Double bufferSizeProp = context.getProperty(BUFFER_SIZE).asDataSize(DataUnit.B);
int bufferSize = bufferSizeProp != null ? bufferSizeProp.intValue() : conf.getInt(BUFFER_SIZE_KEY,
BUFFER_SIZE_DEFAULT);
- final Path rootDir = new Path(context.getProperty(DIRECTORY).getValue());
+ final Path rootDir = new Path(context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue());
final CompressionType compressionType = CompressionType.valueOf(context.getProperty(COMPRESSION_CODEC).toString());
final boolean inferCompressionCodec = compressionType == CompressionType.AUTOMATIC;
@@ -501,7 +502,7 @@ public class GetHDFS extends AbstractHadoopProcessor {
final private PathFilter pathFilter;
ProcessorConfiguration(final ProcessContext context) {
- configuredRootDirPath = new Path(context.getProperty(DIRECTORY).getValue());
+ configuredRootDirPath = new Path(context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue());
ignoreDottedFiles = context.getProperty(IGNORE_DOTTED_FILES).asBoolean();
final String fileFilterRegex = context.getProperty(FILE_FILTER_REGEX).getValue();
fileFilterPattern = (fileFilterRegex == null) ? null : Pattern.compile(fileFilterRegex);
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
index 4cb8d25..524bf64 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
@@ -102,6 +102,7 @@ public class ListHDFS extends AbstractHadoopProcessor {
.description("The HDFS directory from which files should be read")
.required(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor RECURSE_SUBDIRS = new PropertyDescriptor.Builder()
@@ -248,7 +249,7 @@ public class ListHDFS extends AbstractHadoopProcessor {
}
lastRunTimestamp = now;
- final String directory = context.getProperty(DIRECTORY).getValue();
+ final String directory = context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue();
// Ensure that we are using the latest listing information before we try to perform a listing of HDFS files.
try {
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java
index 48e0f89..34efcb2 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java
@@ -58,7 +58,7 @@ public class PutHDFSTest {
private KerberosProperties kerberosProperties;
@BeforeClass
- public static void setUpClass() throws Exception{
+ public static void setUpClass() throws Exception {
/*
* Running Hadoop on Windows requires a special build which will produce required binaries and native modules [1]. Since functionality
* provided by this module and validated by these test does not have any native implication we do not distribute required binaries and native modules
@@ -199,7 +199,6 @@ public class PutHDFSTest {
TestRunner runner = TestRunners.newTestRunner(proc);
runner.setProperty(PutHDFS.DIRECTORY, "target/test-classes");
runner.setProperty(PutHDFS.CONFLICT_RESOLUTION, "replace");
- runner.setValidateExpressionUsage(false);
try (FileInputStream fis = new FileInputStream("src/test/resources/testdata/randombytes-1");) {
Map<String, String> attributes = new HashMap<String, String>();
attributes.put(CoreAttributes.FILENAME.key(), "randombytes-1");
@@ -232,7 +231,6 @@ public class PutHDFSTest {
runner.setProperty(PutHDFS.DIRECTORY, "target/test-classes");
runner.setProperty(PutHDFS.CONFLICT_RESOLUTION, "replace");
runner.setProperty(PutHDFS.COMPRESSION_CODEC, "GZIP");
- runner.setValidateExpressionUsage(false);
try (FileInputStream fis = new FileInputStream("src/test/resources/testdata/randombytes-1");) {
Map<String, String> attributes = new HashMap<String, String>();
attributes.put(CoreAttributes.FILENAME.key(), "randombytes-1");
@@ -281,7 +279,6 @@ public class PutHDFSTest {
});
runner.setProperty(PutHDFS.DIRECTORY, dirName);
runner.setProperty(PutHDFS.CONFLICT_RESOLUTION, "replace");
- runner.setValidateExpressionUsage(false);
try (FileInputStream fis = new FileInputStream("src/test/resources/testdata/randombytes-1");) {
Map<String, String> attributes = new HashMap<String, String>();
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProvider.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProvider.java
index e279bae..c4608be 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProvider.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProvider.java
@@ -117,11 +117,11 @@ public class JMSConnectionFactoryProvider extends AbstractControllerService impl
if (!this.configured) {
if (logger.isInfoEnabled()) {
logger.info("Configuring " + this.getClass().getSimpleName() + " for '"
- + context.getProperty(CONNECTION_FACTORY_IMPL).getValue() + "' to be conected to '"
+ + context.getProperty(CONNECTION_FACTORY_IMPL).evaluateAttributeExpressions().getValue() + "' to be conected to '"
+ BROKER_URI + "'");
}
// will load user provided libraries/resources on the classpath
- Utils.addResourcesToClasspath(context.getProperty(CLIENT_LIB_DIR_PATH).getValue());
+ Utils.addResourcesToClasspath(context.getProperty(CLIENT_LIB_DIR_PATH).evaluateAttributeExpressions().getValue());
this.createConnectionFactoryInstance(context);
@@ -177,7 +177,7 @@ public class JMSConnectionFactoryProvider extends AbstractControllerService impl
this.setProperty(propertyName, entry.getValue());
} else {
if (propertyName.equals(BROKER)) {
- if (context.getProperty(CONNECTION_FACTORY_IMPL).getValue().startsWith("org.apache.activemq")) {
+ if (context.getProperty(CONNECTION_FACTORY_IMPL).evaluateAttributeExpressions().getValue().startsWith("org.apache.activemq")) {
this.setProperty("brokerURL", entry.getValue());
} else {
String[] hostPort = entry.getValue().split(":");
@@ -248,7 +248,7 @@ public class JMSConnectionFactoryProvider extends AbstractControllerService impl
* 'CONNECTION_FACTORY_IMPL'.
*/
private void createConnectionFactoryInstance(ConfigurationContext context) {
- String connectionFactoryImplName = context.getProperty(CONNECTION_FACTORY_IMPL).getValue();
+ String connectionFactoryImplName = context.getProperty(CONNECTION_FACTORY_IMPL).evaluateAttributeExpressions().getValue();
this.connectionFactory = Utils.newDefaultInstance(connectionFactoryImplName);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderDefinition.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderDefinition.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderDefinition.java
index 4375e22..3057ec6 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderDefinition.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/main/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderDefinition.java
@@ -44,6 +44,7 @@ public interface JMSConnectionFactoryProviderDefinition extends ControllerServic
+ "class (i.e., org.apache.activemq.ActiveMQConnectionFactory)")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.required(true)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor CLIENT_LIB_DIR_PATH = new PropertyDescriptor.Builder()
.name(CF_LIB)
@@ -53,6 +54,7 @@ public interface JMSConnectionFactoryProviderDefinition extends ControllerServic
+ "ConnectionFactory implementation.")
.addValidator(new ClientLibValidator())
.required(true)
+ .expressionLanguageSupported(true)
.build();
// ConnectionFactory specific properties
@@ -63,6 +65,7 @@ public interface JMSConnectionFactoryProviderDefinition extends ControllerServic
+ "'tcp://myhost:61616' for ActiveMQ or 'myhost:1414' for IBM MQ")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.required(true)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/test/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/test/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderTest.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/test/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderTest.java
index 95bdbe5..ab74ddd 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/test/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderTest.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-cf-service/src/test/java/org/apache/nifi/jms/cf/JMSConnectionFactoryProviderTest.java
@@ -16,14 +16,13 @@
*/
package org.apache.nifi.jms.cf;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.lang.reflect.Method;
-import java.util.Iterator;
-import java.util.ServiceLoader;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -32,15 +31,15 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
+import java.io.File;
+import java.lang.reflect.Method;
+import java.util.Iterator;
+import java.util.ServiceLoader;
-import org.apache.nifi.controller.ControllerService;
-import org.apache.nifi.processor.Processor;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.Test;
-import org.mockito.Mockito;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
/**
*
@@ -51,7 +50,7 @@ public class JMSConnectionFactoryProviderTest {
@Test
public void validateFullConfigWithUserLib() throws Exception {
- TestRunner runner = TestRunners.newTestRunner(Mockito.mock(Processor.class));
+ TestRunner runner = TestRunners.newTestRunner(mock(Processor.class));
JMSConnectionFactoryProvider cfProvider = new JMSConnectionFactoryProvider();
runner.addControllerService("cfProvider", cfProvider);
runner.setProperty(cfProvider, JMSConnectionFactoryProvider.BROKER_URI, "myhost:1234");
@@ -76,7 +75,7 @@ public class JMSConnectionFactoryProviderTest {
@Test(expected = AssertionError.class)
public void validateOnConfigureFailsIfCNFonConnectionFactory() throws Exception {
- TestRunner runner = TestRunners.newTestRunner(Mockito.mock(Processor.class));
+ TestRunner runner = TestRunners.newTestRunner(mock(Processor.class));
JMSConnectionFactoryProvider cfProvider = new JMSConnectionFactoryProvider();
runner.addControllerService("cfProvider", cfProvider);
runner.setProperty(cfProvider, JMSConnectionFactoryProvider.BROKER_URI, "myhost:1234");
@@ -89,7 +88,7 @@ public class JMSConnectionFactoryProviderTest {
@Test
public void validateNotValidForNonExistingLibPath() throws Exception {
- TestRunner runner = TestRunners.newTestRunner(Mockito.mock(Processor.class));
+ TestRunner runner = TestRunners.newTestRunner(mock(Processor.class));
JMSConnectionFactoryProvider cfProvider = new JMSConnectionFactoryProvider();
runner.addControllerService("cfProvider", cfProvider);
runner.setProperty(cfProvider, JMSConnectionFactoryProvider.BROKER_URI, "myhost:1234");
@@ -102,7 +101,7 @@ public class JMSConnectionFactoryProviderTest {
@Test(expected = AssertionError.class)
public void validateFailsIfURINotHostPortAndNotActiveMQ() throws Exception {
- TestRunner runner = TestRunners.newTestRunner(Mockito.mock(Processor.class));
+ TestRunner runner = TestRunners.newTestRunner(mock(Processor.class));
JMSConnectionFactoryProvider cfProvider = new JMSConnectionFactoryProvider();
runner.addControllerService("cfProvider", cfProvider);
runner.setProperty(cfProvider, JMSConnectionFactoryProvider.BROKER_URI, "myhost");
@@ -116,7 +115,7 @@ public class JMSConnectionFactoryProviderTest {
@Test
public void validateNotValidForNonDirectoryPath() throws Exception {
- TestRunner runner = TestRunners.newTestRunner(Mockito.mock(Processor.class));
+ TestRunner runner = TestRunners.newTestRunner(mock(Processor.class));
JMSConnectionFactoryProvider cfProvider = new JMSConnectionFactoryProvider();
runner.addControllerService("cfProvider", cfProvider);
runner.setProperty(cfProvider, JMSConnectionFactoryProvider.BROKER_URI, "myhost:1234");
@@ -146,7 +145,7 @@ public class JMSConnectionFactoryProviderTest {
try {
String libPath = TestUtils.setupActiveMqLibForTesting(true);
- TestRunner runner = TestRunners.newTestRunner(Mockito.mock(Processor.class));
+ TestRunner runner = TestRunners.newTestRunner(mock(Processor.class));
JMSConnectionFactoryProvider cfProvider = new JMSConnectionFactoryProvider();
runner.addControllerService("cfProvider", cfProvider);
runner.setProperty(cfProvider, JMSConnectionFactoryProvider.BROKER_URI,
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java
index f8030db..398c5c1 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/AbstractJMSProcessor.java
@@ -69,6 +69,7 @@ abstract class AbstractJMSProcessor<T extends JMSWorker> extends AbstractProcess
.description("The name of the JMS Destination. Usually provided by the administrator (e.g., 'topic://myTopic').")
.required(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
static final PropertyDescriptor DESTINATION_TYPE = new PropertyDescriptor.Builder()
.name("Destination Type")
@@ -198,7 +199,7 @@ abstract class AbstractJMSProcessor<T extends JMSWorker> extends AbstractProcess
JmsTemplate jmsTemplate = new JmsTemplate();
jmsTemplate.setConnectionFactory(this.cachingConnectionFactory);
- jmsTemplate.setDefaultDestinationName(context.getProperty(DESTINATION).getValue());
+ jmsTemplate.setDefaultDestinationName(context.getProperty(DESTINATION).evaluateAttributeExpressions().getValue());
jmsTemplate.setPubSubDomain(TOPIC.equals(context.getProperty(DESTINATION_TYPE).getValue()));
// set of properties that may be good candidates for exposure via configuration
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
index a4cad0d..ac05f2c 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/ConsumeJMS.java
@@ -87,7 +87,7 @@ public class ConsumeJMS extends AbstractJMSProcessor<JMSConsumer> {
});
Map<String, Object> jmsHeaders = response.getMessageHeaders();
flowFile = this.updateFlowFileAttributesWithJmsHeaders(jmsHeaders, flowFile, processSession);
- processSession.getProvenanceReporter().receive(flowFile, context.getProperty(DESTINATION).getValue());
+ processSession.getProvenanceReporter().receive(flowFile, context.getProperty(DESTINATION).evaluateAttributeExpressions().getValue());
processSession.transfer(flowFile, REL_SUCCESS);
} else {
context.yield();
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/PublishJMS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/PublishJMS.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/PublishJMS.java
index 8f6c18b..0802b43 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/PublishJMS.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/PublishJMS.java
@@ -101,7 +101,7 @@ public class PublishJMS extends AbstractJMSProcessor<JMSPublisher> {
this.targetResource.publish(this.extractMessageBody(flowFile, processSession),
flowFile.getAttributes());
processSession.transfer(flowFile, REL_SUCCESS);
- processSession.getProvenanceReporter().send(flowFile, context.getProperty(DESTINATION).getValue());
+ processSession.getProvenanceReporter().send(flowFile, context.getProperty(DESTINATION).evaluateAttributeExpressions().getValue());
} catch (Exception e) {
processSession.transfer(flowFile, REL_FAILURE);
this.getLogger().error("Failed while sending message to JMS via " + this.targetResource, e);
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/ConsumeJMSTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/ConsumeJMSTest.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/ConsumeJMSTest.java
index 68b95fd..9366e8d 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/ConsumeJMSTest.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/ConsumeJMSTest.java
@@ -16,11 +16,6 @@
*/
package org.apache.nifi.jms.processors;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
import org.apache.nifi.jms.cf.JMSConnectionFactoryProviderDefinition;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.util.MockFlowFile;
@@ -31,6 +26,11 @@ import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.JmsHeaders;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
public class ConsumeJMSTest {
@Test
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/PublishJMSTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/PublishJMSTest.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/PublishJMSTest.java
index 28d4ac6..36edf79 100644
--- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/PublishJMSTest.java
+++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/PublishJMSTest.java
@@ -16,19 +16,6 @@
*/
package org.apache.nifi.jms.processors;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import javax.jms.BytesMessage;
-import javax.jms.ConnectionFactory;
-import javax.jms.Queue;
-
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.nifi.jms.cf.JMSConnectionFactoryProviderDefinition;
import org.apache.nifi.util.MockFlowFile;
@@ -38,6 +25,18 @@ import org.junit.Test;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.JmsHeaders;
+import javax.jms.BytesMessage;
+import javax.jms.ConnectionFactory;
+import javax.jms.Queue;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
public class PublishJMSTest {
@Test
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/main/java/org/apache/nifi/dbcp/DBCPConnectionPool.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/main/java/org/apache/nifi/dbcp/DBCPConnectionPool.java b/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/main/java/org/apache/nifi/dbcp/DBCPConnectionPool.java
index 08aa966..8fc7530 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/main/java/org/apache/nifi/dbcp/DBCPConnectionPool.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/main/java/org/apache/nifi/dbcp/DBCPConnectionPool.java
@@ -16,18 +16,6 @@
*/
package org.apache.nifi.dbcp;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.sql.Connection;
-import java.sql.Driver;
-import java.sql.DriverManager;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
import org.apache.commons.dbcp.BasicDataSource;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
@@ -40,6 +28,18 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.reporting.InitializationException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
/**
* Implementation of for Database Connection Pooling Service. Apache DBCP is used for connection pooling functionality.
*
@@ -55,6 +55,7 @@ public class DBCPConnectionPool extends AbstractControllerService implements DBC
.defaultValue(null)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.required(true)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor DB_DRIVERNAME = new PropertyDescriptor.Builder()
@@ -63,6 +64,7 @@ public class DBCPConnectionPool extends AbstractControllerService implements DBC
.defaultValue(null)
.required(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor DB_DRIVER_JAR_URL = new PropertyDescriptor.Builder()
@@ -71,6 +73,7 @@ public class DBCPConnectionPool extends AbstractControllerService implements DBC
.defaultValue(null)
.required(false)
.addValidator(StandardValidators.URL_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor DB_USER = new PropertyDescriptor.Builder()
@@ -78,6 +81,7 @@ public class DBCPConnectionPool extends AbstractControllerService implements DBC
.description("Database user name")
.defaultValue(null)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor DB_PASSWORD = new PropertyDescriptor.Builder()
@@ -87,6 +91,7 @@ public class DBCPConnectionPool extends AbstractControllerService implements DBC
.required(false)
.sensitive(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public static final PropertyDescriptor MAX_WAIT_TIME = new PropertyDescriptor.Builder()
@@ -148,9 +153,9 @@ public class DBCPConnectionPool extends AbstractControllerService implements DBC
@OnEnabled
public void onConfigured(final ConfigurationContext context) throws InitializationException {
- final String drv = context.getProperty(DB_DRIVERNAME).getValue();
- final String user = context.getProperty(DB_USER).getValue();
- final String passw = context.getProperty(DB_PASSWORD).getValue();
+ final String drv = context.getProperty(DB_DRIVERNAME).evaluateAttributeExpressions().getValue();
+ final String user = context.getProperty(DB_USER).evaluateAttributeExpressions().getValue();
+ final String passw = context.getProperty(DB_PASSWORD).evaluateAttributeExpressions().getValue();
final Long maxWaitMillis = context.getProperty(MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS);
final Integer maxTotal = context.getProperty(MAX_TOTAL_CONNECTIONS).asInteger();
@@ -158,10 +163,10 @@ public class DBCPConnectionPool extends AbstractControllerService implements DBC
dataSource.setDriverClassName(drv);
// Optional driver URL, when exist, this URL will be used to locate driver jar file location
- final String urlString = context.getProperty(DB_DRIVER_JAR_URL).getValue();
+ final String urlString = context.getProperty(DB_DRIVER_JAR_URL).evaluateAttributeExpressions().getValue();
dataSource.setDriverClassLoader(getDriverClassLoader(urlString, drv));
- final String dburl = context.getProperty(DATABASE_URL).getValue();
+ final String dburl = context.getProperty(DATABASE_URL).evaluateAttributeExpressions().getValue();
dataSource.setMaxWait(maxWaitMillis);
dataSource.setMaxActive(maxTotal);
http://git-wip-us.apache.org/repos/asf/nifi/blob/1ea49ad8/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/test/java/org/apache/nifi/dbcp/DBCPServiceTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/test/java/org/apache/nifi/dbcp/DBCPServiceTest.java b/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/test/java/org/apache/nifi/dbcp/DBCPServiceTest.java
index 6683d2a..fcf58ea 100644
--- a/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/test/java/org/apache/nifi/dbcp/DBCPServiceTest.java
+++ b/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/test/java/org/apache/nifi/dbcp/DBCPServiceTest.java
@@ -16,8 +16,16 @@
*/
package org.apache.nifi.dbcp;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
import java.io.File;
import java.net.MalformedURLException;
@@ -32,16 +40,8 @@ import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.reporting.InitializationException;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
public class DBCPServiceTest {