nifi-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mattyb...@apache.org
Subject nifi git commit: NIFI-4082 - Added EL on GetMongo properties
Date Mon, 07 Aug 2017 17:54:26 GMT
Repository: nifi
Updated Branches:
  refs/heads/master 329b1caf8 -> 8b54c2604


NIFI-4082 - Added EL on GetMongo properties

NIFI-4082 - Added EL on DB, URI and Collection

NIFI-4082 - Added UT for EL evaluation (URI, DB, Collection) and fixed ex. message for document
validator.

This closes #1969


Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/8b54c260
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/8b54c260
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/8b54c260

Branch: refs/heads/master
Commit: 8b54c2604cde4f3516d89a0f48650f7c30e34c3b
Parents: 329b1ca
Author: Pierre Villard <pierre.villard.fr@gmail.com>
Authored: Fri Jun 30 20:43:50 2017 +0200
Committer: Matt Burgess <mattyb149@apache.org>
Committed: Mon Aug 7 13:40:27 2017 -0400

----------------------------------------------------------------------
 .../mongodb/AbstractMongoProcessor.java         | 27 +++++++++++++++-----
 .../nifi/processors/mongodb/GetMongo.java       | 27 +++++++++++++++-----
 .../nifi/processors/mongodb/PutMongo.java       |  4 +--
 .../nifi/processors/mongodb/GetMongoTest.java   | 18 ++++++++-----
 .../nifi/processors/mongodb/PutMongoTest.java   |  9 ++++---
 5 files changed, 61 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/nifi/blob/8b54c260/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoProcessor.java
b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoProcessor.java
index 486a077..6f165c2 100644
--- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoProcessor.java
+++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoProcessor.java
@@ -29,6 +29,7 @@ import org.apache.nifi.annotation.lifecycle.OnScheduled;
 import org.apache.nifi.annotation.lifecycle.OnStopped;
 import org.apache.nifi.authentication.exception.ProviderCreationException;
 import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.processor.AbstractProcessor;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.util.StandardValidators;
