NIFI-5022 This closes #2588. InvokeAWSGatewayApi processor per review, simplify header building and remove new map Per review: - removed apache headers from 3rd party files per https://www.apache.org/legal/src-headers.html#3party - referenced commit/repo in notice file - created rat configuration to account for files without headers
Signed-off-by: joewitt <joew...@apache.org> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/946c2fe6 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/946c2fe6 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/946c2fe6 Branch: refs/heads/master Commit: 946c2fe6b3a094026c5d376676c67e7684bd18ee Parents: e3b0949 Author: Otto Fowler <ottobackwa...@gmail.com> Authored: Fri Mar 23 13:00:19 2018 -0400 Committer: joewitt <joew...@apache.org> Committed: Mon Jun 11 10:49:41 2018 -0400 ---------------------------------------------------------------------- NOTICE | 25 +- .../nifi-aws-abstract-processors/pom.xml | 21 +- .../processors/aws/AbstractAWSProcessor.java | 4 +- .../aws/wag/AbstractAWSGatewayApiProcessor.java | 675 +++++++ .../aws/wag/client/GenericApiGatewayClient.java | 114 ++ .../client/GenericApiGatewayClientBuilder.java | 76 + .../wag/client/GenericApiGatewayException.java | 13 + .../wag/client/GenericApiGatewayRequest.java | 45 + .../client/GenericApiGatewayRequestBuilder.java | 49 + .../wag/client/GenericApiGatewayResponse.java | 27 + .../processors/aws/wag/client/Validate.java | 17 + .../src/main/resources/META-INF/NOTICE | 25 + .../nifi-aws-bundle/nifi-aws-processors/pom.xml | 17 + .../processors/aws/wag/InvokeAWSGatewayApi.java | 385 ++++ .../org.apache.nifi.processor.Processor | 1 + .../nifi/processors/aws/wag/RequestMatcher.java | 48 + .../aws/wag/TestInvokeAWSGatewayApiCommon.java | 1729 ++++++++++++++++++ .../aws/wag/TestInvokeAmazonGatewayApiMock.java | 223 +++ ...keAmazonGatewayApiWithControllerService.java | 66 + ...nvokeInvokeAmazonGatewayApiWithCredFile.java | 66 + ...okeInvokeAmazonGatewayApiWithStaticAuth.java | 66 + 21 files changed, 3688 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/NOTICE ---------------------------------------------------------------------- diff --git a/NOTICE b/NOTICE index e0bb225..f5f1a7f 100644 --- a/NOTICE +++ b/NOTICE @@ -64,4 +64,27 @@ This includes derived works from Apache Solr available under Apache Software Lic This includes derived works from Apache Hadoop available under Apache Software License V2. Portions of the code found in https://github.com/apache/hadoop/blob/release-2.7.3-RC2/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/UserGroupInformation.java - The code can be found nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/KeytabUser.java \ No newline at end of file + The code can be found nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/KeytabUser.java +This includes derived works from apigateway-generic-java-sdk (ASLv2 licenced) project (https://github.com/rpgreen/apigateway-generic-java-sdk): + The derived work is adapted from + main/ca/ryangreen/apigateway/generic/ + GenericApiGatewayClient.java + GenericApiGatewayClientBuilder.java + GenericApiGatewayException.java + GenericApiGatewayRequest.java + GenericApiGatewayRequestBuilder.java + test/ca/ryangreen/apigateway/generic/ + GenericApiGatewayClientTest.java + LambdaMatcher.java + and can be found in the directories: + nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/../wag/client/ + GenericApiGatewayClient.java + GenericApiGatewayClientBuilder.java + GenericApiGatewayException.java + GenericApiGatewayRequest.java + GenericApiGatewayRequestBuilder.java + Validate.java + nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/../wag/ + RequestMatcher.java + GetAWSGatewayApiTest.java + http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/pom.xml b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/pom.xml index e41a921..8db4d9b 100644 --- a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/pom.xml +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/pom.xml @@ -54,6 +54,11 @@ <artifactId>aws-java-sdk-sqs</artifactId> </dependency> <dependency> + <groupId>commons-io</groupId> + <artifactId>commons-io</artifactId> + <version>2.6</version> + </dependency> + <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.7</version> @@ -81,5 +86,19 @@ <artifactId>nifi-proxy-configuration-api</artifactId> </dependency> </dependencies> - + <build> + <plugins> + <plugin> + <groupId>org.apache.rat</groupId> + <artifactId>apache-rat-plugin</artifactId> + <configuration> + <excludes> + <!-- 3rd party APL 2.0 code brought in, should not have headers --> + <!--https://www.apache.org/legal/src-headers.html#3party--> + <exclude>src/main/java/org/apache/nifi/processors/aws/wag/client/*.java</exclude> + </excludes> + </configuration> + </plugin> + </plugins> + </build> </project> http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java index bc8486d..d3867b0 100644 --- a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java @@ -156,11 +156,11 @@ public abstract class AbstractAWSProcessor<ClientType extends AmazonWebServiceCl private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH}; public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration.createProxyConfigPropertyDescriptor(true, PROXY_SPECS); - private static AllowableValue createAllowableValue(final Regions region) { + protected static AllowableValue createAllowableValue(final Regions region) { return new AllowableValue(region.getName(), AWSRegions.getRegionDisplayName(region.getName())); } - private static AllowableValue[] getAvailableRegions() { + protected static AllowableValue[] getAvailableRegions() { final List<AllowableValue> values = new ArrayList<>(); for (final Regions region : Regions.values()) { values.add(createAllowableValue(region)); http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/AbstractAWSGatewayApiProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/AbstractAWSGatewayApiProcessor.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/AbstractAWSGatewayApiProcessor.java new file mode 100644 index 0000000..0c310db --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/AbstractAWSGatewayApiProcessor.java @@ -0,0 +1,675 @@ +/* + * 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.processors.aws.wag; + +import static org.apache.commons.lang3.StringUtils.trimToEmpty; + +import com.amazonaws.ClientConfiguration; +import com.amazonaws.auth.AWSCredentials; +import com.amazonaws.auth.AWSCredentialsProvider; +import com.amazonaws.auth.AWSStaticCredentialsProvider; +import com.amazonaws.http.AmazonHttpClient; +import com.amazonaws.http.HttpMethodName; +import com.amazonaws.regions.Region; +import com.amazonaws.regions.Regions; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.InputStream; +import java.net.MalformedURLException; +import java.net.URI; +import java.net.URL; +import java.net.URLEncoder; +import java.util.ArrayList; +import java.util.Arrays; +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.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.commons.io.Charsets; +import org.apache.commons.lang3.StringUtils; +import org.apache.http.NameValuePair; +import org.apache.http.client.utils.URLEncodedUtils; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.expression.AttributeExpression; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.aws.AbstractAWSCredentialsProviderProcessor; +import org.apache.nifi.processors.aws.AbstractAWSProcessor; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayClient; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayClientBuilder; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayRequest; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayRequestBuilder; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayResponse; + +/** + * This class is the base class for invoking aws gateway api endpoints + */ +public abstract class AbstractAWSGatewayApiProcessor extends + AbstractAWSCredentialsProviderProcessor<GenericApiGatewayClient> { + + private volatile Set<String> dynamicPropertyNames = new HashSet<>(); + private volatile Pattern regexAttributesToSend = null; + private volatile AmazonHttpClient providedClient = null; + + public final static String STATUS_CODE = "aws.gateway.api.status.code"; + public final static String STATUS_MESSAGE = "aws.gateway.api.status.message"; + public final static String RESPONSE_BODY = "aws.gateway.api.response.body"; + public final static String RESOURCE_NAME_ATTR = "aws.gateway.api.resource"; + public final static String ENDPOINT_ATTR = "aws.gateway.api.endpoint"; + public final static String TRANSACTION_ID = "aws.gateway.api.tx.id"; + public final static String EXCEPTION_CLASS = "aws.gateway.api.java.exception.class"; + public final static String EXCEPTION_MESSAGE = "aws.gateway.api.java.exception.message"; + + protected static final String REL_RESPONSE_NAME = "Response"; + protected static final String REL_SUCCESS_REQ_NAME = "Original"; + protected static final String REL_RETRY_NAME = "Retry"; + protected static final String REL_NO_RETRY_NAME = "No Retry"; + protected static final String REL_FAILURE_NAME = "Failure"; + public static final String DEFAULT_CONTENT_TYPE = "application/octet-stream"; + + public AbstractAWSGatewayApiProcessor() { + } + + public AbstractAWSGatewayApiProcessor(AmazonHttpClient client) { + providedClient = client; + } + + + // Set of flowfile attributes which we generally always ignore during + // processing, including when converting http headers, copying attributes, etc. + // This set includes our strings defined above as well as some standard flowfile + // attributes. + public static final Set<String> IGNORED_ATTRIBUTES = Collections.unmodifiableSet(new HashSet<>( + Arrays.asList(STATUS_CODE, STATUS_MESSAGE, RESOURCE_NAME_ATTR, TRANSACTION_ID, "uuid", + "filename", "path"))); + + public static final PropertyDescriptor PROP_METHOD = new PropertyDescriptor.Builder() + .name("aws-gateway-http-method") + .displayName("HTTP Method") + .description( + "HTTP request method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS)." + + "Methods other than POST, PUT and PATCH will be sent without a message body.") + .required(true) + .defaultValue("GET") + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .addValidator(StandardValidators + .createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING)) + .build(); + + public static final PropertyDescriptor PROP_AWS_API_KEY = new PropertyDescriptor.Builder() + .name("aws-gateway-api-key") + .displayName("Amazon Gateway Api Key") + .description("The API Key") + .required(false) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_AWS_GATEWAY_API_ENDPOINT = new PropertyDescriptor.Builder() + .name("aws-gateway-api-endpoint") + .displayName("Amazon Gateway Api Endpoint") + .description("The Api Endpoint") + .required(true) + .addValidator(StandardValidators.URL_VALIDATOR) + .build(); + + // we use our own region, because the way the base sets the region after the client is created + // resets the endpoint and breaks everything + public static final PropertyDescriptor PROP_AWS_GATEWAY_API_REGION = new PropertyDescriptor.Builder() + .name("aws-gateway-region") + .displayName("Amazon Region") + .required(true) + .allowableValues(AbstractAWSProcessor.getAvailableRegions()) + .defaultValue(AbstractAWSProcessor.createAllowableValue(Regions.DEFAULT_REGION).getValue()) + .build(); + + public static final PropertyDescriptor PROP_RESOURCE_NAME = new PropertyDescriptor.Builder() + .name("aws-gateway-resource") + .displayName("Amazon Gateway Api ResourceName") + .description("The Name of the Gateway API Resource") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build(); + + public static final PropertyDescriptor PROP_QUERY_PARAMS = new PropertyDescriptor.Builder() + .name("aws-gateway-query-parameters") + .displayName("Query Parameters") + .description( + "The query parameters for this request in the form of Name=Value separated by &") + .displayName("Query Parameters") + .required(false) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .addValidator(StandardValidators + .createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING)) + .build(); + + public static final PropertyDescriptor PROP_ATTRIBUTES_TO_SEND = new PropertyDescriptor.Builder() + .name("aws-gateway-attributes-to-send") + .displayName("Attributes to Send") + .description( + "Regular expression that defines which attributes to send as HTTP headers in the request. " + + "If not defined, no attributes are sent as headers. Also any dynamic properties set will be sent as headers. " + + "The dynamic property key will be the header key and the dynamic property value will be interpreted as expression " + + "language will be the header value.") + .required(false) + .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_PUT_OUTPUT_IN_ATTRIBUTE = new PropertyDescriptor.Builder() + .name("aws-gateway-put-response-body-in-attribute") + .displayName("Put Response Body In Attribute") + .description( + "If set, the response body received back will be put into an attribute of the original FlowFile instead of a separate " + + "FlowFile. The attribute key to put to is determined by evaluating value of this property. ") + .addValidator(StandardValidators.createAttributeExpressionLanguageValidator( + AttributeExpression.ResultType.STRING)) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + + public static final PropertyDescriptor PROP_OUTPUT_RESPONSE_REGARDLESS = new PropertyDescriptor.Builder() + .name("aws-gateway-always-output-response") + .displayName("Always Output Response") + .description( + "Will force a response FlowFile to be generated and routed to the 'Response' relationship regardless of what the server status code received is " + + "or if the processor is configured to put the server response body in the request attribute. In the later configuration a request FlowFile with the " + + "response body in the attribute and a typical response FlowFile will be emitted to their respective relationships.") + .required(false) + .defaultValue("false") + .allowableValues("true", "false") + .build(); + + public static final PropertyDescriptor PROP_PENALIZE_NO_RETRY = new PropertyDescriptor.Builder() + .name("aws-gateway-penalize-no-retry") + .displayName("Penalize on \"No Retry\"") + .description("Enabling this property will penalize FlowFiles that are routed to the \"No Retry\" " + + "relationship.") + .required(false) + .defaultValue("false") + .allowableValues("true", "false") + .build(); + + public static final PropertyDescriptor PROP_PUT_ATTRIBUTE_MAX_LENGTH = new PropertyDescriptor.Builder() + .name("aws-gateway-max-attribute-length") + .displayName("Max Length To Put In Attribute") + .description( + "If routing the response body to an attribute of the original (by setting the \"Put response body in attribute\" " + + "property or by receiving an error status code), the number of characters put to the attribute value will be at " + + "most this amount. This is important because attributes are held in memory and large attributes will quickly " + + "cause out of memory issues. If the output goes longer than this value, it will be truncated to fit. " + + "Consider making this smaller if able.") + .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) + .defaultValue("256") + .build(); + + public static final PropertyDescriptor PROP_CONTENT_TYPE = new PropertyDescriptor.Builder() + .name("aws-gateway-content-type") + .displayName("Content-Type") + .description( + "The Content-Type to specify for when content is being transmitted through a PUT, POST or PATCH. " + + "In the case of an empty value after evaluating an expression language expression, Content-Type defaults to " + + DEFAULT_CONTENT_TYPE).required(true).expressionLanguageSupported(true) + .defaultValue("${" + CoreAttributes.MIME_TYPE.key() + "}") + .addValidator(StandardValidators + .createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING)) + .build(); + + public static final PropertyDescriptor PROP_SEND_BODY = new PropertyDescriptor.Builder() + .name("aws-gateway-send-message-body") + .displayName("Send Message Body") + .description("If true, sends the HTTP message body on POST/PUT/PATCH requests (default). If false, " + + "suppresses the message body and content-type header for these requests.") + .defaultValue("true") + .allowableValues("true", "false") + .required(false) + .build(); + + public static final PropertyDescriptor PROP_ADD_HEADERS_TO_REQUEST = new PropertyDescriptor.Builder() + .name("aws-gateway-add-response-headers-request") + .displayName("Add Response Headers To Request") + .description("Enabling this property saves all the response " + + "headers to the original request. This may be when the response headers are needed " + + "but a response is not generated due to the status code received.") + .required(false) + .defaultValue("false") + .allowableValues("true", "false") + .build(); + + public static final PropertyDescriptor PROP_CONNECT_TIMEOUT = new PropertyDescriptor.Builder() + .name("aws-gateway-connection-timeout") + .displayName("Connection Timeout") + .description("Max wait time for connection to remote service.") + .required(false) + .defaultValue("10 secs") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_READ_TIMEOUT = new PropertyDescriptor.Builder() + .name("aws-gateway-read-timeout") + .displayName("Read Timeout") + .description("Max wait time for response from remote service.") + .required(false) + .defaultValue("50 secs") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + @Override + protected PropertyDescriptor getSupportedDynamicPropertyDescriptor( + String propertyDescriptorName) { + return new PropertyDescriptor.Builder() + .required(false) + .name(propertyDescriptorName) + .addValidator(StandardValidators.createAttributeExpressionLanguageValidator( + AttributeExpression.ResultType.STRING, + true)) + .dynamic(true) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES).build(); + } + + + @Override + public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, + final String newValue) { + if (descriptor.isDynamic()) { + final Set<String> newDynamicPropertyNames = new HashSet<>(dynamicPropertyNames); + if (newValue == null) { + newDynamicPropertyNames.remove(descriptor.getName()); + } else if (oldValue == null) { // new property + newDynamicPropertyNames.add(descriptor.getName()); + } + this.dynamicPropertyNames = Collections.unmodifiableSet(newDynamicPropertyNames); + } else { + // compile the attributes-to-send filter pattern + if (PROP_ATTRIBUTES_TO_SEND.getName().equalsIgnoreCase(descriptor.getName())) { + if (newValue == null || newValue.isEmpty()) { + regexAttributesToSend = null; + } else { + final String trimmedValue = StringUtils.trimToEmpty(newValue); + regexAttributesToSend = Pattern.compile(trimmedValue); + } + } + } + } + + @Override + protected Collection<ValidationResult> customValidate( + final ValidationContext validationContext) { + final List<ValidationResult> results = new ArrayList<>(3); + results.addAll(super.customValidate(validationContext)); + final boolean querySet = validationContext.getProperty(PROP_QUERY_PARAMS).isSet(); + + if (querySet) { + String input = validationContext.getProperty(PROP_QUERY_PARAMS).getValue(); + // if we have expressions, we don't do further validation + if (!(validationContext.isExpressionLanguageSupported(PROP_QUERY_PARAMS.getName()) + && validationContext.isExpressionLanguagePresent(input))) { + + try { + final String evaluatedInput = validationContext.newPropertyValue(input) + .evaluateAttributeExpressions() + .getValue(); + // user is not expected to encode, that will be done by the aws client + // but we may need to when validating + final String encodedInput = URLEncoder.encode(evaluatedInput, "UTF-8"); + final String url = String.format("http://www.foo.com?%s";, encodedInput); + new URL(url); + results.add(new ValidationResult.Builder().subject(PROP_QUERY_PARAMS.getName()) + .input(input) + .explanation("Valid URL params") + .valid(true).build()); + } catch (final Exception e) { + results.add(new ValidationResult.Builder().subject(PROP_QUERY_PARAMS.getName()) + .input(input).explanation( + "Not a valid set of URL params").valid(false).build()); + } + } + } + String method = trimToEmpty(validationContext.getProperty(PROP_METHOD).getValue()) + .toUpperCase(); + + // if there are expressions do not validate + if (!(validationContext.isExpressionLanguageSupported(PROP_METHOD.getName()) + && validationContext.isExpressionLanguagePresent(method))) { + try { + HttpMethodName methodName = HttpMethodName.fromValue(method); + } catch (IllegalArgumentException e) { + results.add(new ValidationResult.Builder().subject(PROP_METHOD.getName()).input(method) + .explanation("Unsupported METHOD") + .valid(false).build()); + } + } + + return results; + } + + @Override + protected GenericApiGatewayClient createClient(ProcessContext context, + AWSCredentialsProvider awsCredentialsProvider, + ClientConfiguration clientConfiguration) { + + GenericApiGatewayClientBuilder builder = new GenericApiGatewayClientBuilder() + .withCredentials(awsCredentialsProvider).withClientConfiguration(clientConfiguration) + .withEndpoint(context.getProperty(PROP_AWS_GATEWAY_API_ENDPOINT).getValue()).withRegion( + Region.getRegion( + Regions.fromName(context.getProperty(PROP_AWS_GATEWAY_API_REGION).getValue()))); + if (context.getProperty(PROP_AWS_API_KEY).isSet()) { + builder = builder.withApiKey(context.getProperty(PROP_AWS_API_KEY).evaluateAttributeExpressions().getValue()); + } + if (providedClient != null) { + builder = builder.withHttpClient(providedClient); + } + return builder.build(); + } + + @Override + @Deprecated + protected GenericApiGatewayClient createClient(final ProcessContext context, + final AWSCredentials credentials, + final ClientConfiguration clientConfiguration) { + return createClient(context, new AWSStaticCredentialsProvider(credentials), + clientConfiguration); + } + + protected GenericApiGatewayRequest configureRequest(final ProcessContext context, + final ProcessSession session, + final String resourcePath, + final FlowFile requestFlowFile) { + String method = trimToEmpty( + context.getProperty(PROP_METHOD).evaluateAttributeExpressions(requestFlowFile) + .getValue()).toUpperCase(); + HttpMethodName methodName = HttpMethodName.fromValue(method); + return configureRequest(context, session, resourcePath,requestFlowFile, methodName); + } + + protected GenericApiGatewayRequest configureRequest(final ProcessContext context, + final ProcessSession session, + final String resourcePath, + final FlowFile requestFlowFile, + final HttpMethodName methodName) { + + GenericApiGatewayRequestBuilder builder = new GenericApiGatewayRequestBuilder() + .withResourcePath(resourcePath); + final Map<String, List<String>> parameters = getParameters(context); + builder = builder.withParameters(parameters); + + InputStream requestBody = null; + switch (methodName) { + case GET: + builder = builder.withHttpMethod(HttpMethodName.GET); + break; + case POST: + requestBody = getRequestBodyToSend(session, context, requestFlowFile); + builder = builder.withHttpMethod(HttpMethodName.POST).withBody(requestBody); + break; + case PUT: + requestBody = getRequestBodyToSend(session, context, requestFlowFile); + builder = builder.withHttpMethod(HttpMethodName.PUT).withBody(requestBody); + break; + case PATCH: + requestBody = getRequestBodyToSend(session, context, requestFlowFile); + builder = builder.withHttpMethod(HttpMethodName.PATCH).withBody(requestBody); + break; + case HEAD: + builder = builder.withHttpMethod(HttpMethodName.HEAD); + break; + case DELETE: + builder = builder.withHttpMethod(HttpMethodName.DELETE); + break; + case OPTIONS: + requestBody = getRequestBodyToSend(session, context, requestFlowFile); + builder = builder.withHttpMethod(HttpMethodName.OPTIONS).withBody(requestBody); + break; + } + + builder = setHeaderProperties(context, builder, methodName, requestFlowFile); + return builder.build(); + } + + protected InputStream getRequestBodyToSend(final ProcessSession session, + final ProcessContext context, + final FlowFile requestFlowFile) { + + if (context.getProperty(PROP_SEND_BODY).asBoolean() && requestFlowFile != null) { + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + session.exportTo(requestFlowFile, outputStream); + return new ByteArrayInputStream(outputStream.toByteArray()); + + } else { + return new ByteArrayInputStream(new byte[0]); + } + } + + protected GenericApiGatewayRequestBuilder setHeaderProperties(final ProcessContext context, + GenericApiGatewayRequestBuilder requestBuilder, + HttpMethodName methodName, + final FlowFile requestFlowFile) { + + Map<String, String> headers = new HashMap<>(); + for (String headerKey : dynamicPropertyNames) { + String headerValue = context.getProperty(headerKey) + .evaluateAttributeExpressions(requestFlowFile).getValue(); + headers.put(headerKey, headerValue); + } + + // iterate through the flowfile attributes, adding any attribute that + // matches the attributes-to-send pattern. if the pattern is not set + // (it's an optional property), ignore that attribute entirely + if (regexAttributesToSend != null && requestFlowFile != null) { + Map<String, String> attributes = requestFlowFile.getAttributes(); + Matcher m = regexAttributesToSend.matcher(""); + for (Map.Entry<String, String> entry : attributes.entrySet()) { + String headerKey = trimToEmpty(entry.getKey()); + + // don't include any of the ignored attributes + if (IGNORED_ATTRIBUTES.contains(headerKey)) { + continue; + } + + // check if our attribute key matches the pattern + // if so, include in the request as a header + m.reset(headerKey); + if (m.matches()) { + String headerVal = trimToEmpty(entry.getValue()); + headers.put(headerKey, headerVal); + } + } + } + + String contentType = context.getProperty(PROP_CONTENT_TYPE) + .evaluateAttributeExpressions(requestFlowFile).getValue(); + boolean sendBody = context.getProperty(PROP_SEND_BODY).asBoolean(); + contentType = StringUtils.isBlank(contentType) ? DEFAULT_CONTENT_TYPE : contentType; + if (methodName == HttpMethodName.PUT || methodName == HttpMethodName.POST + || methodName == HttpMethodName.PATCH) { + if (sendBody) { + headers.put("Content-Type", contentType); + } + } else { + headers.put("Content-Type", contentType); + } + + if (!headers.isEmpty()) { + requestBuilder = requestBuilder.withHeaders(headers); + } + + return requestBuilder; + } + + /** + * Returns a map of Query Parameter Name to Values + * + * @param context ProcessContext + * @return Map of names and values + */ + protected Map<String, List<String>> getParameters(ProcessContext context) { + + if (!context.getProperty(PROP_QUERY_PARAMS).isSet()) { + return new HashMap<>(); + } + final String queryString = context.getProperty(PROP_QUERY_PARAMS) + .evaluateAttributeExpressions().getValue(); + List<NameValuePair> params = URLEncodedUtils + .parse(queryString, Charsets.toCharset("UTF-8")); + + if (params.isEmpty()) { + return new HashMap<>(); + } + + Map<String, List<String>> map = new HashMap<>(); + + for (NameValuePair nvp : params) { + if (!map.containsKey(nvp.getName())) { + map.put(nvp.getName(), new ArrayList<>()); + } + map.get(nvp.getName()).add(nvp.getValue()); + } + return map; + } + + /** + * Returns a Map of flowfile attributes from the response http headers. Multivalue headers are naively converted to comma separated strings. + */ + protected Map<String, String> convertAttributesFromHeaders( + GenericApiGatewayResponse responseHttp) { + // create a new hashmap to store the values from the connection + Map<String, String> map = new HashMap<>(); + responseHttp.getHttpResponse().getHeaders().entrySet().forEach((entry) -> { + + String key = entry.getKey(); + String value = entry.getValue(); + + if (key == null) { + return; + } + + // we ignore any headers with no actual values (rare) + if (StringUtils.isBlank(value)) { + return; + } + + // put the csv into the map + map.put(key, value); + }); + + return map; + } + + + protected Relationship getRelationshipForName(String name, Set<Relationship> relationships) { + for (Relationship relationship : relationships) { + if (relationship.getName().equals(name)) { + return relationship; + } + } + throw new IllegalStateException("Unknown relationship " + name); + } + + protected void route(FlowFile request, FlowFile response, ProcessSession session, + ProcessContext context, int statusCode, Set<Relationship> relationships) { + // check if we should yield the processor + if (!isSuccess(statusCode) && request == null) { + context.yield(); + } + + // If the property to output the response flowfile regardless of status code is set then transfer it + boolean responseSent = false; + if (context.getProperty(PROP_OUTPUT_RESPONSE_REGARDLESS).asBoolean()) { + session.transfer(response, getRelationshipForName(REL_RESPONSE_NAME, relationships)); + responseSent = true; + } + + // transfer to the correct relationship + // 2xx -> SUCCESS + if (isSuccess(statusCode)) { + // we have two flowfiles to transfer + if (request != null) { + session + .transfer(request, getRelationshipForName(REL_SUCCESS_REQ_NAME, relationships)); + } + if (response != null && !responseSent) { + session + .transfer(response, getRelationshipForName(REL_RESPONSE_NAME, relationships)); + } + + // 5xx -> RETRY + } else if (statusCode / 100 == 5) { + if (request != null) { + request = session.penalize(request); + session.transfer(request, getRelationshipForName(REL_RETRY_NAME, relationships)); + } + + // 1xx, 3xx, 4xx -> NO RETRY + } else { + if (request != null) { + if (context.getProperty(PROP_PENALIZE_NO_RETRY).asBoolean()) { + request = session.penalize(request); + } + session.transfer(request, getRelationshipForName(REL_NO_RETRY_NAME, relationships)); + } + } + + } + + protected boolean isSuccess(int statusCode) { + return statusCode / 100 == 2; + } + + protected void logRequest(ComponentLog logger, URI endpoint, GenericApiGatewayRequest request) { + try { + logger.debug("\nRequest to remote service:\n\t{}\t{}\t\n{}", + new Object[]{endpoint.toURL().toExternalForm(), request.getHttpMethod(), getLogString(request.getHeaders())}); + } catch (MalformedURLException e) { + logger.debug(e.getMessage()); + } + } + + protected void logResponse(ComponentLog logger, GenericApiGatewayResponse response) { + try { + logger.debug("\nResponse from remote service:\n\t{}\n{}", + new Object[]{response.getHttpResponse().getHttpRequest().getURI().toURL().toExternalForm(), getLogString(response.getHttpResponse().getHeaders())}); + } catch (MalformedURLException e) { + logger.debug(e.getMessage()); + } + } + + protected String getLogString(Map<String, String> map) { + StringBuilder sb = new StringBuilder(); + if(map != null && map.size() > 0) { + for (Map.Entry<String, String> entry : map.entrySet()) { + String value = entry.getValue(); + sb.append("\t"); + sb.append(entry.getKey()); + sb.append(": "); + sb.append(value); + sb.append("\n"); + } + } + return sb.toString(); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClient.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClient.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClient.java new file mode 100644 index 0000000..e156370 --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClient.java @@ -0,0 +1,114 @@ +package org.apache.nifi.processors.aws.wag.client; + +import com.amazonaws.AmazonServiceException; +import com.amazonaws.AmazonWebServiceClient; +import com.amazonaws.ClientConfiguration; +import com.amazonaws.DefaultRequest; +import com.amazonaws.auth.AWS4Signer; +import com.amazonaws.auth.AWSCredentialsProvider; +import com.amazonaws.http.AmazonHttpClient; +import com.amazonaws.http.ExecutionContext; +import com.amazonaws.http.HttpMethodName; +import com.amazonaws.http.HttpResponseHandler; +import com.amazonaws.http.JsonResponseHandler; +import com.amazonaws.internal.auth.DefaultSignerProvider; +import com.amazonaws.protocol.json.JsonOperationMetadata; +import com.amazonaws.protocol.json.SdkStructuredPlainJsonFactory; +import com.amazonaws.regions.Region; +import com.amazonaws.transform.JsonErrorUnmarshaller; +import com.amazonaws.transform.JsonUnmarshallerContext; +import com.amazonaws.transform.Unmarshaller; +import com.fasterxml.jackson.databind.JsonNode; +import java.io.InputStream; +import java.net.URI; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class GenericApiGatewayClient extends AmazonWebServiceClient { + private static final String API_GATEWAY_SERVICE_NAME = "execute-api"; + private static final String API_KEY_HEADER = "x-api-key"; + + private final JsonResponseHandler<GenericApiGatewayResponse> responseHandler; + private final HttpResponseHandler<AmazonServiceException> errorResponseHandler; + private final AWSCredentialsProvider credentials; + private String apiKey; + private final AWS4Signer signer; + + GenericApiGatewayClient(ClientConfiguration clientConfiguration, String endpoint, Region region, + AWSCredentialsProvider credentials, String apiKey, AmazonHttpClient httpClient) { + super(clientConfiguration); + setRegion(region); + setEndpoint(endpoint); + this.credentials = credentials; + this.apiKey = apiKey; + this.signer = new AWS4Signer(); + this.signer.setServiceName(API_GATEWAY_SERVICE_NAME); + this.signer.setRegionName(region.getName()); + + final JsonOperationMetadata metadata = new JsonOperationMetadata().withHasStreamingSuccessResponse(false).withPayloadJson(false); + final Unmarshaller<GenericApiGatewayResponse, JsonUnmarshallerContext> responseUnmarshaller = in -> new GenericApiGatewayResponse(in.getHttpResponse()); + this.responseHandler = SdkStructuredPlainJsonFactory.SDK_JSON_FACTORY.createResponseHandler(metadata, responseUnmarshaller); + JsonErrorUnmarshaller defaultErrorUnmarshaller = new JsonErrorUnmarshaller(GenericApiGatewayException.class, null) { + @Override + public AmazonServiceException unmarshall(JsonNode jsonContent) throws Exception { + return new GenericApiGatewayException(jsonContent.toString()); + } + }; + this.errorResponseHandler = SdkStructuredPlainJsonFactory.SDK_JSON_FACTORY.createErrorResponseHandler( + Collections.singletonList(defaultErrorUnmarshaller), null); + + if (httpClient != null) { + super.client = httpClient; + } + } + + public GenericApiGatewayResponse execute(GenericApiGatewayRequest request) { + return execute(request.getHttpMethod(), request.getResourcePath(), request.getHeaders(), request.getParameters(), request.getBody()); + } + + private GenericApiGatewayResponse execute(HttpMethodName method, String resourcePath, Map<String, String> headers, Map<String,List<String>> parameters, InputStream content) { + final ExecutionContext executionContext = buildExecutionContext(); + + DefaultRequest request = new DefaultRequest(API_GATEWAY_SERVICE_NAME); + request.setHttpMethod(method); + request.setContent(content); + request.setEndpoint(this.endpoint); + request.setResourcePath(resourcePath); + request.setHeaders(buildRequestHeaders(headers, apiKey)); + if (parameters != null) { + request.setParameters(parameters); + } + return this.client.execute(request, responseHandler, errorResponseHandler, executionContext).getAwsResponse(); + } + + private ExecutionContext buildExecutionContext() { + final ExecutionContext executionContext = ExecutionContext.builder().withSignerProvider( + new DefaultSignerProvider(this, signer)).build(); + executionContext.setCredentialsProvider(credentials); + executionContext.setSigner(signer); + return executionContext; + } + + private Map<String, String> buildRequestHeaders(Map<String, String> headers, String apiKey) { + if (headers == null) { + headers = new HashMap<>(); + } + if (apiKey != null) { + headers.put(API_KEY_HEADER, apiKey); + } + return headers; + } + + public URI getEndpoint() { + return this.endpoint; + } + + @Override + protected String getServiceNameIntern() { + return API_GATEWAY_SERVICE_NAME; + } +} + + http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClientBuilder.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClientBuilder.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClientBuilder.java new file mode 100644 index 0000000..990f369 --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayClientBuilder.java @@ -0,0 +1,76 @@ +package org.apache.nifi.processors.aws.wag.client; + +import com.amazonaws.ClientConfiguration; +import com.amazonaws.auth.AWSCredentialsProvider; +import com.amazonaws.http.AmazonHttpClient; +import com.amazonaws.regions.Region; + +public class GenericApiGatewayClientBuilder { + private String endpoint; + private Region region; + private AWSCredentialsProvider credentials; + private ClientConfiguration clientConfiguration; + private String apiKey; + private AmazonHttpClient httpClient; + + public GenericApiGatewayClientBuilder withEndpoint(String endpoint) { + this.endpoint = endpoint; + return this; + } + + public GenericApiGatewayClientBuilder withRegion(Region region) { + this.region = region; + return this; + } + + public GenericApiGatewayClientBuilder withClientConfiguration(ClientConfiguration clientConfiguration) { + this.clientConfiguration = clientConfiguration; + return this; + } + + public GenericApiGatewayClientBuilder withCredentials(AWSCredentialsProvider credentials) { + this.credentials = credentials; + return this; + } + + public GenericApiGatewayClientBuilder withApiKey(String apiKey) { + this.apiKey = apiKey; + return this; + } + + public GenericApiGatewayClientBuilder withHttpClient(AmazonHttpClient client) { + this.httpClient = client; + return this; + } + + public AWSCredentialsProvider getCredentials() { + return credentials; + } + + public String getApiKey() { + return apiKey; + } + + public AmazonHttpClient getHttpClient() { + return httpClient; + } + + public String getEndpoint() { + return endpoint; + } + + public Region getRegion() { + return region; + } + + public ClientConfiguration getClientConfiguration() { + return clientConfiguration; + } + + public GenericApiGatewayClient build() { + Validate.notEmpty(endpoint, "Endpoint"); + Validate.notNull(region, "Region"); + return new GenericApiGatewayClient(clientConfiguration, endpoint, region, credentials, apiKey, httpClient); + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayException.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayException.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayException.java new file mode 100644 index 0000000..606143d --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayException.java @@ -0,0 +1,13 @@ +package org.apache.nifi.processors.aws.wag.client; + +import com.amazonaws.AmazonServiceException; + +public class GenericApiGatewayException extends AmazonServiceException { + public GenericApiGatewayException(String errorMessage) { + super(errorMessage); + } + + public GenericApiGatewayException(String errorMessage, Exception cause) { + super(errorMessage, cause); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequest.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequest.java new file mode 100644 index 0000000..f0649cd --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequest.java @@ -0,0 +1,45 @@ +package org.apache.nifi.processors.aws.wag.client; + +import com.amazonaws.http.HttpMethodName; +import java.io.InputStream; +import java.util.List; +import java.util.Map; + +public class GenericApiGatewayRequest { + + private final HttpMethodName httpMethod; + private final String resourcePath; + private final InputStream body; + private final Map<String, String> headers; + private final Map<String, List<String>> parameters; + + public GenericApiGatewayRequest(HttpMethodName httpMethod, String resourcePath, + InputStream body, Map<String, String> headers, + Map<String, List<String>> parameters) { + this.httpMethod = httpMethod; + this.resourcePath = resourcePath; + this.body = body; + this.headers = headers; + this.parameters = parameters; + } + + public HttpMethodName getHttpMethod() { + return httpMethod; + } + + public String getResourcePath() { + return resourcePath; + } + + public InputStream getBody() { + return body; + } + + public Map<String, String> getHeaders() { + return headers; + } + + public Map<String, List<String>> getParameters() { + return parameters; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequestBuilder.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequestBuilder.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequestBuilder.java new file mode 100644 index 0000000..10e712a --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayRequestBuilder.java @@ -0,0 +1,49 @@ +package org.apache.nifi.processors.aws.wag.client; + +import com.amazonaws.http.HttpMethodName; +import java.io.InputStream; +import java.util.List; +import java.util.Map; + +public class GenericApiGatewayRequestBuilder { + private HttpMethodName httpMethod; + private String resourcePath; + private InputStream body; + private Map<String, String> headers; + private Map<String, List<String>> parameters; + + public GenericApiGatewayRequestBuilder withHttpMethod(HttpMethodName name) { + httpMethod = name; + return this; + } + + public GenericApiGatewayRequestBuilder withResourcePath(String path) { + resourcePath = path; + return this; + } + + public GenericApiGatewayRequestBuilder withBody(InputStream content) { + this.body = content; + return this; + } + + public GenericApiGatewayRequestBuilder withHeaders(Map<String, String> headers) { + this.headers = headers; + return this; + } + + public GenericApiGatewayRequestBuilder withParameters(Map<String,List<String>> parameters) { + this.parameters = parameters; + return this; + } + + public boolean hasBody() { + return this.body != null; + } + + public GenericApiGatewayRequest build() { + Validate.notNull(httpMethod, "HTTP method"); + Validate.notEmpty(resourcePath, "Resource path"); + return new GenericApiGatewayRequest(httpMethod, resourcePath, body, headers, parameters); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayResponse.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayResponse.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayResponse.java new file mode 100644 index 0000000..0f616d8 --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/GenericApiGatewayResponse.java @@ -0,0 +1,27 @@ +package org.apache.nifi.processors.aws.wag.client; + +import com.amazonaws.http.HttpResponse; +import com.amazonaws.util.IOUtils; +import java.io.IOException; + +public class GenericApiGatewayResponse { + private final HttpResponse httpResponse; + private final String body; + + public GenericApiGatewayResponse(HttpResponse httpResponse) throws IOException { + this.httpResponse = httpResponse; + if(httpResponse.getContent() != null) { + this.body = IOUtils.toString(httpResponse.getContent()); + }else { + this.body = null; + } + } + + public HttpResponse getHttpResponse() { + return httpResponse; + } + + public String getBody() { + return body; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/Validate.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/Validate.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/Validate.java new file mode 100644 index 0000000..a85197b --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/wag/client/Validate.java @@ -0,0 +1,17 @@ +package org.apache.nifi.processors.aws.wag.client; + +import com.amazonaws.util.StringUtils; + +public class Validate { + public static void notEmpty(String in, String fieldName) { + if (StringUtils.isNullOrEmpty(in)) { + throw new IllegalArgumentException(String.format("%s cannot be empty", fieldName)); + } + } + + public static void notNull(Object in, String fieldName) { + if (in == null) { + throw new IllegalArgumentException(String.format("%s cannot be null", fieldName)); + } + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-nar/src/main/resources/META-INF/NOTICE ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-nar/src/main/resources/META-INF/NOTICE b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-nar/src/main/resources/META-INF/NOTICE index c4a5688..1099746 100644 --- a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-nar/src/main/resources/META-INF/NOTICE +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-nar/src/main/resources/META-INF/NOTICE @@ -92,3 +92,28 @@ The following binary components are provided under the Apache Software License v A list of contributors may be found from CREDITS file, which is included in some artifacts (usually source distributions); but is always available from the source code management (SCM) system project uses. + + + (ASLv2) This includes derived works from apigateway-generic-java-sdk project (https://github.com/rpgreen/apigateway-generic-java-sdk) + https://github.com/rpgreen/apigateway-generic-java-sdk/commit/32eea44cc855a530c9b4a28b9f3601a41bc85618 as the point reference: + The derived work is adapted from + main/ca/ryangreen/apigateway/generic/ + GenericApiGatewayClient.java + GenericApiGatewayClientBuilder.java + GenericApiGatewayException.java + GenericApiGatewayRequest.java + GenericApiGatewayRequestBuilder.java + test/ca/ryangreen/apigateway/generic/ + GenericApiGatewayClientTest.java + LambdaMatcher.java + and can be found in the directories: + nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/../wag/client/ + GenericApiGatewayClient.java + GenericApiGatewayClientBuilder.java + GenericApiGatewayException.java + GenericApiGatewayRequest.java + GenericApiGatewayRequestBuilder.java + Validate.java + nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/../wag/ + RequestMatcher.java + GetAWSGatewayApiTest.java \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/pom.xml b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/pom.xml index e9041b3..db2b989 100644 --- a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/pom.xml +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/pom.xml @@ -52,6 +52,23 @@ </dependency> <dependency> <groupId>org.apache.nifi</groupId> + <artifactId>nifi-standard-processors</artifactId> + <version>1.7.0-SNAPSHOT</version> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.eclipse.jetty</groupId> + <artifactId>jetty-server</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-simple</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> <artifactId>nifi-ssl-context-service-api</artifactId> </dependency> <dependency> http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java new file mode 100644 index 0000000..de990e4 --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/wag/InvokeAWSGatewayApi.java @@ -0,0 +1,385 @@ +/* + * 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.processors.aws.wag; + +import com.amazonaws.http.AmazonHttpClient; +import org.apache.http.impl.EnglishReasonPhraseCatalog; +import org.apache.nifi.annotation.behavior.DynamicProperty; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.SupportsBatching; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayClient; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayException; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayRequest; +import org.apache.nifi.processors.aws.wag.client.GenericApiGatewayResponse; +import org.apache.nifi.stream.io.StreamUtils; + +import java.io.ByteArrayInputStream; +import java.util.Arrays; +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.UUID; +import java.util.concurrent.TimeUnit; + +@TriggerWhenEmpty +@SupportsBatching +@InputRequirement(Requirement.INPUT_ALLOWED) +@Tags({"Amazon", "AWS", "Client", "Gateway-API", "Rest", "http", "https"}) +@CapabilityDescription("Client for AWS Gateway API endpoint") +@WritesAttributes({ + @WritesAttribute(attribute = "mime.type", description = "The MIME Type of the flowfiles"), + @WritesAttribute(attribute = "aws.gateway.api.status.code", description = "The status code that is returned"), + @WritesAttribute(attribute = "aws.gateway.api.status.message", description = "The status message that is returned"), + @WritesAttribute(attribute = "aws.gateway.api.response.body", description = "In the instance where the status code received is not a success (2xx)"), + @WritesAttribute(attribute = "aws.gateway.api.resource", description = "The request resource"), + @WritesAttribute(attribute = "aws.gateway.api.tx.id", description = "The transaction ID that is returned after reading the response"), + @WritesAttribute(attribute = "aws.gateway.api.java.exception.class", description = "The Java exception class raised when the processor fails"), + @WritesAttribute(attribute = "aws.gateway.api.java.exception.message", description = "The Java exception message raised when the processor fails"),}) +@DynamicProperty(name = "Header Name", value = "Attribute Expression Language", supportsExpressionLanguage = true, description = + "Send request header " + + "with a key matching the Dynamic Property Key and a value created by evaluating the Attribute Expression Language set in the value " + + "of the Dynamic Property.") +public class InvokeAWSGatewayApi extends AbstractAWSGatewayApiProcessor { + + public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(Arrays + .asList( + PROP_METHOD, + PROP_AWS_GATEWAY_API_REGION, + ACCESS_KEY, + SECRET_KEY, + CREDENTIALS_FILE, + AWS_CREDENTIALS_PROVIDER_SERVICE, + TIMEOUT, + PROP_RESOURCE_NAME, + PROP_AWS_GATEWAY_API_ENDPOINT, + PROP_AWS_API_KEY, + PROP_ATTRIBUTES_TO_SEND, + PROP_PUT_OUTPUT_IN_ATTRIBUTE, + PROP_CONTENT_TYPE, + PROP_SEND_BODY, + PROP_OUTPUT_RESPONSE_REGARDLESS, + PROP_PENALIZE_NO_RETRY, + PROXY_HOST, + PROXY_HOST_PORT, + PROXY_USERNAME, + PROXY_PASSWORD, + PROP_QUERY_PARAMS, + PROP_PUT_ATTRIBUTE_MAX_LENGTH, + PROP_ADD_HEADERS_TO_REQUEST, + PROXY_CONFIGURATION_SERVICE)); + + + public static final Relationship REL_SUCCESS_REQ = new Relationship.Builder() + .name(REL_SUCCESS_REQ_NAME) + .description("The original FlowFile will be routed upon success (2xx status codes). It will have new " + + "attributes detailing the success of the request.") + .build(); + + public static final Relationship REL_RESPONSE = new Relationship.Builder() + .name(REL_RESPONSE_NAME) + .description("A Response FlowFile will be routed upon success (2xx status codes). If the 'Output Response " + + "Regardless' property is true then the response will be sent to this relationship regardless of " + + "the status code received.") + .build(); + + public static final Relationship REL_RETRY = new Relationship.Builder() + .name(REL_RETRY_NAME) + .description("The original FlowFile will be routed on any status code that can be retried " + + "(5xx status codes). It will have new attributes detailing the request.") + .build(); + + public static final Relationship REL_NO_RETRY = new Relationship.Builder() + .name(REL_NO_RETRY_NAME) + .description("The original FlowFile will be routed on any status code that should NOT be retried " + + "(1xx, 3xx, 4xx status codes). It will have new attributes detailing the request.") + .build(); + + public static final Relationship REL_FAILURE = new Relationship.Builder() + .name(REL_FAILURE_NAME) + .description("The original FlowFile will be routed on any type of connection failure, timeout or general " + + "exception. It will have new attributes detailing the request.") + .build(); + + public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>( + Arrays.asList(REL_SUCCESS_REQ, REL_RESPONSE, REL_RETRY, REL_NO_RETRY, REL_FAILURE))); + + @Override + public Set<Relationship> getRelationships() { + return RELATIONSHIPS; + } + + public InvokeAWSGatewayApi() { + super(); + } + + public InvokeAWSGatewayApi(AmazonHttpClient client) { + super(client); + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return properties; + } + + @Override + public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { + ComponentLog logger = getLogger(); + FlowFile requestFlowFile = session.get(); + + // Checking to see if the property to put the body of the response in an attribute was set + boolean putToAttribute = context.getProperty(PROP_PUT_OUTPUT_IN_ATTRIBUTE).isSet(); + if (requestFlowFile == null) { + String request = context.getProperty(PROP_METHOD).evaluateAttributeExpressions() + .getValue().toUpperCase(); + if ("POST".equals(request) || "PUT".equals(request) || "PATCH".equals(request)) { + return; + } else if (putToAttribute) { + requestFlowFile = session.create(); + } + } + + // Every request/response cycle has a unique transaction id which will be stored as a flowfile attribute. + final UUID txId = UUID.randomUUID(); + FlowFile responseFlowFile = null; + + try { + final int maxAttributeSize = context.getProperty(PROP_PUT_ATTRIBUTE_MAX_LENGTH) + .asInteger(); + + final String resourceName = context.getProperty(PROP_RESOURCE_NAME).getValue(); + + final GenericApiGatewayClient client = getClient(); + + final GenericApiGatewayRequest request = configureRequest(context, session, + resourceName, + requestFlowFile); + + logRequest(logger, client.getEndpoint(), request); + final long startNanos = System.nanoTime(); + GenericApiGatewayResponse response = null; + GenericApiGatewayException exception = null; + try { + response = client.execute(request); + logResponse(logger, response); + } catch (GenericApiGatewayException gag) { + // ERROR response codes may come back as exceptions, 404 for example + exception = gag; + } + + final int statusCode; + if (exception != null) { + statusCode = exception.getStatusCode(); + } else { + statusCode = response.getHttpResponse().getStatusCode(); + } + + if (statusCode == 0) { + throw new IllegalStateException( + "Status code unknown, connection hasn't been attempted."); + } + final String endpoint = context.getProperty(PROP_AWS_GATEWAY_API_ENDPOINT).getValue(); + boolean outputRegardless = context.getProperty(PROP_OUTPUT_RESPONSE_REGARDLESS) + .asBoolean(); + + boolean outputBodyToResponseContent = (isSuccess(statusCode) && !putToAttribute + || outputRegardless); + boolean outputBodyToRequestAttribute = + (!isSuccess(statusCode) || putToAttribute) && requestFlowFile != null; + boolean bodyExists = response != null && response.getBody() != null; + + final String statusExplanation; + if (exception != null) { + statusExplanation = EnglishReasonPhraseCatalog.INSTANCE.getReason(statusCode, null); + } else { + statusExplanation = response.getHttpResponse().getStatusText(); + } + + // Create a map of the status attributes that are always written to the request and response FlowFiles + final Map<String, String> statusAttributes = new HashMap<>(); + statusAttributes.put(STATUS_CODE, String.valueOf(statusCode)); + statusAttributes.put(STATUS_MESSAGE, statusExplanation); + statusAttributes.put(ENDPOINT_ATTR, client.getEndpointPrefix()); + statusAttributes.put(RESOURCE_NAME_ATTR, resourceName); + statusAttributes.put(TRANSACTION_ID, txId.toString()); + + if (outputBodyToResponseContent) { + /* + * If successful and putting to response flowfile, store the response body as the flowfile payload + * we include additional flowfile attributes including the response headers and the status codes. + */ + + // clone the flowfile to capture the response + if (requestFlowFile != null) { + responseFlowFile = session.create(requestFlowFile); + // write attributes to request flowfile + requestFlowFile = session.putAllAttributes(requestFlowFile, statusAttributes); + // If the property to add the response headers to the request flowfile is true then add them + if (context.getProperty(PROP_ADD_HEADERS_TO_REQUEST).asBoolean()) { + // write the response headers as attributes + // this will overwrite any existing flowfile attributes + requestFlowFile = session.putAllAttributes(requestFlowFile, + convertAttributesFromHeaders( + response)); + } + } else { + responseFlowFile = session.create(); + } + + // write attributes to response flowfile + responseFlowFile = session.putAllAttributes(responseFlowFile, statusAttributes); + + // write the response headers as attributes + // this will overwrite any existing flowfile attributes + if (response != null) { + responseFlowFile = session + .putAllAttributes(responseFlowFile, convertAttributesFromHeaders(response)); + } else { + responseFlowFile = session + .putAllAttributes(responseFlowFile, exception.getHttpHeaders()); + } + // transfer the message body to the payload + // can potentially be null in edge cases + if (bodyExists) { + final String contentType = response.getHttpResponse().getHeaders() + .get("Content-Type"); + if (!(contentType == null) && !contentType.trim().isEmpty()) { + responseFlowFile = session + .putAttribute(responseFlowFile, CoreAttributes.MIME_TYPE.key(), + contentType.trim()); + } + + responseFlowFile = session + .importFrom(new ByteArrayInputStream(response.getBody().getBytes()), + responseFlowFile); + + // emit provenance event + final long millis = TimeUnit.NANOSECONDS + .toMillis(System.nanoTime() - startNanos); + if (requestFlowFile != null) { + session.getProvenanceReporter().fetch(responseFlowFile, endpoint, millis); + } else { + session.getProvenanceReporter().receive(responseFlowFile, endpoint, millis); + } + } else if (exception != null) { + final String contentType = "application/json"; + responseFlowFile = session + .putAttribute(responseFlowFile, CoreAttributes.MIME_TYPE.key(), + contentType.trim()); + + responseFlowFile = session + .importFrom(new ByteArrayInputStream(exception.getRawResponse()), + responseFlowFile); + + // emit provenance event + final long millis = TimeUnit.NANOSECONDS + .toMillis(System.nanoTime() - startNanos); + if (requestFlowFile != null) { + session.getProvenanceReporter().fetch(responseFlowFile, endpoint, millis); + } else { + session.getProvenanceReporter().receive(responseFlowFile, endpoint, millis); + } + } + } + // if not successful and request flowfile is not null, store the response body into a flowfile attribute + if (outputBodyToRequestAttribute) { + String attributeKey = context.getProperty(PROP_PUT_OUTPUT_IN_ATTRIBUTE) + .evaluateAttributeExpressions(requestFlowFile) + .getValue(); + if (attributeKey == null) { + attributeKey = RESPONSE_BODY; + } + byte[] outputBuffer; + int size = 0; + outputBuffer = new byte[maxAttributeSize]; + if (bodyExists) { + size = StreamUtils + .fillBuffer(new ByteArrayInputStream(response.getBody().getBytes()), + outputBuffer, false); + } else if (exception != null && exception.getRawResponse() != null + && exception.getRawResponse().length > 0) { + size = StreamUtils + .fillBuffer(new ByteArrayInputStream(exception.getRawResponse()), + outputBuffer, false); + } + + if (size > 0) { + String bodyString = new String(outputBuffer, 0, size, "UTF-8"); + requestFlowFile = session + .putAttribute(requestFlowFile, attributeKey, bodyString); + } + + requestFlowFile = session.putAllAttributes(requestFlowFile, statusAttributes); + + final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + session.getProvenanceReporter().modifyAttributes(requestFlowFile, + "The " + attributeKey + + " has been added. The value of which is the body of a http call to " + + endpoint + resourceName + + ". It took " + millis + + "millis,"); + } + + route(requestFlowFile, responseFlowFile, session, context, statusCode, + getRelationships()); + } catch (Exception e) { + // penalize or yield + if (requestFlowFile != null) { + logger.error("Routing to {} due to exception: {}", + new Object[]{REL_FAILURE.getName(), e}, e); + requestFlowFile = session.penalize(requestFlowFile); + requestFlowFile = session + .putAttribute(requestFlowFile, EXCEPTION_CLASS, e.getClass().getName()); + requestFlowFile = session + .putAttribute(requestFlowFile, EXCEPTION_MESSAGE, e.getMessage()); + // transfer original to failure + session.transfer(requestFlowFile, + getRelationshipForName(REL_FAILURE_NAME, getRelationships())); + } else { + logger.error( + "Yielding processor due to exception encountered as a source processor: {}", e); + context.yield(); + } + + // cleanup response flowfile, if applicable + try { + if (responseFlowFile != null) { + session.remove(responseFlowFile); + } + } catch (final Exception e1) { + logger.error("Could not cleanup response flowfile due to exception: {}", + new Object[]{e1}, e1); + } + } + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor index 27e39f0..c628fb1 100644 --- a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor @@ -27,4 +27,5 @@ org.apache.nifi.processors.aws.dynamodb.PutDynamoDB org.apache.nifi.processors.aws.dynamodb.DeleteDynamoDB org.apache.nifi.processors.aws.kinesis.stream.PutKinesisStream org.apache.nifi.processors.aws.cloudwatch.PutCloudWatchMetric +org.apache.nifi.processors.aws.wag.InvokeAWSGatewayApi http://git-wip-us.apache.org/repos/asf/nifi/blob/946c2fe6/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/wag/RequestMatcher.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/wag/RequestMatcher.java b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/wag/RequestMatcher.java new file mode 100644 index 0000000..5842419 --- /dev/null +++ b/nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/wag/RequestMatcher.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.processors.aws.wag; + +import java.util.Optional; +import java.util.function.Predicate; +import org.hamcrest.BaseMatcher; +import org.hamcrest.Description; + +public class RequestMatcher<T> extends BaseMatcher<T> { + + private final Predicate<T> matcher; + private final Optional<String> description; + + public RequestMatcher(Predicate<T> matcher) { + this(matcher, null); + } + + public RequestMatcher(Predicate<T> matcher, String description) { + this.matcher = matcher; + this.description = Optional.ofNullable(description); + } + + @SuppressWarnings("unchecked") + @Override + public boolean matches(Object argument) { + return matcher.test((T) argument); + } + + @Override + public void describeTo(Description description) { + this.description.ifPresent(description::appendText); + } +}
