This is an automated email from the ASF dual-hosted git repository.
pvillard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new ceb9dff NIFI-7761 Allow HandleHttpRequest to add specified form data
to FlowFile attributes
ceb9dff is described below
commit ceb9dff3b9a8511c61e39e0728fcdd63f545ce91
Author: Otto Fowler <[email protected]>
AuthorDate: Mon Sep 7 01:10:31 2020 -0400
NIFI-7761 Allow HandleHttpRequest to add specified form data to FlowFile
attributes
Signed-off-by: Pierre Villard <[email protected]>
This closes #4513.
---
.../processors/standard/HandleHttpRequest.java | 38 ++++++++++-
.../standard/ITestHandleHttpRequest.java | 79 ++++++++++++++++++++--
2 files changed, 112 insertions(+), 5 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
index dbdb995ea..ad9d95d 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
@@ -89,6 +89,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
@InputRequirement(Requirement.INPUT_FORBIDDEN)
@@ -115,6 +116,8 @@ import java.util.regex.Pattern;
@WritesAttribute(attribute = "http.principal.name", description = "The
name of the authenticated user making the request"),
@WritesAttribute(attribute = "http.query.param.XXX", description = "Each
of query parameters in the request will be added as an attribute, "
+ "prefixed with \"http.query.param.\""),
+ @WritesAttribute(attribute = "http.param.XXX", description = "Form
parameters in the request that are configured by \"Parameters to Attributes
List\" will be added as an attribute, "
+ + "prefixed with \"http.param.\". Putting form parameters of large
size is not recommended."),
@WritesAttribute(attribute = HTTPUtils.HTTP_SSL_CERT, description = "The
Distinguished Name of the requestor. This value will not be populated "
+ "unless the Processor is configured to use an SSLContext
Service"),
@WritesAttribute(attribute = "http.issuer.dn", description = "The
Distinguished Name of the entity that issued the Subject's certificate. "
@@ -122,7 +125,7 @@ import java.util.regex.Pattern;
@WritesAttribute(attribute = "http.headers.XXX", description = "Each of
the HTTP Headers that is received in the request will be added as an "
+ "attribute, prefixed with \"http.headers.\" For example, if the
request contains an HTTP Header named \"x-my-header\", then the value "
+ "will be added to an attribute named
\"http.headers.x-my-header\""),
- @WritesAttribute(attribute = "http.headers.multipart.XXX", description =
"Each of the HTTP Headers that is received in the mulipart request will be
added as an "
+ @WritesAttribute(attribute = "http.headers.multipart.XXX", description =
"Each of the HTTP Headers that is received in the multipart request will be
added as an "
+ "attribute, prefixed with \"http.headers.multipart.\" For example,
if the multipart request contains an HTTP Header named \"content-disposition\",
then the value "
+ "will be added to an attribute named
\"http.headers.multipart.content-disposition\""),
@WritesAttribute(attribute = "http.multipart.size",
@@ -247,6 +250,14 @@ public class HandleHttpRequest extends AbstractProcessor {
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(ExpressionLanguageScope.NONE)
.build();
+ public static final PropertyDescriptor PARAMETERS_TO_ATTRIBUTES = new
PropertyDescriptor.Builder()
+ .name("parameters-to-attributes")
+ .displayName("Parameters to Attributes List")
+ .description("A comma-separated list of HTTP parameters or form
data to output as attributes")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .build();
public static final PropertyDescriptor CLIENT_AUTH = new
PropertyDescriptor.Builder()
.name("Client Authentication")
.description("Specifies whether or not the Processor should
authenticate clients. This value is ignored if the <SSL Context Service> "
@@ -304,6 +315,7 @@ public class HandleHttpRequest extends AbstractProcessor {
descriptors.add(CONTAINER_QUEUE_SIZE);
descriptors.add(MULTIPART_REQUEST_MAX_SIZE);
descriptors.add(MULTIPART_READ_BUFFER_SIZE);
+ descriptors.add(PARAMETERS_TO_ATTRIBUTES);
propertyDescriptors = Collections.unmodifiableList(descriptors);
}
@@ -312,6 +324,7 @@ public class HandleHttpRequest extends AbstractProcessor {
private AtomicBoolean initialized = new AtomicBoolean(false);
private volatile BlockingQueue<HttpRequestContainer> containerQueue;
private AtomicBoolean runOnPrimary = new AtomicBoolean(false);
+ private AtomicReference<Set<String>> parameterToAttributesReference = new
AtomicReference<>(null);
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
@@ -427,6 +440,18 @@ public class HandleHttpRequest extends AbstractProcessor {
}
}
+ final Set<String> parametersToMakeAttributes = new HashSet<>();
+ final String parametersToAttributesPropertyValue =
context.getProperty(PARAMETERS_TO_ATTRIBUTES).getValue();
+ if (parametersToAttributesPropertyValue != null) {
+ for (final String paremeterName :
parametersToAttributesPropertyValue.split(",")) {
+ final String trimmed = paremeterName.trim();
+ if (!trimmed.isEmpty()) {
+ parametersToMakeAttributes.add(trimmed);
+ }
+ }
+ parameterToAttributesReference.set(parametersToMakeAttributes);
+ }
+
final String pathRegex = context.getProperty(PATH_REGEX).getValue();
final Pattern pathPattern = (pathRegex == null) ? null :
Pattern.compile(pathRegex);
@@ -740,6 +765,17 @@ public class HandleHttpRequest extends AbstractProcessor {
putAttribute(attributes, "http.server.name",
request.getServerName());
putAttribute(attributes, HTTPUtils.HTTP_PORT,
request.getServerPort());
+ Set<String> parametersToAttributes =
parameterToAttributesReference.get();
+ if (parametersToAttributes != null &&
!parametersToAttributes.isEmpty()){
+ final Enumeration<String> paramEnumeration =
request.getParameterNames();
+ while (paramEnumeration.hasMoreElements()) {
+ final String paramName = paramEnumeration.nextElement();
+ if (parametersToAttributes.contains(paramName)){
+ attributes.put("http.param." + paramName,
request.getParameter(paramName));
+ }
+ }
+ }
+
final Cookie[] cookies = request.getCookies();
if (cookies != null) {
for (final Cookie cookie : cookies) {
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITestHandleHttpRequest.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITestHandleHttpRequest.java
index ece1a93..9ca9f04 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITestHandleHttpRequest.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITestHandleHttpRequest.java
@@ -19,10 +19,6 @@ package org.apache.nifi.processors.standard;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-import com.google.api.client.util.Charsets;
-import com.google.common.base.Optional;
-import com.google.common.collect.Iterables;
-import com.google.common.io.Files;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
@@ -46,6 +42,11 @@ import javax.net.ssl.SSLContext;
import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
+
+import com.google.api.client.util.Charsets;
+import com.google.common.base.Optional;
+import com.google.common.collect.Iterables;
+import com.google.common.io.Files;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.MediaType;
@@ -304,6 +305,76 @@ public class ITestHandleHttpRequest {
}
@Test(timeout = 30000)
+ public void testMultipartFormDataRequestCaptureFormAttributes() throws
InitializationException, IOException,
+ InterruptedException {
+ CountDownLatch serverReady = new CountDownLatch(1);
+ CountDownLatch requestSent = new CountDownLatch(1);
+
+ processor = createProcessor(serverReady, requestSent);
+ final TestRunner runner = TestRunners.newTestRunner(processor);
+ runner.setProperty(HandleHttpRequest.PORT, "0");
+ runner.setProperty(HandleHttpRequest.PARAMETERS_TO_ATTRIBUTES,
"p1,p2");
+
+ final MockHttpContextMap contextMap = new MockHttpContextMap();
+ runner.addControllerService("http-context-map", contextMap);
+ runner.enableControllerService(contextMap);
+ runner.setProperty(HandleHttpRequest.HTTP_CONTEXT_MAP,
"http-context-map");
+
+ final Thread httpThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ serverReady.await();
+
+ final int port = ((HandleHttpRequest)
runner.getProcessor()).getPort();
+
+ MultipartBody multipartBody = new MultipartBody.Builder()
+ .setType(MultipartBody.FORM)
+ .addFormDataPart("p1", "v1")
+ .addFormDataPart("p2", "v2")
+ .addFormDataPart("p3", "v3")
+ .build();
+
+ Request request = new Request.Builder()
+ .url(String.format("http://localhost:%s/my/path";,
port))
+ .post(multipartBody).build();
+
+ OkHttpClient client =
+ new OkHttpClient.Builder()
+ .readTimeout(3000, TimeUnit.MILLISECONDS)
+ .writeTimeout(3000, TimeUnit.MILLISECONDS)
+ .build();
+
+ sendRequest(client, request, requestSent);
+ } catch (Exception e) {
+ // Do nothing as HandleHttpRequest doesn't respond normally
+ }
+ }
+ });
+
+ httpThread.start();
+ runner.run(1, false, false);
+
+ runner.assertAllFlowFilesTransferred(HandleHttpRequest.REL_SUCCESS, 3);
+ assertEquals(1, contextMap.size());
+
+ List<MockFlowFile> flowFilesForRelationship =
runner.getFlowFilesForRelationship(HandleHttpRequest.REL_SUCCESS);
+
+ // Part fragments are not processed in the order we submitted them.
+ // We cannot rely on the order we sent them in.
+ for (int i = 1; i < 4; i++) {
+ MockFlowFile mff = findFlowFile(flowFilesForRelationship,
"http.multipart.name", String.format("p%d", i));
+ String contextId = mff.getAttribute(HTTPUtils.HTTP_CONTEXT_ID);
+ mff.assertAttributeEquals("http.multipart.name",
String.format("p%d", i));
+ mff.assertAttributeExists("http.param.p1");
+ mff.assertAttributeEquals("http.param.p1", "v1");
+ mff.assertAttributeExists("http.param.p2");
+ mff.assertAttributeEquals("http.param.p2", "v2");
+ mff.assertAttributeNotExists("http.param.p3");
+ }
+ }
+
+ @Test(timeout = 30000)
public void testMultipartFormDataRequestFailToRegisterContext() throws
InitializationException, IOException, InterruptedException {
CountDownLatch serverReady = new CountDownLatch(1);
CountDownLatch requestSent = new CountDownLatch(1);