@@ -48,18 +49,21 @@ public abstract class AbstractMongoProcessor extends AbstractProcessor
{
         .name("Mongo URI")
         .description("MongoURI, typically of the form: mongodb://host1[:port1][,host2[:port2],...]")
         .required(true)
+        .expressionLanguageSupported(true)
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .build();
     protected static final PropertyDescriptor DATABASE_NAME = new PropertyDescriptor.Builder()
         .name("Mongo Database Name")
         .description("The name of the database to use")
         .required(true)
+        .expressionLanguageSupported(true)
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .build();
     protected static final PropertyDescriptor COLLECTION_NAME = new PropertyDescriptor.Builder()
         .name("Mongo Collection Name")
         .description("The name of the collection to use")
         .required(true)
+        .expressionLanguageSupported(true)
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .build();
     public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()
@@ -124,11 +128,10 @@ public abstract class AbstractMongoProcessor extends AbstractProcessor
{
         }
 
         try {
-            final String uri = context.getProperty(URI).getValue();
             if(sslContext == null) {
-                mongoClient = new MongoClient(new MongoClientURI(uri));
+                mongoClient = new MongoClient(new MongoClientURI(getURI(context)));
             } else {
-                mongoClient = new MongoClient(new MongoClientURI(uri, getClientOptions(sslContext)));
+                mongoClient = new MongoClient(new MongoClientURI(getURI(context), getClientOptions(sslContext)));
             }
         } catch (Exception e) {
             getLogger().error("Failed to schedule {} due to {}", new Object[] { this.getClass().getName(),
e }, e);
@@ -153,12 +156,24 @@ public abstract class AbstractMongoProcessor extends AbstractProcessor
{
     }
 
     protected MongoDatabase getDatabase(final ProcessContext context) {
-        final String databaseName = context.getProperty(DATABASE_NAME).getValue();
+        return getDatabase(context, null);
+    }
+
+    protected MongoDatabase getDatabase(final ProcessContext context, final FlowFile flowFile)
{
+        final String databaseName = context.getProperty(DATABASE_NAME).evaluateAttributeExpressions(flowFile).getValue();
         return mongoClient.getDatabase(databaseName);
     }
 
     protected MongoCollection<Document> getCollection(final ProcessContext context)
{
-        final String collectionName = context.getProperty(COLLECTION_NAME).getValue();
-        return getDatabase(context).getCollection(collectionName);
+        return getCollection(context, null);
+    }
+
+    protected MongoCollection<Document> getCollection(final ProcessContext context,
final FlowFile flowFile) {
+        final String collectionName = context.getProperty(COLLECTION_NAME).evaluateAttributeExpressions(flowFile).getValue();
+        return getDatabase(context, flowFile).getCollection(collectionName);
+    }
+
+    protected String getURI(final ProcessContext context) {
+        return context.getProperty(URI).evaluateAttributeExpressions().getValue();
     }
 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/8b54c260/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java
b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java
index 8b73b02..79f50b2 100644
--- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java
+++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java
@@ -59,14 +59,21 @@ public class GetMongo extends AbstractMongoProcessor {
     public static final Validator DOCUMENT_VALIDATOR = new Validator() {
         @Override
         public ValidationResult validate(final String subject, final String value, final
ValidationContext context) {
+            final ValidationResult.Builder builder = new ValidationResult.Builder();
+            builder.subject(subject).input(value);
+
+            if (context.isExpressionLanguageSupported(subject) && context.isExpressionLanguagePresent(value))
{
+                return builder.valid(true).explanation("Contains Expression Language").build();
+            }
+
             String reason = null;
             try {
                 Document.parse(value);
             } catch (final RuntimeException e) {
-                reason = e.getClass().getName();
+                reason = e.getLocalizedMessage();
             }
 
-            return new ValidationResult.Builder().subject(subject).input(value).explanation(reason).valid(reason
== null).build();
+            return builder.explanation(reason).valid(reason == null).build();
         }
     };
 
@@ -76,18 +83,21 @@ public class GetMongo extends AbstractMongoProcessor {
         .name("Query")
         .description("The selection criteria; must be a valid MongoDB Extended JSON format;
if omitted the entire collection will be queried")
         .required(false)
+        .expressionLanguageSupported(true)
         .addValidator(DOCUMENT_VALIDATOR)
         .build();
     static final PropertyDescriptor PROJECTION = new PropertyDescriptor.Builder()
         .name("Projection")
         .description("The fields to be returned from the documents in the result set; must
be a valid BSON document")
         .required(false)
+        .expressionLanguageSupported(true)
         .addValidator(DOCUMENT_VALIDATOR)
         .build();
     static final PropertyDescriptor SORT = new PropertyDescriptor.Builder()
         .name("Sort")
         .description("The fields by which to sort; must be a valid BSON document")
         .required(false)
+        .expressionLanguageSupported(true)
         .addValidator(DOCUMENT_VALIDATOR)
         .build();
     static final PropertyDescriptor LIMIT = new PropertyDescriptor.Builder()
@@ -163,7 +173,7 @@ public class GetMongo extends AbstractMongoProcessor {
             }
         });
         flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "application/json");
-        session.getProvenanceReporter().receive(flowFile, context.getProperty(URI).getValue());
+        session.getProvenanceReporter().receive(flowFile, getURI(context));
         session.transfer(flowFile, REL_SUCCESS);
     }
 
@@ -171,9 +181,12 @@ public class GetMongo extends AbstractMongoProcessor {
     public void onTrigger(final ProcessContext context, final ProcessSession session) throws
ProcessException {
         final ComponentLog logger = getLogger();
 
-        final Document query = context.getProperty(QUERY).isSet() ? Document.parse(context.getProperty(QUERY).getValue())
: null;
-        final Document projection = context.getProperty(PROJECTION).isSet() ? Document.parse(context.getProperty(PROJECTION).getValue())
: null;
-        final Document sort = context.getProperty(SORT).isSet() ? Document.parse(context.getProperty(SORT).getValue())
: null;
+        final Document query = context.getProperty(QUERY).isSet()
+                ? Document.parse(context.getProperty(QUERY).evaluateAttributeExpressions().getValue())
: null;
+        final Document projection = context.getProperty(PROJECTION).isSet()
+                ? Document.parse(context.getProperty(PROJECTION).evaluateAttributeExpressions().getValue())
: null;
+        final Document sort = context.getProperty(SORT).isSet()
+                ? Document.parse(context.getProperty(SORT).evaluateAttributeExpressions().getValue())
: null;
 
         final MongoCollection<Document> collection = getCollection(context);
 
@@ -233,7 +246,7 @@ public class GetMongo extends AbstractMongoProcessor {
                         });
                         flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(),
"application/json");
 
-                        session.getProvenanceReporter().receive(flowFile, context.getProperty(URI).getValue());
+                        session.getProvenanceReporter().receive(flowFile, getURI(context));
                         session.transfer(flowFile, REL_SUCCESS);
                     }
                 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/8b54c260/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/PutMongo.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/PutMongo.java
b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/PutMongo.java
index 51e5265..5b5ad52 100644
--- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/PutMongo.java
+++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/PutMongo.java
@@ -148,7 +148,7 @@ public class PutMongo extends AbstractMongoProcessor {
         final String mode = context.getProperty(MODE).getValue();
         final WriteConcern writeConcern = getWriteConcern(context);
 
-        final MongoCollection<Document> collection = getCollection(context).withWriteConcern(writeConcern);
+        final MongoCollection<Document> collection = getCollection(context, flowFile).withWriteConcern(writeConcern);
 
         try {
             // Read the contents of the FlowFile into a byte array
@@ -176,7 +176,7 @@ public class PutMongo extends AbstractMongoProcessor {
                 logger.info("updated {} into MongoDB", new Object[] { flowFile });
             }
 
-            session.getProvenanceReporter().send(flowFile, context.getProperty(URI).getValue());
+            session.getProvenanceReporter().send(flowFile, getURI(context));
             session.transfer(flowFile, REL_SUCCESS);
         } catch (Exception e) {
             logger.error("Failed to insert {} into MongoDB due to {}", new Object[] {flowFile,
e}, e);

http://git-wip-us.apache.org/repos/asf/nifi/blob/8b54c260/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoTest.java
b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoTest.java
index e39148d..455d705 100644
--- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoTest.java
+++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoTest.java
@@ -60,9 +60,12 @@ public class GetMongoTest {
     @Before
     public void setup() {
         runner = TestRunners.newTestRunner(GetMongo.class);
-        runner.setProperty(AbstractMongoProcessor.URI, MONGO_URI);
-        runner.setProperty(AbstractMongoProcessor.DATABASE_NAME, DB_NAME);
-        runner.setProperty(AbstractMongoProcessor.COLLECTION_NAME, COLLECTION_NAME);
+        runner.setVariable("uri", MONGO_URI);
+        runner.setVariable("db", DB_NAME);
+        runner.setVariable("collection", COLLECTION_NAME);
+        runner.setProperty(AbstractMongoProcessor.URI, "${uri}");
+        runner.setProperty(AbstractMongoProcessor.DATABASE_NAME, "${db}");
+        runner.setProperty(AbstractMongoProcessor.COLLECTION_NAME, "${collection}");
 
         mongoClient = new MongoClient(new MongoClientURI(MONGO_URI));
 
@@ -120,8 +123,9 @@ public class GetMongoTest {
         Assert.assertTrue(results.iterator().next().toString().matches("'Query' .* is invalid
because org.bson.json.JsonParseException"));
 
         // invalid projection
+        runner.setVariable("projection", "{a: x,y,z}");
         runner.setProperty(GetMongo.QUERY, "{a: 1}");
-        runner.setProperty(GetMongo.PROJECTION, "{a: x,y,z}");
+        runner.setProperty(GetMongo.PROJECTION, "${projection}");
         runner.enqueue(new byte[0]);
         pc = runner.getProcessContext();
         results = new HashSet<>();
@@ -146,7 +150,8 @@ public class GetMongoTest {
 
     @Test
     public void testReadOneDocument() throws Exception {
-        runner.setProperty(GetMongo.QUERY, "{a: 1, b: 3}");
+        runner.setVariable("query", "{a: 1, b: 3}");
+        runner.setProperty(GetMongo.QUERY, "${query}");
         runner.run();
 
         runner.assertAllFlowFilesTransferred(GetMongo.REL_SUCCESS, 1);
@@ -180,8 +185,9 @@ public class GetMongoTest {
 
     @Test
     public void testSort() throws Exception {
+        runner.setVariable("sort", "{a: -1, b: -1, c: 1}");
         runner.setProperty(GetMongo.QUERY, "{a: {$exists: true}}");
-        runner.setProperty(GetMongo.SORT, "{a: -1, b: -1, c: 1}");
+        runner.setProperty(GetMongo.SORT, "${sort}");
         runner.run();
 
         runner.assertAllFlowFilesTransferred(GetMongo.REL_SUCCESS, 3);

http://git-wip-us.apache.org/repos/asf/nifi/blob/8b54c260/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/PutMongoTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/PutMongoTest.java
b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/PutMongoTest.java
index 6f22976..10f81d1 100644
--- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/PutMongoTest.java
+++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/PutMongoTest.java
@@ -61,9 +61,12 @@ public class PutMongoTest {
     @Before
     public void setup() {
         runner = TestRunners.newTestRunner(PutMongo.class);
-        runner.setProperty(AbstractMongoProcessor.URI, MONGO_URI);
-        runner.setProperty(AbstractMongoProcessor.DATABASE_NAME, DATABASE_NAME);
-        runner.setProperty(AbstractMongoProcessor.COLLECTION_NAME, COLLECTION_NAME);
+        runner.setVariable("uri", MONGO_URI);
+        runner.setVariable("db", DATABASE_NAME);
+        runner.setVariable("collection", COLLECTION_NAME);
+        runner.setProperty(AbstractMongoProcessor.URI, "${uri}");
+        runner.setProperty(AbstractMongoProcessor.DATABASE_NAME, "${db}");
+        runner.setProperty(AbstractMongoProcessor.COLLECTION_NAME, "${collection}");
 
         mongoClient = new MongoClient(new MongoClientURI(MONGO_URI));
 


Mime
View raw message