You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mt...@apache.org on 2018/09/14 18:05:58 UTC
nifi git commit: NIFI-5449: Added Base64 Encode/Decode functions to
RecordPath NIFI-5449: Incorporated review comments
Repository: nifi
Updated Branches:
refs/heads/master 57ae9b65a -> 5a84d650c
NIFI-5449: Added Base64 Encode/Decode functions to RecordPath
NIFI-5449: Incorporated review comments
This closes #2920
Signed-off-by: Mike Thomsen <mi...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/5a84d650
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/5a84d650
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/5a84d650
Branch: refs/heads/master
Commit: 5a84d650c3a35b6d48d662c47e10fd2fc733f89b
Parents: 57ae9b6
Author: Matthew Burgess <ma...@apache.org>
Authored: Thu Jul 26 17:24:14 2018 -0400
Committer: Mike Thomsen <mi...@gmail.com>
Committed: Fri Sep 14 14:04:25 2018 -0400
----------------------------------------------------------------------
.../record/path/functions/Base64Decode.java | 54 +++++++++++++++
.../record/path/functions/Base64Encode.java | 58 ++++++++++++++++
.../record/path/paths/RecordPathCompiler.java | 10 +++
.../apache/nifi/record/path/TestRecordPath.java | 70 ++++++++++++++++++++
.../src/main/asciidoc/record-path-guide.adoc | 59 +++++++++++++++++
5 files changed, 251 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/5a84d650/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Decode.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Decode.java b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Decode.java
new file mode 100644
index 0000000..f5a1e5c
--- /dev/null
+++ b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Decode.java
@@ -0,0 +1,54 @@
+/*
+ * 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.record.path.functions;
+
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPathEvaluationContext;
+import org.apache.nifi.record.path.StandardFieldValue;
+import org.apache.nifi.record.path.paths.RecordPathSegment;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.stream.Stream;
+
+public class Base64Decode extends RecordPathSegment {
+ private final RecordPathSegment recordPath;
+
+ public Base64Decode(final RecordPathSegment recordPath, final boolean absolute) {
+ super("base64Decode", null, absolute);
+ this.recordPath = recordPath;
+ }
+
+ @Override
+ public Stream<FieldValue> evaluate(final RecordPathEvaluationContext context) {
+ final Stream<FieldValue> fieldValues = recordPath.evaluate(context);
+ return fieldValues.filter(fv -> fv.getValue() != null)
+ .map(fv -> {
+
+ Object value = fv.getValue();
+ if (value instanceof String) {
+ return new StandardFieldValue(new String(Base64.getDecoder().decode(fv.getValue().toString()), StandardCharsets.UTF_8), fv.getField(), fv.getParent().orElse(null));
+ } else if (value instanceof byte[]) {
+ return new StandardFieldValue(Base64.getDecoder().decode((byte[]) value), fv.getField(), fv.getParent().orElse(null));
+ } else {
+ throw new IllegalArgumentException("Argument supplied to base64Decode must be a String or byte[]");
+ }
+ });
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/5a84d650/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Encode.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Encode.java b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Encode.java
new file mode 100644
index 0000000..69a878b
--- /dev/null
+++ b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/functions/Base64Encode.java
@@ -0,0 +1,58 @@
+/*
+ * 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.record.path.functions;
+
+import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.RecordPathEvaluationContext;
+import org.apache.nifi.record.path.StandardFieldValue;
+import org.apache.nifi.record.path.paths.RecordPathSegment;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Base64;
+import java.util.stream.Stream;
+
+public class Base64Encode extends RecordPathSegment {
+ private final RecordPathSegment recordPath;
+
+ public Base64Encode(final RecordPathSegment recordPath, final boolean absolute) {
+ super("base64Encode", null, absolute);
+ this.recordPath = recordPath;
+ }
+
+ @Override
+ public Stream<FieldValue> evaluate(final RecordPathEvaluationContext context) {
+ final Stream<FieldValue> fieldValues = recordPath.evaluate(context);
+ return fieldValues.filter(fv -> fv.getValue() != null)
+ .map(fv -> {
+
+ Object value = fv.getValue();
+ if (value instanceof String) {
+ try {
+ return new StandardFieldValue(Base64.getEncoder().encodeToString(value.toString().getBytes("UTF-8")), fv.getField(), fv.getParent().orElse(null));
+ } catch (final UnsupportedEncodingException e) {
+ return null; // won't happen.
+ }
+ } else if (value instanceof byte[]) {
+ return new StandardFieldValue(Base64.getEncoder().encode((byte[]) value), fv.getField(), fv.getParent().orElse(null));
+ } else {
+ throw new IllegalArgumentException("Argument supplied to base64Encode must be a String or byte[]");
+ }
+ });
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/5a84d650/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/RecordPathCompiler.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/RecordPathCompiler.java b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/RecordPathCompiler.java
index 9a2821a..a30419d 100644
--- a/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/RecordPathCompiler.java
+++ b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/RecordPathCompiler.java
@@ -64,6 +64,8 @@ import org.apache.nifi.record.path.filter.NotEqualsFilter;
import org.apache.nifi.record.path.filter.NotFilter;
import org.apache.nifi.record.path.filter.RecordPathFilter;
import org.apache.nifi.record.path.filter.StartsWith;
+import org.apache.nifi.record.path.functions.Base64Decode;
+import org.apache.nifi.record.path.functions.Base64Encode;
import org.apache.nifi.record.path.functions.Concat;
import org.apache.nifi.record.path.functions.Format;
import org.apache.nifi.record.path.functions.FieldName;
@@ -264,6 +266,14 @@ public class RecordPathCompiler {
final RecordPathSegment[] args = getArgPaths(argumentListTree, 2, functionName, absolute);
return new Format(args[0], args[1], absolute);
}
+ case "base64Encode": {
+ final RecordPathSegment[] args = getArgPaths(argumentListTree, 1, functionName, absolute);
+ return new Base64Encode(args[0], absolute);
+ }
+ case "base64Decode": {
+ final RecordPathSegment[] args = getArgPaths(argumentListTree, 1, functionName, absolute);
+ return new Base64Decode(args[0], absolute);
+ }
default: {
throw new RecordPathException("Invalid function call: The '" + functionName + "' function does not exist or can only "
+ "be used within a predicate, not as a standalone function");
http://git-wip-us.apache.org/repos/asf/nifi/blob/5a84d650/nifi-commons/nifi-record-path/src/test/java/org/apache/nifi/record/path/TestRecordPath.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-record-path/src/test/java/org/apache/nifi/record/path/TestRecordPath.java b/nifi-commons/nifi-record-path/src/test/java/org/apache/nifi/record/path/TestRecordPath.java
index 89b0893..67c14e6 100644
--- a/nifi-commons/nifi-record-path/src/test/java/org/apache/nifi/record/path/TestRecordPath.java
+++ b/nifi-commons/nifi-record-path/src/test/java/org/apache/nifi/record/path/TestRecordPath.java
@@ -27,11 +27,14 @@ import java.sql.Date;
import java.text.DateFormat;
import java.text.ParseException;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import org.apache.nifi.record.path.exception.RecordPathException;
import org.apache.nifi.serialization.SimpleRecordSchema;
@@ -1279,6 +1282,73 @@ public class TestRecordPath {
RecordPath.compile("toBytes(/s, \"NOT A REAL CHARSET\")").evaluate(record).getSelectedFields().findFirst().get().getValue();
}
+ @Test
+ public void testBase64Encode() {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new RecordField("firstName", RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField("lastName", RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField("b", RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType())));
+ final RecordSchema schema = new SimpleRecordSchema(fields);
+
+ final List<Object> expectedValues = Arrays.asList(
+ Base64.getEncoder().encodeToString("John".getBytes(StandardCharsets.UTF_8)),
+ Base64.getEncoder().encodeToString("Doe".getBytes(StandardCharsets.UTF_8)),
+ Base64.getEncoder().encode("xyz".getBytes(StandardCharsets.UTF_8))
+ );
+ final Map<String, Object> values = new HashMap<>();
+ values.put("firstName", "John");
+ values.put("lastName", "Doe");
+ values.put("b", "xyz".getBytes(StandardCharsets.UTF_8));
+ final Record record = new MapRecord(schema, values);
+
+ assertEquals(Base64.getEncoder().encodeToString("John".getBytes(StandardCharsets.UTF_8)),
+ RecordPath.compile("base64Encode(/firstName)").evaluate(record).getSelectedFields().findFirst().get().getValue());
+ assertEquals(Base64.getEncoder().encodeToString("Doe".getBytes(StandardCharsets.UTF_8)),
+ RecordPath.compile("base64Encode(/lastName)").evaluate(record).getSelectedFields().findFirst().get().getValue());
+ assertTrue(Arrays.equals(Base64.getEncoder().encode("xyz".getBytes(StandardCharsets.UTF_8)),
+ (byte[]) RecordPath.compile("base64Encode(/b)").evaluate(record).getSelectedFields().findFirst().get().getValue()));
+ List<Object> actualValues = RecordPath.compile("base64Encode(/*)").evaluate(record).getSelectedFields().map(FieldValue::getValue).collect(Collectors.toList());
+ IntStream.range(0, 3).forEach(i -> {
+ Object expectedObject = expectedValues.get(i);
+ Object actualObject = actualValues.get(i);
+ if (actualObject instanceof String) {
+ assertEquals(expectedObject, actualObject);
+ } else if (actualObject instanceof byte[]) {
+ assertTrue(Arrays.equals((byte[]) expectedObject, (byte[]) actualObject));
+ }
+ });
+ }
+
+ @Test
+ public void testBase64Decode() {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new RecordField("firstName", RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField("lastName", RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField("b", RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType())));
+ final RecordSchema schema = new SimpleRecordSchema(fields);
+
+ final List<Object> expectedValues = Arrays.asList("John", "Doe", "xyz".getBytes(StandardCharsets.UTF_8));
+ final Map<String, Object> values = new HashMap<>();
+ values.put("firstName", Base64.getEncoder().encodeToString("John".getBytes(StandardCharsets.UTF_8)));
+ values.put("lastName", Base64.getEncoder().encodeToString("Doe".getBytes(StandardCharsets.UTF_8)));
+ values.put("b", Base64.getEncoder().encode("xyz".getBytes(StandardCharsets.UTF_8)));
+ final Record record = new MapRecord(schema, values);
+
+ assertEquals("John", RecordPath.compile("base64Decode(/firstName)").evaluate(record).getSelectedFields().findFirst().get().getValue());
+ assertEquals("Doe", RecordPath.compile("base64Decode(/lastName)").evaluate(record).getSelectedFields().findFirst().get().getValue());
+ assertTrue(Arrays.equals("xyz".getBytes(StandardCharsets.UTF_8), (byte[]) RecordPath.compile("base64Decode(/b)").evaluate(record).getSelectedFields().findFirst().get().getValue()));
+ List<Object> actualValues = RecordPath.compile("base64Decode(/*)").evaluate(record).getSelectedFields().map(FieldValue::getValue).collect(Collectors.toList());
+ IntStream.range(0, 3).forEach(i -> {
+ Object expectedObject = expectedValues.get(i);
+ Object actualObject = actualValues.get(i);
+ if (actualObject instanceof String) {
+ assertEquals(expectedObject, actualObject);
+ } else if (actualObject instanceof byte[]) {
+ assertTrue(Arrays.equals((byte[]) expectedObject, (byte[]) actualObject));
+ }
+ });
+ }
+
private List<RecordField> getDefaultFields() {
final List<RecordField> fields = new ArrayList<>();
fields.add(new RecordField("id", RecordFieldType.INT.getDataType()));
http://git-wip-us.apache.org/repos/asf/nifi/blob/5a84d650/nifi-docs/src/main/asciidoc/record-path-guide.adoc
----------------------------------------------------------------------
diff --git a/nifi-docs/src/main/asciidoc/record-path-guide.adoc b/nifi-docs/src/main/asciidoc/record-path-guide.adoc
index 8de98ee..14a2299 100644
--- a/nifi-docs/src/main/asciidoc/record-path-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/record-path-guide.adoc
@@ -627,6 +627,65 @@ The following record path expression would re-format the date String:
| `format( toDate(/eventDate, "yyyy-MM-dd'T'HH:mm:ss'Z'"), 'yyyy-MM-dd')` | 2017-10-20
|==========================================================
+=== base64Encode
+
+Converts a String or byte[] using Base64 encoding, using the UTF-8 character set. For example, given a schema such as:
+
+----
+{
+ "type": "record",
+ "name": "events",
+ "fields": [
+ { "name": "name", "type": "string" }
+ ]
+}
+----
+
+and a record such as:
+
+----
+{
+ "name" : "John"
+}
+----
+
+The following record path expression would encode the String using Base64:
+
+|==========================================================
+| RecordPath | Return value
+| `base64Encode(/name)` | Sm9obg==
+|==========================================================
+
+=== base64Decode
+
+Decodes a Base64-encoded String or byte[]. For example, given a schema such as:
+
+----
+{
+ "type": "record",
+ "name": "events",
+ "fields": [
+ { "name": "name", "type": "string" }
+ ]
+}
+----
+
+and a record such as:
+
+----
+{
+ "name" : "Sm9obg=="
+}
+----
+
+The following record path expression would decode the String using Base64:
+
+|==========================================================
+| RecordPath | Return value
+| `base64Decode(/name)` | John
+|==========================================================
+
+
[[filter_functions]]
== Filter Functions