/*
 * 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.standard;

import com.burgstaller.okhttp.AuthenticationCacheInterceptor;
import com.burgstaller.okhttp.CachingAuthenticatorDecorator;
import com.burgstaller.okhttp.digest.CachingAuthenticator;
import com.burgstaller.okhttp.digest.DigestAuthenticator;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.CookieManager;
import java.net.CookiePolicy;
import java.net.Proxy;
import java.net.Proxy.Type;
import java.net.URI;
import java.net.URL;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.security.Principal;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
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.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.annotation.Nullable;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.X509TrustManager;

import okhttp3.Cache;
import okhttp3.ConnectionPool;
import okhttp3.Credentials;
import okhttp3.Handshake;
import okhttp3.Headers;
import okhttp3.JavaNetCookieJar;
import okhttp3.MediaType;
import okhttp3.MultipartBody;
import okhttp3.MultipartBody.Builder;
import okhttp3.OkHttpClient;
import okhttp3.Protocol;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.BufferedSink;
import okio.GzipSink;
import okio.Okio;
import okio.Source;
import org.apache.commons.io.input.TeeInputStream;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.behavior.DynamicProperties;
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.SupportsSensitiveDynamicProperties;
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.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.deprecation.log.DeprecationLogger;
import org.apache.nifi.deprecation.log.DeprecationLoggerFactory;
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.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.DataUnit;
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.processor.util.StandardValidators;
import org.apache.nifi.processors.standard.http.ContentEncodingStrategy;
import org.apache.nifi.processors.standard.http.FlowFileNamingStrategy;
import org.apache.nifi.processors.standard.http.CookieStrategy;
import org.apache.nifi.processors.standard.http.HttpHeader;
import org.apache.nifi.processors.standard.http.HttpMethod;
import org.apache.nifi.processors.standard.util.ProxyAuthenticator;
import org.apache.nifi.processors.standard.util.SoftLimitBoundedByteArrayOutputStream;
import org.apache.nifi.proxy.ProxyConfiguration;
import org.apache.nifi.proxy.ProxySpec;
import org.apache.nifi.security.util.SslContextFactory;
import org.apache.nifi.security.util.TlsConfiguration;
import org.apache.nifi.security.util.TlsException;
import org.apache.nifi.ssl.SSLContextService;
import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.UriUtils;

import static org.apache.commons.lang3.StringUtils.trimToEmpty;

@SupportsSensitiveDynamicProperties
@SupportsBatching
@Tags({"http", "https", "rest", "client"})
@InputRequirement(Requirement.INPUT_ALLOWED)
@CapabilityDescription("An HTTP client processor which can interact with a configurable HTTP Endpoint. The destination URL and HTTP Method are configurable."
        + " When the HTTP Method is PUT, POST or PATCH, the FlowFile contents are included as the body of the request and FlowFile attributes are converted"
        + " to HTTP headers, optionally, based on configuration properties.")
@WritesAttributes({
        @WritesAttribute(attribute = InvokeHTTP.STATUS_CODE, description = "The status code that is returned"),
        @WritesAttribute(attribute = InvokeHTTP.STATUS_MESSAGE, description = "The status message that is returned"),
        @WritesAttribute(attribute = InvokeHTTP.RESPONSE_BODY, description = "In the instance where the status code received is not a success (2xx) "
                + "then the response body will be put to the 'invokehttp.response.body' attribute of the request FlowFile."),
        @WritesAttribute(attribute = InvokeHTTP.REQUEST_URL, description = "The original request URL"),
        @WritesAttribute(attribute = InvokeHTTP.REQUEST_DURATION, description = "Duration (in milliseconds) of the HTTP call to the external endpoint"),
        @WritesAttribute(attribute = InvokeHTTP.RESPONSE_URL, description = "The URL that was ultimately requested after any redirects were followed"),
        @WritesAttribute(attribute = InvokeHTTP.TRANSACTION_ID, description = "The transaction ID that is returned after reading the response"),
        @WritesAttribute(attribute = InvokeHTTP.REMOTE_DN, description = "The DN of the remote server"),
        @WritesAttribute(attribute = InvokeHTTP.EXCEPTION_CLASS, description = "The Java exception class raised when the processor fails"),
        @WritesAttribute(attribute = InvokeHTTP.EXCEPTION_MESSAGE, description = "The Java exception message raised when the processor fails"),
        @WritesAttribute(attribute = "user-defined", description = "If the 'Put Response Body In Attribute' property is set then whatever it is set to "
                + "will become the attribute key and the value would be the body of the HTTP response.")})
@DynamicProperties({
        @DynamicProperty(name = "Header Name", value = "Attribute Expression Language", expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
                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."),
        @DynamicProperty(name = "post:form:<NAME>", value = "Attribute Expression Language", expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
                description =
                        "When the HTTP Method is POST, dynamic properties with the property name in the form of post:form:<NAME>,"
                                + " where the <NAME> will be the form data name, will be used to fill out the multipart form parts."
                                + "  If send message body is false, the flowfile will not be sent, but any other form data will be.")
})
public class InvokeHTTP extends AbstractProcessor {
    public final static String STATUS_CODE = "invokehttp.status.code";
    public final static String STATUS_MESSAGE = "invokehttp.status.message";
    public final static String RESPONSE_BODY = "invokehttp.response.body";
    public final static String REQUEST_URL = "invokehttp.request.url";
    public final static String REQUEST_DURATION = "invokehttp.request.duration";
    public final static String RESPONSE_URL = "invokehttp.response.url";
    public final static String TRANSACTION_ID = "invokehttp.tx.id";
    public final static String REMOTE_DN = "invokehttp.remote.dn";
    public final static String EXCEPTION_CLASS = "invokehttp.java.exception.class";
    public final static String EXCEPTION_MESSAGE = "invokehttp.java.exception.message";

    public static final String DEFAULT_CONTENT_TYPE = "application/octet-stream";

    protected static final String FORM_DATA_NAME_BASE = "post:form";
    private static final Pattern FORM_DATA_NAME_PARAMETER_PATTERN = Pattern.compile("post:form:(?<formDataName>.*)$");
    private static final String FORM_DATA_NAME_GROUP = "formDataName";

    private static final Set<String> IGNORED_REQUEST_ATTRIBUTES = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
            STATUS_CODE,
            STATUS_MESSAGE,
            RESPONSE_BODY,
            REQUEST_URL,
            RESPONSE_URL,
            TRANSACTION_ID,
            REMOTE_DN,
            EXCEPTION_CLASS,
            EXCEPTION_MESSAGE,
            CoreAttributes.UUID.key(),
            CoreAttributes.PATH.key()
    )));

    public static final PropertyDescriptor HTTP_METHOD = new PropertyDescriptor.Builder()
            .name("HTTP Method")
            .description("HTTP request method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS). Arbitrary methods are also supported. "
                    + "Methods other than POST, PUT and PATCH will be sent without a message body.")
            .required(true)
            .defaultValue(HttpMethod.GET.name())
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING))
            .build();

    public static final PropertyDescriptor HTTP_URL = new PropertyDescriptor.Builder()
            .name("Remote URL")
            .displayName("HTTP URL")
            .description("HTTP remote URL including a scheme of http or https, as well as a hostname or IP address with optional port and path elements." +
                    " Any encoding of the URL must be done by the user.")
            .required(true)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .addValidator(StandardValidators.URL_VALIDATOR)
            .build();

    public static final PropertyDescriptor HTTP2_DISABLED = new PropertyDescriptor.Builder()
            .name("disable-http2")
            .displayName("HTTP/2 Disabled")
            .description("Disable negotiation of HTTP/2 protocol. HTTP/2 requires TLS. HTTP/1.1 protocol supported is required when HTTP/2 is disabled.")
            .required(true)
            .defaultValue("False")
            .allowableValues("True", "False")
            .build();

    public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()
            .name("SSL Context Service")
            .description("SSL Context Service provides trusted certificates and client certificates for TLS communication.")
            .required(false)
            .identifiesControllerService(SSLContextService.class)
            .build();

    public static final PropertyDescriptor SOCKET_CONNECT_TIMEOUT = new PropertyDescriptor.Builder()
            .name("Connection Timeout")
            .displayName("Socket Connect Timeout")
            .description("Maximum time to wait for initial socket connection to the HTTP URL.")
            .required(true)
            .defaultValue("5 secs")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .build();

    public static final PropertyDescriptor SOCKET_READ_TIMEOUT = new PropertyDescriptor.Builder()
            .name("Read Timeout")
            .displayName("Socket Read Timeout")
            .description("Maximum time to wait for receiving responses from a socket connection to the HTTP URL.")
            .required(true)
            .defaultValue("15 secs")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .build();

    public static final PropertyDescriptor SOCKET_WRITE_TIMEOUT = new PropertyDescriptor.Builder()
            .name("Socket Write Timeout")
            .displayName("Socket Write Timeout")
            .description("Maximum time to wait for write operations while sending requests from a socket connection to the HTTP URL.")
            .required(true)
            .defaultValue("15 secs")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .build();

    public static final PropertyDescriptor SOCKET_IDLE_TIMEOUT = new PropertyDescriptor.Builder()
            .name("idle-timeout")
            .displayName("Socket Idle Timeout")
            .description("Maximum time to wait before closing idle connections to the HTTP URL.")
            .required(true)
            .defaultValue("5 mins")
            .addValidator(StandardValidators.createTimePeriodValidator(1, TimeUnit.MILLISECONDS, Integer.MAX_VALUE, TimeUnit.SECONDS))
            .build();

    public static final PropertyDescriptor SOCKET_IDLE_CONNECTIONS = new PropertyDescriptor.Builder()
            .name("max-idle-connections")
            .displayName("Socket Idle Connections")
            .description("Maximum number of idle connections to the HTTP URL.")
            .required(true)
            .defaultValue("5")
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .build();

    @Deprecated
    public static final PropertyDescriptor PROXY_HOST = new PropertyDescriptor.Builder()
            .name("Proxy Host")
            .description("Proxy Host and dependent properties are deprecated in favor of Proxy Configuration Service. Proxy Host can be configured using an IP address or DNS address.")
            .required(false)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    @Deprecated
    public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder()
            .name("Proxy Port")
            .description("Proxy Port and dependent properties are deprecated in favor of Proxy Configuration Service. Port number for the configured Proxy Host address.")
            .required(false)
            .addValidator(StandardValidators.PORT_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .dependsOn(PROXY_HOST)
            .build();

    @Deprecated
    public static final PropertyDescriptor PROXY_TYPE = new PropertyDescriptor.Builder()
            .name("Proxy Type")
            .displayName("Proxy Type")
            .description("Proxy Type and dependent properties are deprecated in favor of Proxy Configuration Service. Proxy protocol type is not used")
            .defaultValue("http")
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
            .dependsOn(PROXY_HOST)
            .build();

    @Deprecated
    public static final PropertyDescriptor PROXY_USERNAME = new PropertyDescriptor.Builder()
            .name("invokehttp-proxy-user")
            .displayName("Proxy Username")
            .description("Proxy Username and dependent properties are deprecated in favor of Proxy Configuration Service. Username to set when authenticating with a Proxy server.")
            .required(false)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .dependsOn(PROXY_HOST)
            .build();

    @Deprecated
    public static final PropertyDescriptor PROXY_PASSWORD = new PropertyDescriptor.Builder()
            .name("invokehttp-proxy-password")
            .displayName("Proxy Password")
            .description("Proxy Password and dependent properties are deprecated in favor of Proxy Configuration Service. Password to set when authenticating with a Proxy server.")
            .required(false)
            .sensitive(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .dependsOn(PROXY_HOST)
            .build();

    public static final PropertyDescriptor REQUEST_OAUTH2_ACCESS_TOKEN_PROVIDER = new PropertyDescriptor.Builder()
            .name("oauth2-access-token-provider")
            .displayName("Request OAuth2 Access Token Provider")
            .description("Enables managed retrieval of OAuth2 Bearer Token applied to HTTP requests using the Authorization Header.")
            .identifiesControllerService(OAuth2AccessTokenProvider.class)
            .required(false)
            .build();

    public static final PropertyDescriptor REQUEST_USERNAME = new PropertyDescriptor.Builder()
            .name("Basic Authentication Username")
            .displayName("Request Username")
            .description("The username provided for authentication of HTTP requests. Encoded using Base64 for HTTP Basic Authentication as described in RFC 7617.")
            .required(false)
            .addValidator(StandardValidators.createRegexMatchingValidator(Pattern.compile("^[\\x20-\\x39\\x3b-\\x7e\\x80-\\xff]+$")))
            .build();

    public static final PropertyDescriptor REQUEST_PASSWORD = new PropertyDescriptor.Builder()
            .name("Basic Authentication Password")
            .displayName("Request Password")
            .description("The password provided for authentication of HTTP requests. Encoded using Base64 for HTTP Basic Authentication as described in RFC 7617.")
            .required(false)
            .sensitive(true)
            .addValidator(StandardValidators.createRegexMatchingValidator(Pattern.compile("^[\\x20-\\x7e\\x80-\\xff]+$")))
            .build();

    public static final PropertyDescriptor REQUEST_DIGEST_AUTHENTICATION_ENABLED = new PropertyDescriptor.Builder()
            .name("Digest Authentication")
            .displayName("Request Digest Authentication Enabled")
            .description("Enable Digest Authentication on HTTP requests with Username and Password credentials as described in RFC 7616.")
            .required(false)
            .defaultValue("false")
            .allowableValues("true", "false")
            .dependsOn(REQUEST_USERNAME)
            .build();

    public static final PropertyDescriptor REQUEST_FAILURE_PENALIZATION_ENABLED = new PropertyDescriptor.Builder()
            .name("Penalize on \"No Retry\"")
            .displayName("Request Failure Penalization Enabled")
            .description("Enable penalization of request FlowFiles when receiving HTTP response with a status code between 400 and 499.")
            .required(false)
            .defaultValue(Boolean.FALSE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .build();

    public static final PropertyDescriptor REQUEST_BODY_ENABLED = new PropertyDescriptor.Builder()
            .name("send-message-body")
            .displayName("Request Body Enabled")
            .description("Enable sending HTTP request body for PATCH, POST, or PUT methods.")
            .defaultValue(Boolean.TRUE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .required(false)
            .dependsOn(HTTP_METHOD, HttpMethod.PATCH.name(), HttpMethod.POST.name(), HttpMethod.PUT.name())
            .build();

    public static final PropertyDescriptor REQUEST_FORM_DATA_NAME = new PropertyDescriptor.Builder()
            .name("form-body-form-name")
            .displayName("Request Multipart Form-Data Name")
            .description("Enable sending HTTP request body formatted using multipart/form-data and using the form name configured.")
            .required(false)
            .addValidator(
                    StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING, true)
            )
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .dependsOn(REQUEST_BODY_ENABLED, Boolean.TRUE.toString())
            .build();

    public static final PropertyDescriptor REQUEST_FORM_DATA_FILENAME_ENABLED = new PropertyDescriptor.Builder()
            .name("set-form-filename")
            .displayName("Request Multipart Form-Data Filename Enabled")
            .description("Enable sending the FlowFile filename attribute as the filename parameter in the Content-Disposition Header for multipart/form-data HTTP requests.")
            .required(false)
            .defaultValue(Boolean.TRUE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .dependsOn(REQUEST_FORM_DATA_NAME)
            .build();

    public static final PropertyDescriptor REQUEST_CHUNKED_TRANSFER_ENCODING_ENABLED = new PropertyDescriptor.Builder()
            .name("Use Chunked Encoding")
            .displayName("Request Chunked Transfer-Encoding Enabled")
            .description("Enable sending HTTP requests with the Transfer-Encoding Header set to chunked, and disable sending the Content-Length Header. " +
                    "Transfer-Encoding applies to the body in HTTP/1.1 requests as described in RFC 7230 Section 3.3.1")
            .required(true)
            .defaultValue(Boolean.FALSE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .dependsOn(HTTP_METHOD, HttpMethod.PATCH.name(), HttpMethod.POST.name(), HttpMethod.PUT.name())
            .build();

    public static final PropertyDescriptor REQUEST_CONTENT_ENCODING = new PropertyDescriptor.Builder()
            .name("Content-Encoding")
            .displayName("Request Content-Encoding")
            .description("HTTP Content-Encoding applied to request body during transmission. The receiving server must support the selected encoding to avoid request failures.")
            .required(true)
            .defaultValue(ContentEncodingStrategy.DISABLED.getValue())
            .allowableValues(ContentEncodingStrategy.class)
            .dependsOn(HTTP_METHOD, HttpMethod.PATCH.name(), HttpMethod.POST.name(), HttpMethod.PUT.name())
            .build();

    public static final PropertyDescriptor REQUEST_CONTENT_TYPE = new PropertyDescriptor.Builder()
            .name("Content-Type")
            .displayName("Request Content-Type")
            .description("HTTP Content-Type Header applied to when sending an HTTP request body for PATCH, POST, or PUT methods. " +
                    String.format("The Content-Type defaults to %s when not configured.", DEFAULT_CONTENT_TYPE))
            .required(true)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .defaultValue("${" + CoreAttributes.MIME_TYPE.key() + "}")
            .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING))
            .dependsOn(HTTP_METHOD, HttpMethod.PATCH.name(), HttpMethod.POST.name(), HttpMethod.PUT.name())
            .build();

    public static final PropertyDescriptor REQUEST_DATE_HEADER_ENABLED = new PropertyDescriptor.Builder()
            .name("Include Date Header")
            .displayName("Request Date Header Enabled")
            .description("Enable sending HTTP Date Header on HTTP requests as described in RFC 7231 Section 7.1.1.2.")
            .required(true)
            .defaultValue("True")
            .allowableValues("True", "False")
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .build();

    public static final PropertyDescriptor REQUEST_HEADER_ATTRIBUTES_PATTERN = new PropertyDescriptor.Builder()
            .name("Attributes to Send")
            .displayName("Request Header Attributes Pattern")
            .description("Regular expression that defines which FlowFile attributes to send as HTTP headers in the request. "
                    + "If not defined, no attributes are sent as headers. Dynamic properties will be always be sent as headers. "
                    + "The dynamic property name will be the header key and the dynamic property value, interpreted as Expression "
                    + "Language, will be the header value. Attributes and their values are limited to ASCII characters due to "
                    + "the requirement of the HTTP protocol.")
            .required(false)
            .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
            .build();

    public static final PropertyDescriptor REQUEST_USER_AGENT = new PropertyDescriptor.Builder()
            .name("Useragent")
            .displayName("Request User-Agent")
            .description("HTTP User-Agent Header applied to requests. RFC 7231 Section 5.5.3 describes recommend formatting.")
            .required(false)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor RESPONSE_BODY_ATTRIBUTE_NAME = new PropertyDescriptor.Builder()
            .name("Put Response Body In Attribute")
            .displayName("Response Body Attribute Name")
            .description("FlowFile attribute name used to write an HTTP response body for FlowFiles transferred to the Original relationship.")
            .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING))
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .build();

    public static final PropertyDescriptor RESPONSE_BODY_ATTRIBUTE_SIZE = new PropertyDescriptor.Builder()
            .name("Max Length To Put In Attribute")
            .displayName("Response Body Attribute Size")
            .description("Maximum size in bytes applied when writing an HTTP response body to a FlowFile attribute. Attributes exceeding the maximum will be truncated.")
            .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
            .defaultValue("256")
            .dependsOn(RESPONSE_BODY_ATTRIBUTE_NAME)
            .build();

    public static final PropertyDescriptor RESPONSE_BODY_IGNORED = new PropertyDescriptor.Builder()
            .name("ignore-response-content")
            .displayName("Response Body Ignored")
            .description("Disable writing HTTP response FlowFiles to Response relationship")
            .required(true)
            .defaultValue(Boolean.FALSE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .build();

    public static final PropertyDescriptor RESPONSE_CACHE_ENABLED = new PropertyDescriptor.Builder()
            .name("use-etag")
            .displayName("Response Cache Enabled")
            .description("Enable HTTP response caching described in RFC 7234. Caching responses considers ETag and other headers.")
            .required(true)
            .defaultValue(Boolean.FALSE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .build();

    public static final PropertyDescriptor RESPONSE_CACHE_SIZE = new PropertyDescriptor.Builder()
            .name("etag-max-cache-size")
            .displayName("Response Cache Size")
            .description("Maximum size of HTTP response cache in bytes. Caching responses considers ETag and other headers.")
            .required(true)
            .defaultValue("10MB")
            .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
            .dependsOn(RESPONSE_CACHE_ENABLED, Boolean.TRUE.toString())
            .build();

    public static final PropertyDescriptor RESPONSE_COOKIE_STRATEGY = new PropertyDescriptor.Builder()
            .name("cookie-strategy")
            .description("Strategy for accepting and persisting HTTP cookies. Accepting cookies enables persistence across multiple requests.")
            .displayName("Response Cookie Strategy")
            .required(true)
            .defaultValue(CookieStrategy.DISABLED.name())
            .allowableValues(CookieStrategy.values())
            .build();

    public static final PropertyDescriptor RESPONSE_GENERATION_REQUIRED = new PropertyDescriptor.Builder()
            .name("Always Output Response")
            .displayName("Response Generation Required")
            .description("Enable generation and transfer of a FlowFile to the Response relationship regardless of HTTP response received.")
            .required(false)
            .defaultValue(Boolean.FALSE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .build();

    public static final PropertyDescriptor RESPONSE_FLOW_FILE_NAMING_STRATEGY = new PropertyDescriptor.Builder()
            .name("flow-file-naming-strategy")
            .description("Determines the strategy used for setting the filename attribute of FlowFiles transferred to the Response relationship.")
            .displayName("Response FlowFile Naming Strategy")
            .required(true)
            .defaultValue(FlowFileNamingStrategy.RANDOM.name())
            .allowableValues(
                    Arrays.stream(FlowFileNamingStrategy.values()).map(strategy ->
                            new AllowableValue(strategy.name(), strategy.name(), strategy.getDescription())
                    ).toArray(AllowableValue[]::new)
            )
            .build();

    public static final PropertyDescriptor RESPONSE_HEADER_REQUEST_ATTRIBUTES_ENABLED = new PropertyDescriptor.Builder()
            .name("Add Response Headers to Request")
            .displayName("Response Header Request Attributes Enabled")
            .description("Enable adding HTTP response headers as attributes to FlowFiles transferred to the Original relationship.")
            .required(false)
            .defaultValue(Boolean.FALSE.toString())
            .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
            .build();

    public static final PropertyDescriptor RESPONSE_REDIRECTS_ENABLED = new PropertyDescriptor.Builder()
            .name("Follow Redirects")
            .displayName("Response Redirects Enabled")
            .description("Enable following HTTP redirects sent with HTTP 300 series responses as described in RFC 7231 Section 6.4.")
            .required(true)
            .defaultValue("True")
            .allowableValues("True", "False")
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .build();

    private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH, ProxySpec.SOCKS};

    private static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE = ProxyConfiguration.createProxyConfigPropertyDescriptor(true, PROXY_SPECS);

    public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
            HTTP_METHOD,
            HTTP_URL,
            HTTP2_DISABLED,
            SSL_CONTEXT_SERVICE,
            SOCKET_CONNECT_TIMEOUT,
            SOCKET_READ_TIMEOUT,
            SOCKET_WRITE_TIMEOUT,
            SOCKET_IDLE_TIMEOUT,
            SOCKET_IDLE_CONNECTIONS,
            PROXY_CONFIGURATION_SERVICE,
            PROXY_HOST,
            PROXY_PORT,
            PROXY_TYPE,
            PROXY_USERNAME,
            PROXY_PASSWORD,
            REQUEST_OAUTH2_ACCESS_TOKEN_PROVIDER,
            REQUEST_USERNAME,
            REQUEST_PASSWORD,
            REQUEST_DIGEST_AUTHENTICATION_ENABLED,
            REQUEST_FAILURE_PENALIZATION_ENABLED,
            REQUEST_BODY_ENABLED,
            REQUEST_FORM_DATA_NAME,
            REQUEST_FORM_DATA_FILENAME_ENABLED,
            REQUEST_CHUNKED_TRANSFER_ENCODING_ENABLED,
            REQUEST_CONTENT_ENCODING,
            REQUEST_CONTENT_TYPE,
            REQUEST_DATE_HEADER_ENABLED,
            REQUEST_HEADER_ATTRIBUTES_PATTERN,
            REQUEST_USER_AGENT,
            RESPONSE_BODY_ATTRIBUTE_NAME,
            RESPONSE_BODY_ATTRIBUTE_SIZE,
            RESPONSE_BODY_IGNORED,
            RESPONSE_CACHE_ENABLED,
            RESPONSE_CACHE_SIZE,
            RESPONSE_COOKIE_STRATEGY,
            RESPONSE_GENERATION_REQUIRED,
            RESPONSE_FLOW_FILE_NAMING_STRATEGY,
            RESPONSE_HEADER_REQUEST_ATTRIBUTES_ENABLED,
            RESPONSE_REDIRECTS_ENABLED
    ));

    public static final Relationship ORIGINAL = new Relationship.Builder()
            .name("Original")
            .description("Request FlowFiles transferred when receiving HTTP responses with a status code between 200 and 299.")
            .build();

    public static final Relationship RESPONSE = new Relationship.Builder()
            .name("Response")
            .description("Response FlowFiles transferred when receiving HTTP responses with a status code between 200 and 299.")
            .build();

    public static final Relationship RETRY = new Relationship.Builder()
            .name("Retry")
            .description("Request FlowFiles transferred when receiving HTTP responses with a status code between 500 and 599.")
            .build();

    public static final Relationship NO_RETRY = new Relationship.Builder()
            .name("No Retry")
            .description("Request FlowFiles transferred when receiving HTTP responses with a status code between 400 an 499.")
            .build();

    public static final Relationship FAILURE = new Relationship.Builder()
            .name("Failure")
            .description("Request FlowFiles transferred when receiving socket communication errors.")
            .build();

    public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
            ORIGINAL,
            RESPONSE,
            RETRY,
            NO_RETRY,
            FAILURE
    )));

    // RFC 2616 Date Time Formatter with hard-coded GMT Zone and US Locale. RFC 2616 Section 3.3 indicates the header should not be localized
    private static final DateTimeFormatter RFC_2616_DATE_TIME = DateTimeFormatter.ofPattern("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.US);

    private static final String MULTIPLE_HEADER_DELIMITER = ", ";

    private static final DeprecationLogger deprecationLogger = DeprecationLoggerFactory.getLogger(InvokeHTTP.class);

    private volatile Set<String> dynamicPropertyNames = new HashSet<>();

    private volatile Pattern requestHeaderAttributesPattern = null;

    private volatile boolean chunkedTransferEncoding = false;

    private volatile Optional<OAuth2AccessTokenProvider> oauth2AccessTokenProviderOptional;

    private final AtomicReference<OkHttpClient> okHttpClientAtomicReference = new AtomicReference<>();

    @Override
    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return PROPERTIES;
    }

    @Override
    protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
        if (propertyDescriptorName.startsWith(FORM_DATA_NAME_BASE)) {

            Matcher matcher = FORM_DATA_NAME_PARAMETER_PATTERN.matcher(propertyDescriptorName);
            if (matcher.matches()) {
                return new PropertyDescriptor.Builder()
                        .required(false)
                        .name(propertyDescriptorName)
                        .description("Form Data " + matcher.group(FORM_DATA_NAME_GROUP))
                        .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING, true))
                        .dynamic(true)
                        .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
                        .build();
            }
            return null;
        }

        return new PropertyDescriptor.Builder()
                .required(false)
                .name(propertyDescriptorName)
                .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING, true))
                .dynamic(true)
                .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
                .build();
    }

    @Override
    public Set<Relationship> getRelationships() {
        return RELATIONSHIPS;
    }

    @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 (REQUEST_HEADER_ATTRIBUTES_PATTERN.getName().equalsIgnoreCase(descriptor.getName())) {
                if (newValue == null || newValue.isEmpty()) {
                    requestHeaderAttributesPattern = null;
                } else {
                    final String trimmedValue = StringUtils.trimToEmpty(newValue);
                    requestHeaderAttributesPattern = Pattern.compile(trimmedValue);
                }
            }
        }
    }

    @Override
    protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
        final List<ValidationResult> results = new ArrayList<>(3);
        final boolean proxyHostSet = validationContext.getProperty(PROXY_HOST).isSet();
        if (proxyHostSet) {
            deprecationLogger.warn("{}[id={}] [{}] Property should be replaced with [{}] Property",
                    getClass().getSimpleName(),
                    getIdentifier(),
                    PROXY_HOST.getDisplayName(),
                    PROXY_CONFIGURATION_SERVICE.getDisplayName()
            );
        }

        final boolean proxyPortSet = validationContext.getProperty(PROXY_PORT).isSet();

        if ((proxyHostSet && !proxyPortSet) || (!proxyHostSet && proxyPortSet)) {
            results.add(new ValidationResult.Builder().subject("Proxy Host and Port").valid(false).explanation("If Proxy Host or Proxy Port is set, both must be set").build());
        }

        final boolean proxyUserSet = validationContext.getProperty(PROXY_USERNAME).isSet();
        final boolean proxyPwdSet = validationContext.getProperty(PROXY_PASSWORD).isSet();

        if ((proxyUserSet && !proxyPwdSet) || (!proxyUserSet && proxyPwdSet)) {
            results.add(new ValidationResult.Builder().subject("Proxy User and Password").valid(false).explanation("If Proxy Username or Proxy Password is set, both must be set").build());
        }
        if (proxyUserSet && !proxyHostSet) {
            results.add(new ValidationResult.Builder().subject("Proxy").valid(false).explanation("If Proxy username is set, proxy host must be set").build());
        }

        ProxyConfiguration.validateProxySpec(validationContext, results, PROXY_SPECS);

        // Check for dynamic properties for form components.
        // Even if the flowfile is not sent, we may still send form parameters.
        boolean hasFormData = false;
        for (final PropertyDescriptor descriptor : validationContext.getProperties().keySet()) {
            final Matcher matcher = FORM_DATA_NAME_PARAMETER_PATTERN.matcher(descriptor.getName());
            if (matcher.matches()) {
                hasFormData = true;
                break;
            }
        }

        // if form data exists, and send body is true, Flowfile Form Data Name must be set.
        final boolean requestBodyEnabled = validationContext.getProperty(REQUEST_BODY_ENABLED).asBoolean();
        final boolean contentNameSet = validationContext.getProperty(REQUEST_FORM_DATA_NAME).isSet();
        if (hasFormData) {
            if (requestBodyEnabled && !contentNameSet) {
                final String explanation = String.format("[%s] is required when Form Data properties are configured and [%s] is enabled",
                        REQUEST_FORM_DATA_NAME.getDisplayName(),
                        REQUEST_BODY_ENABLED.getDisplayName());
                results.add(new ValidationResult.Builder()
                        .subject(REQUEST_FORM_DATA_NAME.getDisplayName())
                        .valid(false)
                        .explanation(explanation)
                        .build());
            }
        }
        if (!requestBodyEnabled && contentNameSet) {
            final String explanation = String.format("[%s] must be [true] when Form Data properties are configured and [%s] is configured",
                    REQUEST_BODY_ENABLED.getDisplayName(),
                    REQUEST_FORM_DATA_NAME.getDisplayName());
            results.add(new ValidationResult.Builder()
                    .subject(REQUEST_FORM_DATA_NAME.getDisplayName())
                    .valid(false)
                    .explanation(explanation)
                    .build());
        }

        boolean usingUserNamePasswordAuthorization = validationContext.getProperty(REQUEST_USERNAME).isSet()
            || validationContext.getProperty(REQUEST_PASSWORD).isSet();

        boolean usingOAuth2Authorization = validationContext.getProperty(REQUEST_OAUTH2_ACCESS_TOKEN_PROVIDER).isSet();

        if (usingUserNamePasswordAuthorization && usingOAuth2Authorization) {
            results.add(new ValidationResult.Builder()
                .subject("Authorization properties")
                .valid(false)
                .explanation("OAuth2 Authorization cannot be configured together with Username and Password properties")
                .build());
        }

        return results;
    }

    @OnScheduled
    public void setUpClient(final ProcessContext context) throws TlsException, IOException {
        okHttpClientAtomicReference.set(null);

        OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient().newBuilder();

        final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context, () -> {
            final ProxyConfiguration componentProxyConfig = new ProxyConfiguration();
            final String proxyHost = context.getProperty(PROXY_HOST).evaluateAttributeExpressions().getValue();
            final Integer proxyPort = context.getProperty(PROXY_PORT).evaluateAttributeExpressions().asInteger();
            if (proxyHost != null && proxyPort != null) {
                componentProxyConfig.setProxyType(Type.HTTP);
                componentProxyConfig.setProxyServerHost(proxyHost);
                componentProxyConfig.setProxyServerPort(proxyPort);
                final String proxyUsername = trimToEmpty(context.getProperty(PROXY_USERNAME).evaluateAttributeExpressions().getValue());
                final String proxyPassword = context.getProperty(PROXY_PASSWORD).evaluateAttributeExpressions().getValue();
                componentProxyConfig.setProxyUserName(proxyUsername);
                componentProxyConfig.setProxyUserPassword(proxyPassword);
            }
            return componentProxyConfig;
        });

        final Proxy proxy = proxyConfig.createProxy();
        if (!Type.DIRECT.equals(proxy.type())) {
            okHttpClientBuilder.proxy(proxy);
            if (proxyConfig.hasCredential()) {
                ProxyAuthenticator proxyAuthenticator = new ProxyAuthenticator(proxyConfig.getProxyUserName(), proxyConfig.getProxyUserPassword());
                okHttpClientBuilder.proxyAuthenticator(proxyAuthenticator);
            }
        }

        // Configure caching
        final boolean cachingEnabled = context.getProperty(RESPONSE_CACHE_ENABLED).asBoolean();
        if (cachingEnabled) {
            final int maxCacheSizeBytes = context.getProperty(RESPONSE_CACHE_SIZE).asDataSize(DataUnit.B).intValue();
            okHttpClientBuilder.cache(new Cache(getResponseCacheDirectory(), maxCacheSizeBytes));
        }

        if (context.getProperty(HTTP2_DISABLED).asBoolean()) {
            okHttpClientBuilder.protocols(Collections.singletonList(Protocol.HTTP_1_1));
        }

        okHttpClientBuilder.followRedirects(context.getProperty(RESPONSE_REDIRECTS_ENABLED).asBoolean());
        okHttpClientBuilder.connectTimeout((context.getProperty(SOCKET_CONNECT_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue()), TimeUnit.MILLISECONDS);
        okHttpClientBuilder.readTimeout(context.getProperty(SOCKET_READ_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(), TimeUnit.MILLISECONDS);
        okHttpClientBuilder.writeTimeout(context.getProperty(SOCKET_WRITE_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(), TimeUnit.MILLISECONDS);
        okHttpClientBuilder.connectionPool(
                new ConnectionPool(
                        context.getProperty(SOCKET_IDLE_CONNECTIONS).asInteger(),
                        context.getProperty(SOCKET_IDLE_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(), TimeUnit.MILLISECONDS
                )
        );

        final SSLContextService sslService = context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
        if (sslService != null) {
            final SSLContext sslContext = sslService.createContext();
            final SSLSocketFactory socketFactory = sslContext.getSocketFactory();
            final TlsConfiguration tlsConfiguration = sslService.createTlsConfiguration();
            final X509TrustManager trustManager = Objects.requireNonNull(SslContextFactory.getX509TrustManager(tlsConfiguration), "Trust Manager not found");
            okHttpClientBuilder.sslSocketFactory(socketFactory, trustManager);
        }

        final CookieStrategy cookieStrategy = CookieStrategy.valueOf(context.getProperty(RESPONSE_COOKIE_STRATEGY).getValue());
        switch (cookieStrategy) {
            case DISABLED:
                break;
            case ACCEPT_ALL:
                final CookieManager cookieManager = new CookieManager();
                cookieManager.setCookiePolicy(CookiePolicy.ACCEPT_ALL);
                okHttpClientBuilder.cookieJar(new JavaNetCookieJar(cookieManager));
                break;
        }

        setAuthenticator(okHttpClientBuilder, context);

        chunkedTransferEncoding = context.getProperty(REQUEST_CHUNKED_TRANSFER_ENCODING_ENABLED).asBoolean();

        okHttpClientAtomicReference.set(okHttpClientBuilder.build());
    }

    @OnScheduled
    public void initOauth2AccessTokenProvider(final ProcessContext context) {
        if (context.getProperty(REQUEST_OAUTH2_ACCESS_TOKEN_PROVIDER).isSet()) {
            OAuth2AccessTokenProvider oauth2AccessTokenProvider = context.getProperty(REQUEST_OAUTH2_ACCESS_TOKEN_PROVIDER).asControllerService(OAuth2AccessTokenProvider.class);

            oauth2AccessTokenProvider.getAccessDetails();

            oauth2AccessTokenProviderOptional = Optional.of(oauth2AccessTokenProvider);
        } else {
            oauth2AccessTokenProviderOptional = Optional.empty();
        }
    }

    private void setAuthenticator(OkHttpClient.Builder okHttpClientBuilder, ProcessContext context) {
        final String authUser = trimToEmpty(context.getProperty(REQUEST_USERNAME).getValue());

        // If the username/password properties are set then check if digest auth is being used
        if (!authUser.isEmpty() && "true".equalsIgnoreCase(context.getProperty(REQUEST_DIGEST_AUTHENTICATION_ENABLED).getValue())) {
            final String authPass = trimToEmpty(context.getProperty(REQUEST_PASSWORD).getValue());

            final Map<String, CachingAuthenticator> authCache = new ConcurrentHashMap<>();
            com.burgstaller.okhttp.digest.Credentials credentials = new com.burgstaller.okhttp.digest.Credentials(authUser, authPass);
            final DigestAuthenticator digestAuthenticator = new DigestAuthenticator(credentials);

            okHttpClientBuilder.interceptors().add(new AuthenticationCacheInterceptor(authCache));
            okHttpClientBuilder.authenticator(new CachingAuthenticatorDecorator(digestAuthenticator, authCache));
        }
    }

    @Override
    public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
        OkHttpClient okHttpClient = okHttpClientAtomicReference.get();

        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(RESPONSE_BODY_ATTRIBUTE_NAME).isSet();
        if (requestFlowFile == null) {
            if (context.hasNonLoopConnection()) {
                return;
            }

            final String method = getRequestMethod(context, null);
            final Optional<HttpMethod> requestMethodFound = findRequestMethod(method);
            final HttpMethod requestMethod = requestMethodFound.orElse(HttpMethod.GET);
            if (requestMethod.isRequestBodySupported()) {
                return;
            } else if (putToAttribute) {
                requestFlowFile = session.create();
            }
        }

        final int maxAttributeSize = context.getProperty(RESPONSE_BODY_ATTRIBUTE_SIZE).asInteger();
        final ComponentLog logger = getLogger();
        final UUID txId = UUID.randomUUID();

        FlowFile responseFlowFile = null;
        try {
            final String urlProperty = trimToEmpty(context.getProperty(HTTP_URL).evaluateAttributeExpressions(requestFlowFile).getValue());
            Request httpRequest = configureRequest(context, session, requestFlowFile, urlProperty);
            logRequest(logger, httpRequest);

            if (httpRequest.body() != null) {
                session.getProvenanceReporter().send(requestFlowFile, urlProperty, true);
            }

            final long startNanos = System.nanoTime();

            try (Response responseHttp = okHttpClient.newCall(httpRequest).execute()) {
                logResponse(logger, urlProperty, responseHttp);

                // store the status code and message
                int statusCode = responseHttp.code();
                String statusMessage = responseHttp.message();

                // Create a map of the status attributes that are always written to the request and response FlowFiles
                Map<String, String> statusAttributes = new HashMap<>();
                statusAttributes.put(STATUS_CODE, String.valueOf(statusCode));
                statusAttributes.put(STATUS_MESSAGE, statusMessage);
                statusAttributes.put(REQUEST_URL, urlProperty);
                statusAttributes.put(REQUEST_DURATION, Long.toString(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos)));
                statusAttributes.put(RESPONSE_URL, responseHttp.request().url().toString());
                statusAttributes.put(TRANSACTION_ID, txId.toString());

                if (requestFlowFile != null) {
                    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(RESPONSE_HEADER_REQUEST_ATTRIBUTES_ENABLED).asBoolean() && requestFlowFile != null) {
                    // write the response headers as attributes
                    // this will overwrite any existing flowfile attributes
                    requestFlowFile = session.putAllAttributes(requestFlowFile, convertAttributesFromHeaders(responseHttp));
                }

                boolean outputBodyToRequestAttribute = (!isSuccess(statusCode) || putToAttribute) && requestFlowFile != null;
                boolean outputBodyToResponseContent = (isSuccess(statusCode) && !putToAttribute) || context.getProperty(RESPONSE_GENERATION_REQUIRED).asBoolean();
                ResponseBody responseBody = responseHttp.body();
                boolean bodyExists = responseBody != null && !context.getProperty(RESPONSE_BODY_IGNORED).asBoolean();

                InputStream responseBodyStream = null;
                SoftLimitBoundedByteArrayOutputStream outputStreamToRequestAttribute = null;
                TeeInputStream teeInputStream = null;
                try {
                    responseBodyStream = bodyExists ? responseBody.byteStream() : null;
                    if (responseBodyStream != null && outputBodyToRequestAttribute && outputBodyToResponseContent) {
                        outputStreamToRequestAttribute = new SoftLimitBoundedByteArrayOutputStream(maxAttributeSize);
                        teeInputStream = new TeeInputStream(responseBodyStream, outputStreamToRequestAttribute);
                    }

                    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);
                        } 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
                        responseFlowFile = session.putAllAttributes(responseFlowFile, convertAttributesFromHeaders(responseHttp));

                        // update FlowFile's filename attribute with an extracted value from the remote URL
                        if (FlowFileNamingStrategy.URL_PATH.equals(getFlowFileNamingStrategy(context)) && HttpMethod.GET.name().equals(httpRequest.method())) {
                            final URI uri = UriUtils.create(urlProperty);
                            final URL url = uri.toURL();
                            String fileName = getFileNameFromUrl(url);
                            if (fileName != null) {
                                responseFlowFile = session.putAttribute(responseFlowFile, CoreAttributes.FILENAME.key(), fileName);
                            }
                        }

                        // transfer the message body to the payload
                        // can potentially be null in edge cases
                        if (bodyExists) {
                            // write content type attribute to response flowfile if it is available
                            final MediaType contentType = responseBody.contentType();
                            if (contentType != null) {
                                responseFlowFile = session.putAttribute(responseFlowFile, CoreAttributes.MIME_TYPE.key(), contentType.toString());
                            }
                            if (teeInputStream != null) {
                                responseFlowFile = session.importFrom(teeInputStream, responseFlowFile);
                            } else {
                                responseFlowFile = session.importFrom(responseBodyStream, responseFlowFile);
                            }

                            // emit provenance event
                            final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
                            if (requestFlowFile != null) {
                                session.getProvenanceReporter().fetch(responseFlowFile, urlProperty, millis);
                            } else {
                                session.getProvenanceReporter().receive(responseFlowFile, urlProperty, millis);
                            }
                        }
                    }

                    // if not successful and request flowfile is not null, store the response body into a flowfile attribute
                    if (outputBodyToRequestAttribute && bodyExists) {
                        String attributeKey = context.getProperty(RESPONSE_BODY_ATTRIBUTE_NAME).evaluateAttributeExpressions(requestFlowFile).getValue();
                        if (attributeKey == null) {
                            attributeKey = RESPONSE_BODY;
                        }
                        byte[] outputBuffer;
                        int size;

                        if (outputStreamToRequestAttribute != null) {
                            outputBuffer = outputStreamToRequestAttribute.getBuffer();
                            size = outputStreamToRequestAttribute.size();
                        } else {
                            outputBuffer = new byte[maxAttributeSize];
                            size = StreamUtils.fillBuffer(responseBodyStream, outputBuffer, false);
                        }
                        String bodyString = new String(outputBuffer, 0, size, getCharsetFromMediaType(responseBody.contentType()));
                        requestFlowFile = session.putAttribute(requestFlowFile, attributeKey, bodyString);

                        final long processingDuration = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
                        final String eventDetails = String.format("Response Body Attribute Added [%s] Processing Duration [%d ms]", attributeKey, processingDuration);
                        session.getProvenanceReporter().modifyAttributes(requestFlowFile, eventDetails);
                    }
                } finally {
                    if (outputStreamToRequestAttribute != null) {
                        outputStreamToRequestAttribute.close();
                    }
                    if (teeInputStream != null) {
                        teeInputStream.close();
                    } else if (responseBodyStream != null) {
                        responseBodyStream.close();
                    }
                }

                route(requestFlowFile, responseFlowFile, session, context, statusCode);

            }
        } catch (final Exception e) {
            if (requestFlowFile == null) {
                logger.error("Request Processing failed", e);
                context.yield();
            } else {
                logger.error("Request Processing failed: {}", requestFlowFile, e);
                requestFlowFile = session.penalize(requestFlowFile);
                requestFlowFile = session.putAttribute(requestFlowFile, EXCEPTION_CLASS, e.getClass().getName());
                requestFlowFile = session.putAttribute(requestFlowFile, EXCEPTION_MESSAGE, e.getMessage());
                session.transfer(requestFlowFile, FAILURE);
            }

            if (responseFlowFile != null) {
                session.remove(responseFlowFile);
            }
        }
    }

    private Request configureRequest(final ProcessContext context, final ProcessSession session, final FlowFile requestFlowFile, String url) {
        final Request.Builder requestBuilder = new Request.Builder();

        requestBuilder.url(url);
        final String authUser = trimToEmpty(context.getProperty(REQUEST_USERNAME).getValue());

        // If the username/password properties are set then check if digest auth is being used
        if ("false".equalsIgnoreCase(context.getProperty(REQUEST_DIGEST_AUTHENTICATION_ENABLED).getValue())) {
            if (!authUser.isEmpty()) {
                final String authPass = trimToEmpty(context.getProperty(REQUEST_PASSWORD).getValue());

                String credential = Credentials.basic(authUser, authPass);
                requestBuilder.header(HttpHeader.AUTHORIZATION.getHeader(), credential);
            } else {
                oauth2AccessTokenProviderOptional.ifPresent(oauth2AccessTokenProvider ->
                    requestBuilder.addHeader(HttpHeader.AUTHORIZATION.getHeader(), "Bearer " + oauth2AccessTokenProvider.getAccessDetails().getAccessToken())
                );
            }
        }

        final String contentEncoding = context.getProperty(REQUEST_CONTENT_ENCODING).getValue();
        final ContentEncodingStrategy contentEncodingStrategy = ContentEncodingStrategy.valueOf(contentEncoding);
        if (ContentEncodingStrategy.GZIP == contentEncodingStrategy) {
            requestBuilder.addHeader(HttpHeader.CONTENT_ENCODING.getHeader(), ContentEncodingStrategy.GZIP.getValue().toLowerCase());
        }

        final String method = getRequestMethod(context, requestFlowFile);
        final Optional<HttpMethod> httpMethodFound = findRequestMethod(method);

        final RequestBody requestBody;
        if (httpMethodFound.isPresent()) {
            final HttpMethod httpMethod = httpMethodFound.get();
            if (httpMethod.isRequestBodySupported()) {
                requestBody = getRequestBodyToSend(session, context, requestFlowFile, contentEncodingStrategy);
            } else {
                requestBody = null;
            }
        } else {
            requestBody = null;
        }
        requestBuilder.method(method, requestBody);

        setHeaderProperties(context, requestBuilder, requestFlowFile);
        return requestBuilder.build();
    }

    private RequestBody getRequestBodyToSend(final ProcessSession session, final ProcessContext context,
                                             final FlowFile requestFlowFile,
                                             final ContentEncodingStrategy contentEncodingStrategy
    ) {
        boolean requestBodyEnabled = context.getProperty(REQUEST_BODY_ENABLED).asBoolean();

        String evalContentType = context.getProperty(REQUEST_CONTENT_TYPE)
                .evaluateAttributeExpressions(requestFlowFile).getValue();
        final String contentType = StringUtils.isBlank(evalContentType) ? DEFAULT_CONTENT_TYPE : evalContentType;
        String formDataName = context.getProperty(REQUEST_FORM_DATA_NAME).evaluateAttributeExpressions(requestFlowFile).getValue();

        // Check for dynamic properties for form components.
        // Even if the flowfile is not sent, we may still send form parameters.
        Map<String, PropertyDescriptor> propertyDescriptors = new HashMap<>();
        for (final Map.Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet()) {
            Matcher matcher = FORM_DATA_NAME_PARAMETER_PATTERN.matcher(entry.getKey().getName());
            if (matcher.matches()) {
                propertyDescriptors.put(matcher.group(FORM_DATA_NAME_GROUP), entry.getKey());
            }
        }

        final boolean contentLengthUnknown = chunkedTransferEncoding || ContentEncodingStrategy.GZIP == contentEncodingStrategy;
        RequestBody requestBody = new RequestBody() {
            @Nullable
            @Override
            public MediaType contentType() {
                return MediaType.parse(contentType);
            }

            @Override
            public void writeTo(final BufferedSink sink) throws IOException {
                final BufferedSink outputSink = (ContentEncodingStrategy.GZIP == contentEncodingStrategy)
                        ? Okio.buffer(new GzipSink(sink))
                        : sink;

                session.read(requestFlowFile, inputStream -> {
                    final Source source = Okio.source(inputStream);
                    outputSink.writeAll(source);
                });

                // Close Output Sink for gzip to write trailing bytes
                if (ContentEncodingStrategy.GZIP == contentEncodingStrategy) {
                    outputSink.close();
                }
            }

            @Override
            public long contentLength() {
                return contentLengthUnknown ? -1 : requestFlowFile.getSize();
            }
        };

        if (propertyDescriptors.size() > 0 || StringUtils.isNotEmpty(formDataName)) {
            // we have form data
            MultipartBody.Builder builder = new Builder().setType(MultipartBody.FORM);
            boolean useFileName = context.getProperty(REQUEST_FORM_DATA_FILENAME_ENABLED).asBoolean();
            String contentFileName = null;
            if (useFileName) {
                contentFileName = requestFlowFile.getAttribute(CoreAttributes.FILENAME.key());
            }
            // loop through the dynamic form parameters
            for (final Map.Entry<String, PropertyDescriptor> entry : propertyDescriptors.entrySet()) {
                final String propValue = context.getProperty(entry.getValue().getName())
                        .evaluateAttributeExpressions(requestFlowFile).getValue();
                builder.addFormDataPart(entry.getKey(), propValue);
            }
            if (requestBodyEnabled) {
                builder.addFormDataPart(formDataName, contentFileName, requestBody);
            }
            return builder.build();
        } else if (requestBodyEnabled) {
            return requestBody;
        }
        return RequestBody.create(new byte[0], null);
    }

    private void setHeaderProperties(final ProcessContext context, final Request.Builder requestBuilder, final FlowFile requestFlowFile) {
        final String userAgent = trimToEmpty(context.getProperty(REQUEST_USER_AGENT).evaluateAttributeExpressions(requestFlowFile).getValue());
        requestBuilder.addHeader(HttpHeader.USER_AGENT.getHeader(), userAgent);

        if (context.getProperty(REQUEST_DATE_HEADER_ENABLED).asBoolean()) {
            final ZonedDateTime universalCoordinatedTimeNow = ZonedDateTime.now(ZoneOffset.UTC);
            requestBuilder.addHeader(HttpHeader.DATE.getHeader(), RFC_2616_DATE_TIME.format(universalCoordinatedTimeNow));
        }

        for (String headerKey : dynamicPropertyNames) {
            String headerValue = context.getProperty(headerKey).evaluateAttributeExpressions(requestFlowFile).getValue();

            // ignore form data name dynamic properties
            if (FORM_DATA_NAME_PARAMETER_PATTERN.matcher(headerKey).matches()) {
                continue;
            }

            requestBuilder.addHeader(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 (requestHeaderAttributesPattern != null && requestFlowFile != null) {
            Map<String, String> attributes = requestFlowFile.getAttributes();
            Matcher m = requestHeaderAttributesPattern.matcher("");
            for (Map.Entry<String, String> entry : attributes.entrySet()) {
                String headerKey = trimToEmpty(entry.getKey());
                if (IGNORED_REQUEST_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());
                    requestBuilder.addHeader(headerKey, headerVal);
                }
            }
        }
    }

    private void route(FlowFile request, FlowFile response, ProcessSession session, ProcessContext context, int statusCode) {
        // 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(RESPONSE_GENERATION_REQUIRED).asBoolean()) {
            session.transfer(response, RESPONSE);
            responseSent = true;
        }

        // transfer to the correct relationship
        // 2xx -> SUCCESS
        if (isSuccess(statusCode)) {
            // we have two flowfiles to transfer
            if (request != null) {
                session.transfer(request, ORIGINAL);
            }
            if (response != null && !responseSent) {
                session.transfer(response, RESPONSE);
            }

            // 5xx -> RETRY
        } else if (statusCode / 100 == 5) {
            if (request != null) {
                request = session.penalize(request);
                session.transfer(request, RETRY);
            }

            // 1xx, 3xx, 4xx -> NO RETRY
        } else {
            if (request != null) {
                if (context.getProperty(REQUEST_FAILURE_PENALIZATION_ENABLED).asBoolean()) {
                    request = session.penalize(request);
                }
                session.transfer(request, NO_RETRY);
            }
        }

    }

    private boolean isSuccess(int statusCode) {
        return statusCode / 100 == 2;
    }

    private void logRequest(ComponentLog logger, Request request) {
        if (logger.isDebugEnabled()) {
            logger.debug("\nRequest to remote service:\n\t{}\n{}",
                    request.url().url().toExternalForm(), getLogString(request.headers().toMultimap()));
        }
    }

    private void logResponse(ComponentLog logger, String url, Response response) {
        if (logger.isDebugEnabled()) {
            logger.debug("\nResponse from remote service:\n\t{}\n{}",
                    url, getLogString(response.headers().toMultimap()));
        }
    }

    private String getLogString(Map<String, List<String>> map) {
        StringBuilder sb = new StringBuilder();
        for (Map.Entry<String, List<String>> entry : map.entrySet()) {
            List<String> list = entry.getValue();
            if (!list.isEmpty()) {
                sb.append("\t");
                sb.append(entry.getKey());
                sb.append(": ");
                if (list.size() == 1) {
                    sb.append(list.get(0));
                } else {
                    sb.append(list);
                }
                sb.append("\n");
            }
        }
        return sb.toString();
    }

    /**
     * Returns a Map of flowfile attributes from the response http headers. Multivalue headers are naively converted to comma separated strings.
     */
    private Map<String, String> convertAttributesFromHeaders(final Response responseHttp) {
        // create a new hashmap to store the values from the connection
        final Map<String, String> attributes = new HashMap<>();
        final Headers headers = responseHttp.headers();
        headers.names().forEach((key) -> {
            final List<String> values = headers.values(key);
            // we ignore any headers with no actual values (rare)
            if (!values.isEmpty()) {
                // create a comma separated string from the values, this is stored in the map
                final String value = StringUtils.join(values, MULTIPLE_HEADER_DELIMITER);
                attributes.put(key, value);
            }
        });

        final Handshake handshake = responseHttp.handshake();
        if (handshake != null) {
            final Principal principal = handshake.peerPrincipal();
            if (principal != null) {
                attributes.put(REMOTE_DN, principal.getName());
            }
        }

        return attributes;
    }

    private Charset getCharsetFromMediaType(MediaType contentType) {
        return contentType != null ? contentType.charset(StandardCharsets.UTF_8) : StandardCharsets.UTF_8;
    }

    private static File getResponseCacheDirectory() throws IOException {
        return Files.createTempDirectory(InvokeHTTP.class.getSimpleName()).toFile();
    }

    private FlowFileNamingStrategy getFlowFileNamingStrategy(final ProcessContext context) {
        final String strategy = context.getProperty(RESPONSE_FLOW_FILE_NAMING_STRATEGY).getValue();
        return FlowFileNamingStrategy.valueOf(strategy);
    }

    private String getFileNameFromUrl(URL url) {
        String fileName = null;
        String path = StringUtils.removeEnd(url.getPath(), "/");

        if (!StringUtils.isEmpty(path)) {
            fileName = path.substring(path.lastIndexOf('/') + 1);
        }

        return fileName;
    }

    private Optional<HttpMethod> findRequestMethod(String method) {
        return Arrays.stream(HttpMethod.values())
                .filter(httpMethod -> httpMethod.name().equals(method))
                .findFirst();
    }

    private String getRequestMethod(final PropertyContext context, final FlowFile flowFile) {
        final String method = context.getProperty(HTTP_METHOD).evaluateAttributeExpressions(flowFile).getValue().toUpperCase();
        return trimToEmpty(method);
    }
}
