You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2015/03/01 21:14:21 UTC
[1/5] incubator-nifi git commit: NIFI-387: If possible don't use
ContentRepository.importFrom but just copy stream directly in
StandardProcessSession
Repository: incubator-nifi
Updated Branches:
refs/heads/develop 50744bfdc -> 83b33c805
NIFI-387: If possible don't use ContentRepository.importFrom but just copy stream directly in StandardProcessSession
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/739f0c25
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/739f0c25
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/739f0c25
Branch: refs/heads/develop
Commit: 739f0c25e2c885bcfa6eef81a9e4896e618ec0fb
Parents: 50744bf
Author: Mark Payne <ma...@hotmail.com>
Authored: Thu Feb 26 09:04:05 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Thu Feb 26 09:04:05 2015 -0500
----------------------------------------------------------------------
.../repository/StandardProcessSession.java | 46 +++++++++++---------
1 file changed, 26 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/739f0c25/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index dcb461c..8d2e456 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -2351,35 +2351,41 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
public FlowFile importFrom(final InputStream source, final FlowFile destination) {
validateRecordState(destination);
final StandardRepositoryRecord record = records.get(destination);
- final ContentClaim newClaim;
+ ContentClaim newClaim = null;
long claimOffset = 0L;
- final boolean appendToClaim = isMergeContent();
- if (appendToClaim) {
- enforceCurrentWriteClaimState();
- newClaim = currentWriteClaim;
- claimOffset = currentWriteClaimSize;
- } else {
- try {
- newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant());
- claimLog.debug("Creating ContentClaim {} for 'importFrom' for {}", newClaim, destination);
- } catch (final IOException e) {
- throw new FlowFileAccessException("Unable to create ContentClaim due to " + e.toString(), e);
- }
- }
-
final long newSize;
+ final boolean appendToClaim = isMergeContent();
try {
- final boolean append = isMergeContent();
- newSize = context.getContentRepository().importFrom(source, newClaim, append);
- bytesWritten.increment(newSize);
- currentWriteClaimSize += newSize;
+ if (appendToClaim) {
+ enforceCurrentWriteClaimState();
+ newClaim = currentWriteClaim;
+ claimOffset = currentWriteClaimSize;
+
+ final long bytesCopied = StreamUtils.copy(source, currentWriteClaimStream);
+ bytesWritten.increment(bytesCopied);
+ currentWriteClaimSize += bytesCopied;
+ newSize = bytesCopied;
+ } else {
+ try {
+ newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant());
+ claimLog.debug("Creating ContentClaim {} for 'importFrom' for {}", newClaim, destination);
+
+ newSize = context.getContentRepository().importFrom(source, newClaim, appendToClaim);
+ bytesWritten.increment(newSize);
+ currentWriteClaimSize += newSize;
+ } catch (final IOException e) {
+ throw new FlowFileAccessException("Unable to create ContentClaim due to " + e.toString(), e);
+ }
+ }
} catch (final Throwable t) {
if (appendToClaim) {
resetWriteClaims();
}
- destroyContent(newClaim);
+ if ( newClaim != null ) {
+ destroyContent(newClaim);
+ }
throw new FlowFileAccessException("Failed to import data from " + source + " for " + destination + " due to " + t.toString(), t);
}
[2/5] incubator-nifi git commit: NIFI-221: Fixed bugs and pulled in
changes from develop
Posted by ma...@apache.org.
NIFI-221: Fixed bugs and pulled in changes from develop
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/2d483238
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/2d483238
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/2d483238
Branch: refs/heads/develop
Commit: 2d48323878436c500fc90fd1d839740248d0f63d
Parents: 739f0c2
Author: Mark Payne <ma...@hotmail.com>
Authored: Thu Feb 26 09:05:28 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Thu Feb 26 09:05:28 2015 -0500
----------------------------------------------------------------------
nifi/nifi-assembly/pom.xml | 5 +
.../nifi-standard-processors/pom.xml | 4 +
.../processors/standard/HandleHttpRequest.java | 519 +++++++++++++++++++
.../processors/standard/HandleHttpResponse.java | 178 +++++++
.../org.apache.nifi.processor.Processor | 2 +
.../nifi-http-context-map-api/pom.xml | 23 +
.../org/apache/nifi/http/HttpContextMap.java | 72 +++
.../nifi-http-context-map-nar/pom.xml | 37 ++
.../nifi-http-context-map/pom.xml | 44 ++
.../nifi/http/StandardHttpContextMap.java | 176 +++++++
...org.apache.nifi.controller.ControllerService | 1 +
.../index.html | 67 +++
.../nifi-http-context-map-bundle/pom.xml | 31 ++
.../nifi-standard-services-api-nar/pom.xml | 5 +
.../nifi-standard-services/pom.xml | 2 +
nifi/nifi-nar-bundles/pom.xml | 11 +
nifi/pom.xml | 6 +
17 files changed, 1183 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-assembly/pom.xml b/nifi/nifi-assembly/pom.xml
index 9a55cbc..d65f294 100644
--- a/nifi/nifi-assembly/pom.xml
+++ b/nifi/nifi-assembly/pom.xml
@@ -156,6 +156,11 @@
<artifactId>nifi-kafka-nar</artifactId>
<type>nar</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-nar</artifactId>
+ <type>nar</type>
+ </dependency>
</dependencies>
<properties>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
index a482e19..0ccb3dd 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
@@ -48,6 +48,10 @@
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-distributed-cache-client-service-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-api</artifactId>
+ </dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
new file mode 100644
index 0000000..85d7433
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
@@ -0,0 +1,519 @@
+/*
+ * 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 java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.security.Principal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Enumeration;
+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.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
+import javax.security.cert.X509Certificate;
+import javax.servlet.AsyncContext;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.commons.lang3.StringUtils;
+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.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.http.HttpContextMap;
+import org.apache.nifi.processor.AbstractProcessor;
+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.ssl.SSLContextService;
+import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.server.HttpConfiguration;
+import org.eclipse.jetty.server.HttpConnectionFactory;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.SecureRequestCustomizer;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.SslConnectionFactory;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+
+import com.sun.jersey.api.client.ClientResponse.Status;
+
+@Tags({"http", "https", "request", "listen", "ingress", "web service"})
+@CapabilityDescription("Starts an HTTP Server and listens for HTTP Requests. For each request, creates a FlowFile and transfers to 'success'. This Processor is designed to be used in conjunction with the HandleHttpResponse Processor in order to create a Web Service")
+public class HandleHttpRequest extends AbstractProcessor {
+ public static final String HTTP_CONTEXT_ID = "http.context.identifier";
+
+ // Allowable values for client auth
+ public static final AllowableValue CLIENT_NONE = new AllowableValue("No Authentication", "Processor will not authenticate clients. Anyone can communicate with this Processor anonymously");
+ public static final AllowableValue CLIENT_WANT = new AllowableValue("Want Authentication", "Processor will try to verify the client but if unable to verify will allow the client to communicate anonymously");
+ public static final AllowableValue CLIENT_NEED = new AllowableValue("Need Authentication", "Processor will reject communications from any client unless the client provides a certificate that is trusted by the TrustStore specified in the SSL Context Service");
+
+
+ public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder()
+ .name("Listening Port")
+ .description("The Port to listen on for incoming HTTP requests")
+ .required(true)
+ .addValidator(StandardValidators.createLongValidator(0L, 65535L, true))
+ .expressionLanguageSupported(false)
+ .defaultValue("80")
+ .build();
+ public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder()
+ .name("Hostname")
+ .description("The Hostname to bind to. If not specified, will bind to all hosts")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+ public static final PropertyDescriptor HTTP_CONTEXT_MAP = new PropertyDescriptor.Builder()
+ .name("HTTP Context Map")
+ .description("The HTTP Context Map Controller Service to use for caching the HTTP Request Information")
+ .required(true)
+ .identifiesControllerService(HttpContextMap.class)
+ .build();
+ public static final PropertyDescriptor SSL_CONTEXT = new PropertyDescriptor.Builder()
+ .name("SSL Context Service")
+ .description("The SSL Context Service to use in order to secure the server. If specified, the server will accept only HTTPS requests; otherwise, the server will accept only HTTP requests")
+ .required(false)
+ .identifiesControllerService(SSLContextService.class)
+ .build();
+ public static final PropertyDescriptor PATH_REGEX = new PropertyDescriptor.Builder()
+ .name("Allowed Paths")
+ .description("A Regular Expression that specifies the valid HTTP Paths that are allowed in the incoming URL Requests. If this value is specified and the path of the HTTP Requests does not match this Regular Expression, the Processor will respond with a 404: NotFound")
+ .required(false)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+ public static final PropertyDescriptor ALLOW_GET = new PropertyDescriptor.Builder()
+ .name("Allow GET")
+ .description("Allow HTTP GET Method")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
+ public static final PropertyDescriptor ALLOW_POST = new PropertyDescriptor.Builder()
+ .name("Allow POST")
+ .description("Allow HTTP POST Method")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
+ public static final PropertyDescriptor ALLOW_PUT = new PropertyDescriptor.Builder()
+ .name("Allow PUT")
+ .description("Allow HTTP PUT Method")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
+ public static final PropertyDescriptor ALLOW_DELETE = new PropertyDescriptor.Builder()
+ .name("Allow DELETE")
+ .description("Allow HTTP DELETE Method")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
+ public static final PropertyDescriptor ALLOW_HEAD = new PropertyDescriptor.Builder()
+ .name("Allow HEAD")
+ .description("Allow HTTP HEAD Method")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+ public static final PropertyDescriptor ALLOW_OPTIONS = new PropertyDescriptor.Builder()
+ .name("Allow OPTIONS")
+ .description("Allow HTTP OPTIONS Method")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+ public static final PropertyDescriptor ADDITIONAL_METHODS = new PropertyDescriptor.Builder()
+ .name("Additional HTTP Methods")
+ .description("A comma-separated list of non-standard HTTP Methods that should be allowed")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .build();
+ public static final PropertyDescriptor CLIENT_AUTH = new PropertyDescriptor.Builder()
+ .name("Client Authentication")
+ .description("Specifies whether or not the Processor should authenticate clients. This value is ignored if the <SSL Context Service> Property is not specified or the SSL Context provided uses only a KeyStore and not a TrustStore.")
+ .required(true)
+ .allowableValues(CLIENT_NONE, CLIENT_WANT, CLIENT_NEED)
+ .defaultValue(CLIENT_NONE.getValue())
+ .build();
+
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("All content that is received is routed to the 'success' relationship")
+ .build();
+
+ private volatile Server server;
+ private final BlockingQueue<HttpRequestContainer> containerQueue = new LinkedBlockingQueue<>(50);
+
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(PORT);
+ descriptors.add(HOSTNAME);
+ descriptors.add(SSL_CONTEXT);
+ descriptors.add(HTTP_CONTEXT_MAP);
+ descriptors.add(PATH_REGEX);
+ descriptors.add(ALLOW_GET);
+ descriptors.add(ALLOW_POST);
+ descriptors.add(ALLOW_PUT);
+ descriptors.add(ALLOW_DELETE);
+ descriptors.add(ALLOW_HEAD);
+ descriptors.add(ALLOW_OPTIONS);
+ descriptors.add(ADDITIONAL_METHODS);
+
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return Collections.singleton(REL_SUCCESS);
+ }
+
+
+ @OnScheduled
+ public void initializeServer(final ProcessContext context) throws Exception {
+ final String host = context.getProperty(HOSTNAME).getValue();
+ final int port = context.getProperty(PORT).asInteger();
+ final SSLContextService sslService = context.getProperty(SSL_CONTEXT).asControllerService(SSLContextService.class);
+
+ final String clientAuthValue = context.getProperty(CLIENT_AUTH).getValue();
+ final boolean need;
+ final boolean want;
+ if ( CLIENT_NEED.equals(clientAuthValue) ) {
+ need = true;
+ want = false;
+ } else if ( CLIENT_WANT.equals(clientAuthValue) ) {
+ need = false;
+ want = true;
+ } else {
+ need = false;
+ want = false;
+ }
+
+ final SslContextFactory sslFactory = (sslService == null) ? null : createSslFactory(sslService, need, want);
+ final Server server = new Server(port);
+
+ // create the http configuration
+ final HttpConfiguration httpConfiguration = new HttpConfiguration();
+ if ( sslFactory == null ) {
+ // create the connector
+ final ServerConnector http = new ServerConnector(server, new HttpConnectionFactory(httpConfiguration));
+
+ // set host and port
+ if (StringUtils.isNotBlank(host)) {
+ http.setHost(host);
+ }
+ http.setPort(port);
+
+ // add this connector
+ server.setConnectors(new Connector[] {http});
+ } else {
+ // add some secure config
+ final HttpConfiguration httpsConfiguration = new HttpConfiguration(httpConfiguration);
+ httpsConfiguration.setSecureScheme("https");
+ httpsConfiguration.setSecurePort(port);
+ httpsConfiguration.addCustomizer(new SecureRequestCustomizer());
+
+ // build the connector
+ final ServerConnector https = new ServerConnector(server,
+ new SslConnectionFactory(sslFactory, "http/1.1"),
+ new HttpConnectionFactory(httpsConfiguration));
+
+ // set host and port
+ if (StringUtils.isNotBlank(host)) {
+ https.setHost(host);
+ }
+ https.setPort(port);
+
+ // add this connector
+ server.setConnectors(new Connector[] {https});
+ }
+
+ final Set<String> allowedMethods = new HashSet<>();
+ if ( context.getProperty(ALLOW_GET).asBoolean() ) {
+ allowedMethods.add("GET");
+ }
+ if ( context.getProperty(ALLOW_POST).asBoolean() ) {
+ allowedMethods.add("POST");
+ }
+ if ( context.getProperty(ALLOW_PUT).asBoolean() ) {
+ allowedMethods.add("PUT");
+ }
+ if ( context.getProperty(ALLOW_DELETE).asBoolean() ) {
+ allowedMethods.add("DELETE");
+ }
+ if ( context.getProperty(ALLOW_HEAD).asBoolean() ) {
+ allowedMethods.add("HEAD");
+ }
+ if ( context.getProperty(ALLOW_OPTIONS).asBoolean() ) {
+ allowedMethods.add("OPTIONS");
+ }
+
+ final String additionalMethods = context.getProperty(ADDITIONAL_METHODS).getValue();
+ if ( additionalMethods != null ) {
+ for ( final String additionalMethod : additionalMethods.split(",") ) {
+ final String trimmed = additionalMethod.trim();
+ if ( !trimmed.isEmpty() ) {
+ allowedMethods.add(trimmed.toUpperCase());
+ }
+ }
+ }
+
+ final String pathRegex = context.getProperty(PATH_REGEX).getValue();
+ final Pattern pathPattern = (pathRegex == null) ? null : Pattern.compile(pathRegex);
+
+ server.setHandler(new AbstractHandler() {
+ @Override
+ public void handle(final String target, final Request baseRequest, final HttpServletRequest request,
+ final HttpServletResponse response) throws IOException, ServletException {
+
+ final String requestUri = request.getRequestURI();
+ if ( !allowedMethods.contains(request.getMethod().toUpperCase()) ) {
+ getLogger().info("Sending back METHOD_NOT_ALLOWED response to {}; method was {}; request URI was {}",
+ new Object[] {request.getRemoteAddr(), request.getMethod(), requestUri});
+ response.sendError(Status.METHOD_NOT_ALLOWED.getStatusCode());
+ return;
+ }
+
+ if ( pathPattern != null ) {
+ final URI uri;
+ try {
+ uri = new URI(requestUri);
+ } catch (final URISyntaxException e) {
+ throw new ServletException(e);
+ }
+
+ if ( !pathPattern.matcher(uri.getPath()).matches() ) {
+ response.sendError(Status.NOT_FOUND.getStatusCode());
+ getLogger().info("Sending back NOT_FOUND response to {}; request was {} {}",
+ new Object[] {request.getRemoteAddr(), request.getMethod(), requestUri});
+ return;
+ }
+ }
+
+ // If destination queues full, send back a 503: Service Unavailable.
+ if ( context.getAvailableRelationships().isEmpty() ) {
+ response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
+ return;
+ }
+
+ // Right now, that information, though, is only in the ProcessSession, not the ProcessContext,
+ // so it is not known to us. Should see if it can be added to the ProcessContext.
+ final AsyncContext async = baseRequest.startAsync();
+ final boolean added = containerQueue.offer(new HttpRequestContainer(request, response, async));
+
+ if ( added ) {
+ getLogger().debug("Added Http Request to queue for {} {} from {}", new Object[] {request.getMethod(), requestUri, request.getRemoteAddr()});
+ } else {
+ getLogger().info("Sending back a SERVICE_UNAVAILABLE response to {}; request was {} {}",
+ new Object[] {request.getRemoteAddr(), request.getMethod(), request.getRemoteAddr()});
+
+ response.sendError(Status.SERVICE_UNAVAILABLE.getStatusCode());
+ response.flushBuffer();
+ async.complete();
+ }
+ }
+ });
+
+ this.server = server;
+ server.start();
+
+ getLogger().info("Server started and listening on port " + getPort());
+ }
+
+ protected int getPort() {
+ for ( final Connector connector : server.getConnectors() ) {
+ if ( connector instanceof ServerConnector ) {
+ return ((ServerConnector) connector).getPort();
+ }
+ }
+
+ throw new IllegalStateException("Server is not listening on any ports");
+ }
+
+ protected int getRequestQueueSize() {
+ return containerQueue.size();
+ }
+
+ private SslContextFactory createSslFactory(final SSLContextService sslService, final boolean needClientAuth, final boolean wantClientAuth) {
+ final SslContextFactory sslFactory = new SslContextFactory();
+
+ sslFactory.setNeedClientAuth(needClientAuth);
+ sslFactory.setWantClientAuth(wantClientAuth);
+
+ if ( sslService.isKeyStoreConfigured() ) {
+ sslFactory.setKeyStorePath(sslService.getKeyStoreFile());
+ sslFactory.setKeyStorePassword(sslService.getKeyStorePassword());
+ sslFactory.setKeyStoreType(sslService.getKeyStoreType());
+ }
+
+ if ( sslService.isTrustStoreConfigured() ) {
+ sslFactory.setTrustStorePath(sslService.getTrustStoreFile());
+ sslFactory.setTrustStorePassword(sslService.getTrustStorePassword());
+ sslFactory.setTrustStoreType(sslService.getTrustStoreType());
+ }
+
+ return sslFactory;
+ }
+
+ @OnStopped
+ public void shutdown() throws Exception {
+ if ( server != null ) {
+ getLogger().debug("Shutting down server");
+ server.stop();
+ server.join();
+ getLogger().info("Shut down {}", new Object[] {server});
+ }
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ final HttpRequestContainer container = containerQueue.poll();
+ if ( container == null ) {
+ return;
+ }
+
+ final long start = System.nanoTime();
+ final HttpServletRequest request = container.getRequest();
+ FlowFile flowFile = session.create();
+ try {
+ flowFile = session.importFrom(request.getInputStream(), flowFile);
+ } catch (final IOException e) {
+ getLogger().error("Failed to receive content from HTTP Request from {} due to {}", new Object[] {request.getRemoteAddr(), e});
+ session.remove(flowFile);
+ return;
+ }
+
+ final String contextIdentifier = UUID.randomUUID().toString();
+ final Map<String, String> attributes = new HashMap<>();
+ putAttribute(attributes, HTTP_CONTEXT_ID, contextIdentifier);
+ putAttribute(attributes, "mime.type", request.getContentType());
+ putAttribute(attributes, "http.servlet.path", request.getServletPath());
+ putAttribute(attributes, "http.context.path", request.getContextPath());
+ putAttribute(attributes, "http.method", request.getMethod());
+ putAttribute(attributes, "http.query.string", request.getQueryString());
+ putAttribute(attributes, "http.remote.host", request.getRemoteHost());
+ putAttribute(attributes, "http.remote.addr", request.getRemoteAddr());
+ putAttribute(attributes, "http.remote.user", request.getRemoteUser());
+ putAttribute(attributes, "http.request.uri", request.getRequestURI());
+ putAttribute(attributes, "http.auth.type", request.getAuthType());
+
+ final Enumeration<String> headerNames = request.getHeaderNames();
+ while ( headerNames.hasMoreElements() ) {
+ final String headerName = headerNames.nextElement();
+ final String headerValue = request.getHeader(headerName);
+ putAttribute(attributes, "http.headers." + headerName, headerValue);
+ }
+
+ final Principal principal = request.getUserPrincipal();
+ if ( principal != null ) {
+ putAttribute(attributes, "http.principal.name", principal.getName());
+ }
+
+ final X509Certificate certs[] = (X509Certificate[]) request.getAttribute("javax.servlet.request.X509Certificate");
+ final String subjectDn;
+ if ( certs != null && certs.length > 0 ) {
+ final X509Certificate cert = certs[0];
+ subjectDn = cert.getSubjectDN().getName();
+ final String issuerDn = cert.getIssuerDN().getName();
+
+ putAttribute(attributes, "http.subject.dn", subjectDn);
+ putAttribute(attributes, "http.issuer.dn", issuerDn);
+ } else {
+ subjectDn = null;
+ }
+
+ flowFile = session.putAllAttributes(flowFile, attributes);
+
+ final HttpContextMap contextMap = context.getProperty(HTTP_CONTEXT_MAP).asControllerService(HttpContextMap.class);
+ final boolean registered = contextMap.register(contextIdentifier, request, container.getResponse(), container.getContext());
+
+ if ( !registered ) {
+ getLogger().warn("Received request from {} but could not process it because too many requests are already outstanding; responding with SERVICE_UNAVAILABLE", new Object[] {request.getRemoteAddr()});
+
+ try {
+ container.getResponse().setStatus(Status.SERVICE_UNAVAILABLE.getStatusCode());
+ container.getResponse().flushBuffer();
+ container.getContext().complete();
+ } catch (final Exception e) {
+ getLogger().warn("Failed to respond with SERVICE_UNAVAILABLE message to {} due to {}", new Object[] {request.getRemoteAddr(), e});
+ }
+
+ return;
+ }
+
+ final long receiveMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ session.getProvenanceReporter().receive(flowFile, request.getRequestURI(), "Received from " + request.getRemoteAddr() + (subjectDn == null ? "" : " with DN=" + subjectDn), receiveMillis);
+ session.transfer(flowFile, REL_SUCCESS);
+ getLogger().info("Transferring {} to 'success'; received from {}", new Object[] {flowFile, request.getRemoteAddr()});
+ }
+
+ private void putAttribute(final Map<String, String> map, final String key, final String value) {
+ if ( value == null ) {
+ return;
+ }
+
+ map.put(key, value);
+ }
+
+
+ private static class HttpRequestContainer {
+ private final HttpServletRequest request;
+ private final HttpServletResponse response;
+ private final AsyncContext context;
+
+ public HttpRequestContainer(final HttpServletRequest request, final HttpServletResponse response, final AsyncContext async) {
+ this.request = request;
+ this.response = response;
+ this.context = async;
+ }
+
+ public HttpServletRequest getRequest() {
+ return request;
+ }
+
+ public HttpServletResponse getResponse() {
+ return response;
+ }
+
+ public AsyncContext getContext() {
+ return context;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java
new file mode 100644
index 0000000..fd55927
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpResponse.java
@@ -0,0 +1,178 @@
+/*
+ * 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 java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+import javax.servlet.http.HttpServletResponse;
+
+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.http.HttpContextMap;
+import org.apache.nifi.processor.AbstractProcessor;
+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;
+
+@Tags({"http", "https", "response", "egress", "web service"})
+@CapabilityDescription("Sends an HTTP Response to the Requestor that generated a FlowFile. This Processor is designed to be used in conjunction with the HandleHttpRequest in order to create a web service.")
+public class HandleHttpResponse extends AbstractProcessor {
+ public static final Pattern NUMBER_PATTERN = Pattern.compile("[0-9]+");
+ public static final String HTTP_CONTEXT_ID = "http.context.identifier";
+
+ public static final PropertyDescriptor STATUS_CODE = new PropertyDescriptor.Builder()
+ .name("HTTP Status Code")
+ .description("The HTTP Status Code to use when responding to the HTTP Request. See Section 10 of RFC 2616 for more information.")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ public static final PropertyDescriptor HTTP_CONTEXT_MAP = new PropertyDescriptor.Builder()
+ .name("HTTP Context Map")
+ .description("The HTTP Context Map Controller Service to use for caching the HTTP Request Information")
+ .required(true)
+ .identifiesControllerService(HttpContextMap.class)
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles will be routed to this Relationship after the response has been successfully sent to the requestor")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles will be routed to this Relationship if the Processor is unable to respond to the requestor. This may happen, for instance, if the connection times out or if NiFi is restarted before responding to the HTTP Request.")
+ .build();
+
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new ArrayList<>();
+ properties.add(STATUS_CODE);
+ properties.add(HTTP_CONTEXT_MAP);
+ return properties;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ return relationships;
+ }
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .description("Specifies the value to send for the '" + propertyDescriptorName + "' HTTP Header")
+ .name(propertyDescriptorName)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .dynamic(true)
+ .expressionLanguageSupported(true)
+ .build();
+ }
+
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if ( flowFile == null ) {
+ return;
+ }
+
+ final String contextIdentifier = flowFile.getAttribute(HTTP_CONTEXT_ID);
+ if ( contextIdentifier == null ) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().warn("Failed to respond to HTTP request for {} because FlowFile did not have an "
+ + "'http.context.identifier' attribute", new Object[] {flowFile});
+ return;
+ }
+
+ final String statusCodeValue = context.getProperty(STATUS_CODE).evaluateAttributeExpressions(flowFile).getValue();
+ if ( !isNumber(statusCodeValue) ) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().error("Failed to response to HTTP request for {} because status code was '{}', which is not a valid number", new Object[] {flowFile, statusCodeValue});
+ }
+
+ final HttpContextMap contextMap = context.getProperty(HTTP_CONTEXT_MAP).asControllerService(HttpContextMap.class);
+ final HttpServletResponse response = contextMap.getResponse(contextIdentifier);
+ if ( response == null ) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().error("Failed to respond to HTTP request for {} because FlowFile had an '{}' "
+ + "attribute of {} but could not find an HTTP Response Object for this identifier", new Object[] {flowFile, HTTP_CONTEXT_ID, contextIdentifier});
+ return;
+ }
+
+ final int statusCode = Integer.parseInt(statusCodeValue);
+ response.setStatus(statusCode);
+
+ for ( final Map.Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet() ) {
+ final PropertyDescriptor descriptor = entry.getKey();
+ if ( descriptor.isDynamic() ) {
+ final String headerName = descriptor.getName();
+ final String headerValue = context.getProperty(descriptor).evaluateAttributeExpressions(flowFile).getValue();
+
+ if ( !headerValue.trim().isEmpty() ) {
+ response.setHeader(headerName, headerValue);
+ }
+ }
+ }
+
+ try {
+ session.exportTo(flowFile, response.getOutputStream());
+ response.flushBuffer();
+ } catch (final IOException ioe) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().error("Failed to respond to HTTP request for {} due to {}", new Object[] {flowFile, ioe});
+ return;
+ }
+
+ try {
+ contextMap.complete(contextIdentifier);
+ } catch (final IllegalStateException ise) {
+ getLogger().error("Failed to complete HTTP Transaction for {} due to {}", new Object[] {flowFile, ise});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ getLogger().info("Successfully responded to HTTP Request for {} with status code {}", new Object[] {flowFile, statusCode});
+ }
+
+ private static boolean isNumber(final String value) {
+ if ( value.length() == 0 ) {
+ return false;
+ }
+
+ for (int i=0; i < value.length(); i++) {
+ if ( !Character.isDigit(value.charAt(i)) ) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 66c9697..9702b56 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -28,6 +28,8 @@ org.apache.nifi.processors.standard.GetFile
org.apache.nifi.processors.standard.GetFTP
org.apache.nifi.processors.standard.GetHTTP
org.apache.nifi.processors.standard.GetSFTP
+org.apache.nifi.processors.standard.HandleHttpRequest
+org.apache.nifi.processors.standard.HandleHttpResponse
org.apache.nifi.processors.standard.HashAttribute
org.apache.nifi.processors.standard.HashContent
org.apache.nifi.processors.standard.IdentifyMimeType
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/pom.xml
new file mode 100644
index 0000000..262a41b
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/pom.xml
@@ -0,0 +1,23 @@
+<?xml version="1.0"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-http-context-map-api</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>javax.servlet</groupId>
+ <artifactId>javax.servlet-api</artifactId>
+ </dependency>
+ </dependencies>
+
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/src/main/java/org/apache/nifi/http/HttpContextMap.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/src/main/java/org/apache/nifi/http/HttpContextMap.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/src/main/java/org/apache/nifi/http/HttpContextMap.java
new file mode 100644
index 0000000..04ff6ce
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-api/src/main/java/org/apache/nifi/http/HttpContextMap.java
@@ -0,0 +1,72 @@
+/*
+ * 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.http;
+
+import javax.servlet.AsyncContext;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.nifi.controller.ControllerService;
+
+
+/**
+ * <p>
+ * An interface that provides the capability of receiving an HTTP servlet request in one component
+ * and responding to that request in another component.
+ * </p>
+ *
+ * <p>
+ * The intended flow is for the component receiving the HTTP request to register the request, response,
+ * and AsyncContext with a particular identifier via the
+ * {@link #register(String, HttpServletRequest, HttpServletResponse, AsyncContext)}
+ * method. Another component is then able to obtain the response
+ * by providing that identifier to the {@link #getResponse(String)} method. After writing to the
+ * HttpServletResponse, the transaction is to then be completed via the {@link #complete(String)} method.
+ * </p>
+ */
+public interface HttpContextMap extends ControllerService {
+
+ /**
+ * Registers an HttpServletRequest, HttpServletResponse, and the AsyncContext for a given identifier
+ *
+ * @param identifier
+ * @param request
+ * @param response
+ * @param context
+ *
+ * @return true if register is successful, false if the context map is too full because too many requests have already been received and not processed
+ *
+ * @throws IllegalStateException if the identifier is already registered
+ */
+ boolean register(String identifier, HttpServletRequest request, HttpServletResponse response, AsyncContext context);
+
+ /**
+ * Retrieves the HttpServletResponse for the given identifier, if it exists
+ * @param identifier
+ * @return the HttpServletResponse for the given identifier, or {@code null} if it does not exist
+ */
+ HttpServletResponse getResponse(String identifier);
+
+ /**
+ * Marks the HTTP request/response for the given identifier as complete
+ * @param identifier
+ *
+ * @throws IllegalStateException if the identifier is not registered to a valid AsyncContext
+ */
+ void complete(String identifier);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map-nar/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map-nar/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map-nar/pom.xml
new file mode 100644
index 0000000..653c218
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map-nar/pom.xml
@@ -0,0 +1,37 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <!--
+ 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.
+ -->
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-bundle</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-http-context-map-nar</artifactId>
+ <packaging>nar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map</artifactId>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/pom.xml
new file mode 100644
index 0000000..abf78db
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/pom.xml
@@ -0,0 +1,44 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <!--
+ 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.
+ -->
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-bundle</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-http-context-map</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>javax.servlet</groupId>
+ <artifactId>javax.servlet-api</artifactId>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/java/org/apache/nifi/http/StandardHttpContextMap.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/java/org/apache/nifi/http/StandardHttpContextMap.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/java/org/apache/nifi/http/StandardHttpContextMap.java
new file mode 100644
index 0000000..c73b98b
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/java/org/apache/nifi/http/StandardHttpContextMap.java
@@ -0,0 +1,176 @@
+/*
+ * 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.http;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import javax.servlet.AsyncContext;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.util.StandardValidators;
+
+public class StandardHttpContextMap extends AbstractControllerService implements HttpContextMap {
+ public static final PropertyDescriptor MAX_OUTSTANDING_REQUESTS = new PropertyDescriptor.Builder()
+ .name("Maximum Outstanding Requests")
+ .description("The maximum number of HTTP requests that can be outstanding at any one time. Any attempt to register an additional HTTP Request will cause an error")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5000")
+ .build();
+ public static final PropertyDescriptor REQUEST_EXPIRATION = new PropertyDescriptor.Builder()
+ .name("Request Expiration")
+ .description("Specifies how long an HTTP Request should be left unanswered before being evicted from the cache and being responded to with a Service Unavailable status code")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .defaultValue("1 min")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .build();
+
+ private final ConcurrentMap<String, Wrapper> wrapperMap = new ConcurrentHashMap<>();
+
+ private volatile int maxSize = 5000;
+ private volatile long maxRequestNanos;
+ private volatile ScheduledExecutorService executor;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ final List<PropertyDescriptor> properties = new ArrayList<>(2);
+ properties.add(MAX_OUTSTANDING_REQUESTS);
+ properties.add(REQUEST_EXPIRATION);
+ return properties;
+ }
+
+ @OnEnabled
+ public void onConfigured(final ConfigurationContext context) {
+ maxSize = context.getProperty(MAX_OUTSTANDING_REQUESTS).asInteger();
+ executor = Executors.newSingleThreadScheduledExecutor();
+
+ maxRequestNanos = context.getProperty(REQUEST_EXPIRATION).asTimePeriod(TimeUnit.NANOSECONDS);
+ final long scheduleNanos = maxRequestNanos / 2;
+ executor.scheduleWithFixedDelay(new CleanupExpiredRequests(), scheduleNanos, scheduleNanos, TimeUnit.NANOSECONDS);
+ }
+
+ @OnDisabled
+ public void cleanup() {
+ if ( executor != null ) {
+ executor.shutdown();
+ }
+ }
+
+ @Override
+ public boolean register(final String identifier, final HttpServletRequest request, final HttpServletResponse response, final AsyncContext context) {
+ // fail if there are too many already. Maybe add a configuration property for how many
+ // outstanding, with a default of say 5000
+ if ( wrapperMap.size() >= maxSize ) {
+ return false;
+ }
+ final Wrapper wrapper = new Wrapper(request, response, context);
+ final Wrapper existing = wrapperMap.putIfAbsent(identifier, wrapper);
+ if ( existing != null ) {
+ throw new IllegalStateException("HTTP Request already registered with identifier " + identifier);
+ }
+
+ return true;
+ }
+
+ @Override
+ public HttpServletResponse getResponse(final String identifier) {
+ final Wrapper wrapper = wrapperMap.get(identifier);
+ if ( wrapper == null ) {
+ return null;
+ }
+
+ return wrapper.getResponse();
+ }
+
+ @Override
+ public void complete(final String identifier) {
+ final Wrapper wrapper = wrapperMap.remove(identifier);
+ if ( wrapper == null ) {
+ throw new IllegalStateException("No HTTP Request registered with identifier " + identifier);
+ }
+
+ wrapper.getAsync().complete();
+ }
+
+ private static class Wrapper {
+ @SuppressWarnings("unused")
+ private final HttpServletRequest request;
+ private final HttpServletResponse response;
+ private final AsyncContext async;
+ private final long nanoTimeAdded = System.nanoTime();
+
+ public Wrapper(final HttpServletRequest request, final HttpServletResponse response, final AsyncContext async) {
+ this.request = request;
+ this.response = response;
+ this.async = async;
+ }
+
+ public HttpServletResponse getResponse() {
+ return response;
+ }
+
+ public AsyncContext getAsync() {
+ return async;
+ }
+
+ public long getNanoTimeAdded() {
+ return nanoTimeAdded;
+ }
+ }
+
+ private class CleanupExpiredRequests implements Runnable {
+ @Override
+ public void run() {
+ final long now = System.nanoTime();
+ final long threshold = now - maxRequestNanos;
+
+ final Iterator<Map.Entry<String, Wrapper>> itr = wrapperMap.entrySet().iterator();
+ while ( itr.hasNext() ) {
+ final Map.Entry<String, Wrapper> entry = itr.next();
+ if ( entry.getValue().getNanoTimeAdded() < threshold ) {
+ itr.remove();
+
+ // send SERVICE_UNAVAILABLE
+ try {
+ final AsyncContext async = entry.getValue().getAsync();
+ ((HttpServletResponse) async.getResponse()).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
+ async.complete();
+ } catch (final IOException ioe) {
+ // we are trying to indicate that we are unavailable. If we have an IOException and cannot respond,
+ // then so be it. Nothing to really do here.
+ }
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
new file mode 100644
index 0000000..8c5721c
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -0,0 +1 @@
+org.apache.nifi.http.StandardHttpContextMap
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html
new file mode 100644
index 0000000..5c8b83a
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html
@@ -0,0 +1,67 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>StandardHttpContextMap</title>
+
+ <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <h2>Description:</h2>
+ <p>
+ This is the standard implementation of the SSL Context Map. This service is used to provide
+ coordination between
+ <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a>
+ and
+ <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a>
+ Processors.
+ </p>
+
+ <!-- Service Documentation ================================================== -->
+ <h2>Configuring the HTTP Context Map:</h2>
+ <p>
+ The <code>controller-services.xml</code> file is located in the NiFi <code>conf</code>
+ directory. The user may set up any number of controller services within this file.
+ </p>
+
+ <p>
+ This controller service exposes a single property named <code>Maximum Outstanding Requests</code>.
+ This property determines the maximum number of HTTP requests that can be outstanding at any one time.
+ Any attempt to register an additional HTTP Request will cause an error. The default value is 5000.
+ Below is an example of the template for a StandardHttpContextMap controller service.
+ </p>
+
+ <pre>
+<?xml version="1.0" encoding="UTF-8" ?>
+<services>
+ <service>
+ <identifier>http-context-map</identifier>
+ <class>org.apache.nifi.http.StandardHttpContextMap</class>
+ <property name="Maximum Outstanding Requests">5000</property>
+ </service>
+</services>
+ </pre>
+
+ <p>
+ <strong>See Also:</strong><br />
+ <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a><br />
+ <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a><br />
+ </p>
+
+ </body>
+</html>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/pom.xml
new file mode 100644
index 0000000..5539bd8
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/pom.xml
@@ -0,0 +1,31 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <!--
+ 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.
+ -->
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-http-context-map-bundle</artifactId>
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>nifi-http-context-map</module>
+ <module>nifi-http-context-map-nar</module>
+ </modules>
+
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/nifi-standard-services-api-nar/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-standard-services-api-nar/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-standard-services-api-nar/pom.xml
index 0da53fc..7668cda 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-standard-services-api-nar/pom.xml
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-standard-services-api-nar/pom.xml
@@ -37,5 +37,10 @@
<artifactId>nifi-load-distribution-service-api</artifactId>
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-api</artifactId>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/nifi-standard-services/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-services/pom.xml
index 449170c..47762b3 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/pom.xml
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/pom.xml
@@ -26,8 +26,10 @@
<module>nifi-distributed-cache-client-service-api</module>
<module>nifi-distributed-cache-services-bundle</module>
<module>nifi-load-distribution-service-api</module>
+ <module>nifi-http-context-map-api</module>
<module>nifi-ssl-context-bundle</module>
<module>nifi-ssl-context-service-api</module>
+ <module>nifi-http-context-map-bundle</module>
<module>nifi-standard-services-api-nar</module>
</modules>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/nifi-nar-bundles/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/pom.xml b/nifi/nifi-nar-bundles/pom.xml
index 229353f..758fe58 100644
--- a/nifi/nifi-nar-bundles/pom.xml
+++ b/nifi/nifi-nar-bundles/pom.xml
@@ -60,6 +60,12 @@
<version>0.0.2-incubating-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-api</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-distributed-cache-protocol</artifactId>
@@ -75,6 +81,11 @@
<artifactId>nifi-ssl-context-service</artifactId>
<version>0.0.2-incubating-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-volatile-provenance-repository</artifactId>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2d483238/nifi/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/pom.xml b/nifi/pom.xml
index 5881db7..3e42e28 100644
--- a/nifi/pom.xml
+++ b/nifi/pom.xml
@@ -753,6 +753,12 @@
<version>0.0.2-incubating-SNAPSHOT</version>
<type>nar</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-http-context-map-nar</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-properties</artifactId>
[5/5] incubator-nifi git commit: NIFI-221: Added additionally
flowfile attributes
Posted by ma...@apache.org.
NIFI-221: Added additionally flowfile attributes
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/83b33c80
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/83b33c80
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/83b33c80
Branch: refs/heads/develop
Commit: 83b33c8050c024404209ad504935a50d478e0db1
Parents: 7c99054
Author: Mark Payne <ma...@hotmail.com>
Authored: Sun Mar 1 15:12:24 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Sun Mar 1 15:12:24 2015 -0500
----------------------------------------------------------------------
.../processors/standard/HandleHttpRequest.java | 48 ++++++++++++++++++--
.../standard/TestHandleHttpRequest.java | 3 +-
2 files changed, 47 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/83b33c80/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
index e72df7d..575bf95 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
@@ -39,6 +39,7 @@ import java.util.regex.Pattern;
import javax.security.cert.X509Certificate;
import javax.servlet.AsyncContext;
import javax.servlet.ServletException;
+import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@@ -111,8 +112,8 @@ public class HandleHttpRequest extends AbstractProcessor {
.identifiesControllerService(SSLContextService.class)
.build();
public static final PropertyDescriptor URL_CHARACTER_SET = new PropertyDescriptor.Builder()
- .name("URL Character Set")
- .description("The character set to use for decoding URL parameters")
+ .name("Default URL Character Set")
+ .description("The character set to use for decoding URL parameters if the HTTP Request does not supply one")
.required(true)
.defaultValue("UTF-8")
.addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
@@ -432,7 +433,7 @@ public class HandleHttpRequest extends AbstractProcessor {
return;
}
- final String charset = context.getProperty(URL_CHARACTER_SET).getValue();
+ final String charset = request.getCharacterEncoding() == null ? context.getProperty(URL_CHARACTER_SET).getValue() : request.getCharacterEncoding();
final String contextIdentifier = UUID.randomUUID().toString();
final Map<String, String> attributes = new HashMap<>();
@@ -442,6 +443,8 @@ public class HandleHttpRequest extends AbstractProcessor {
putAttribute(attributes, "http.servlet.path", request.getServletPath());
putAttribute(attributes, "http.context.path", request.getContextPath());
putAttribute(attributes, "http.method", request.getMethod());
+ putAttribute(attributes, "http.local.addr", request.getLocalAddr());
+ putAttribute(attributes, "http.local.name", request.getLocalName());
if ( request.getQueryString() != null ) {
putAttribute(attributes, "http.query.string", URLDecoder.decode(request.getQueryString(), charset));
}
@@ -449,8 +452,39 @@ public class HandleHttpRequest extends AbstractProcessor {
putAttribute(attributes, "http.remote.addr", request.getRemoteAddr());
putAttribute(attributes, "http.remote.user", request.getRemoteUser());
putAttribute(attributes, "http.request.uri", request.getRequestURI());
+ putAttribute(attributes, "http.request.url", request.getRequestURL().toString());
putAttribute(attributes, "http.auth.type", request.getAuthType());
+ putAttribute(attributes, "http.requested.session.id", request.getRequestedSessionId());
+ if ( request.getDispatcherType() != null ) {
+ putAttribute(attributes, "http.dispatcher.type", request.getDispatcherType().name());
+ }
+ putAttribute(attributes, "http.character.encoding", request.getCharacterEncoding());
+ putAttribute(attributes, "http.locale", request.getLocale());
+ putAttribute(attributes, "http.server.name", request.getServerName());
+ putAttribute(attributes, "http.server.port", request.getServerPort());
+
+ final Enumeration<String> paramEnumeration = request.getParameterNames();
+ while ( paramEnumeration.hasMoreElements() ) {
+ final String paramName = paramEnumeration.nextElement();
+ final String value = request.getParameter(paramName);
+ attributes.put("http.param." + paramName, value);
+ }
+
+ final Cookie[] cookies = request.getCookies();
+ if ( cookies != null ) {
+ for ( final Cookie cookie : cookies ) {
+ final String name = cookie.getName();
+ final String cookiePrefix = "http.cookie." + name + ".";
+ attributes.put(cookiePrefix + "value", cookie.getValue());
+ attributes.put(cookiePrefix + "domain", cookie.getDomain());
+ attributes.put(cookiePrefix + "path", cookie.getPath());
+ attributes.put(cookiePrefix + "max.age", String.valueOf(cookie.getMaxAge()));
+ attributes.put(cookiePrefix + "version", String.valueOf(cookie.getVersion()));
+ attributes.put(cookiePrefix + "secure", String.valueOf(cookie.getSecure()));
+ }
+ }
+
final String queryString = request.getQueryString();
if ( queryString != null ) {
final String[] params = URL_QUERY_PARAM_DELIMITER.split(queryString);
@@ -527,6 +561,14 @@ public class HandleHttpRequest extends AbstractProcessor {
getLogger().info("Transferring {} to 'success'; received from {}", new Object[] {flowFile, request.getRemoteAddr()});
}
+ private void putAttribute(final Map<String, String> map, final String key, final Object value) {
+ if ( value == null ) {
+ return;
+ }
+
+ putAttribute(map, key, value.toString());
+ }
+
private void putAttribute(final Map<String, String> map, final String key, final String value) {
if ( value == null ) {
return;
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/83b33c80/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
index 85f35e2..363fc0f 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
@@ -78,7 +78,8 @@ public class TestHandleHttpRequest {
});
httpThread.start();
- try { Thread.sleep(100L); } catch (final InterruptedException ie) {}
+ // give processor a bit to handle the http request
+ try { Thread.sleep(1000L); } catch (final InterruptedException ie) {}
// process the request.
runner.run(1, false);
[4/5] incubator-nifi git commit: NIFI-221: Finished initial
implementation of http procs
Posted by ma...@apache.org.
NIFI-221: Finished initial implementation of http procs
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/7c990541
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/7c990541
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/7c990541
Branch: refs/heads/develop
Commit: 7c99054183428b0a696cdf16be46267089f9faf4
Parents: c53b0f9
Author: Mark Payne <ma...@hotmail.com>
Authored: Sun Mar 1 14:31:26 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Sun Mar 1 14:31:26 2015 -0500
----------------------------------------------------------------------
.../processors/standard/HandleHttpRequest.java | 66 +++++--
.../standard/TestHandleHttpRequest.java | 127 ++++++++++++++
.../standard/TestHandleHttpResponse.java | 172 +++++++++++++++++++
3 files changed, 353 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7c990541/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
index 98b5fdd..e72df7d 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
@@ -17,8 +17,10 @@
package org.apache.nifi.processors.standard;
import java.io.IOException;
+import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.net.URLDecoder;
import java.security.Principal;
import java.util.ArrayList;
import java.util.Collections;
@@ -73,6 +75,7 @@ import com.sun.jersey.api.client.ClientResponse.Status;
@CapabilityDescription("Starts an HTTP Server and listens for HTTP Requests. For each request, creates a FlowFile and transfers to 'success'. This Processor is designed to be used in conjunction with the HandleHttpResponse Processor in order to create a Web Service")
public class HandleHttpRequest extends AbstractProcessor {
public static final String HTTP_CONTEXT_ID = "http.context.identifier";
+ private static final Pattern URL_QUERY_PARAM_DELIMITER = Pattern.compile("&");
// Allowable values for client auth
public static final AllowableValue CLIENT_NONE = new AllowableValue("No Authentication", "Processor will not authenticate clients. Anyone can communicate with this Processor anonymously");
@@ -107,6 +110,13 @@ public class HandleHttpRequest extends AbstractProcessor {
.required(false)
.identifiesControllerService(SSLContextService.class)
.build();
+ public static final PropertyDescriptor URL_CHARACTER_SET = new PropertyDescriptor.Builder()
+ .name("URL Character Set")
+ .description("The character set to use for decoding URL parameters")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .build();
public static final PropertyDescriptor PATH_REGEX = new PropertyDescriptor.Builder()
.name("Allowed Paths")
.description("A Regular Expression that specifies the valid HTTP Paths that are allowed in the incoming URL Requests. If this value is specified and the path of the HTTP Requests does not match this Regular Expression, the Processor will respond with a 404: NotFound")
@@ -189,6 +199,7 @@ public class HandleHttpRequest extends AbstractProcessor {
descriptors.add(SSL_CONTEXT);
descriptors.add(HTTP_CONTEXT_MAP);
descriptors.add(PATH_REGEX);
+ descriptors.add(URL_CHARACTER_SET);
descriptors.add(ALLOW_GET);
descriptors.add(ALLOW_POST);
descriptors.add(ALLOW_PUT);
@@ -360,7 +371,7 @@ public class HandleHttpRequest extends AbstractProcessor {
protected int getPort() {
for ( final Connector connector : server.getConnectors() ) {
if ( connector instanceof ServerConnector ) {
- return ((ServerConnector) connector).getPort();
+ return ((ServerConnector) connector).getLocalPort();
}
}
@@ -421,19 +432,50 @@ public class HandleHttpRequest extends AbstractProcessor {
return;
}
+ final String charset = context.getProperty(URL_CHARACTER_SET).getValue();
+
final String contextIdentifier = UUID.randomUUID().toString();
final Map<String, String> attributes = new HashMap<>();
- putAttribute(attributes, HTTP_CONTEXT_ID, contextIdentifier);
- putAttribute(attributes, "mime.type", request.getContentType());
- putAttribute(attributes, "http.servlet.path", request.getServletPath());
- putAttribute(attributes, "http.context.path", request.getContextPath());
- putAttribute(attributes, "http.method", request.getMethod());
- putAttribute(attributes, "http.query.string", request.getQueryString());
- putAttribute(attributes, "http.remote.host", request.getRemoteHost());
- putAttribute(attributes, "http.remote.addr", request.getRemoteAddr());
- putAttribute(attributes, "http.remote.user", request.getRemoteUser());
- putAttribute(attributes, "http.request.uri", request.getRequestURI());
- putAttribute(attributes, "http.auth.type", request.getAuthType());
+ try {
+ putAttribute(attributes, HTTP_CONTEXT_ID, contextIdentifier);
+ putAttribute(attributes, "mime.type", request.getContentType());
+ putAttribute(attributes, "http.servlet.path", request.getServletPath());
+ putAttribute(attributes, "http.context.path", request.getContextPath());
+ putAttribute(attributes, "http.method", request.getMethod());
+ if ( request.getQueryString() != null ) {
+ putAttribute(attributes, "http.query.string", URLDecoder.decode(request.getQueryString(), charset));
+ }
+ putAttribute(attributes, "http.remote.host", request.getRemoteHost());
+ putAttribute(attributes, "http.remote.addr", request.getRemoteAddr());
+ putAttribute(attributes, "http.remote.user", request.getRemoteUser());
+ putAttribute(attributes, "http.request.uri", request.getRequestURI());
+ putAttribute(attributes, "http.auth.type", request.getAuthType());
+
+ final String queryString = request.getQueryString();
+ if ( queryString != null ) {
+ final String[] params = URL_QUERY_PARAM_DELIMITER.split(queryString);
+ for ( final String keyValueString : params ) {
+ final int indexOf = keyValueString.indexOf("=");
+ if ( indexOf < 0 ) {
+ // no =, then it's just a key with no value
+ attributes.put("http.query.param." + URLDecoder.decode(keyValueString, charset), "");
+ } else {
+ final String key = keyValueString.substring(0, indexOf);
+ final String value;
+
+ if ( indexOf == keyValueString.length() - 1 ) {
+ value = "";
+ } else {
+ value = keyValueString.substring(indexOf + 1);
+ }
+
+ attributes.put("http.query.param." + URLDecoder.decode(key, charset), URLDecoder.decode(value, charset));
+ }
+ }
+ }
+ } catch (final UnsupportedEncodingException uee) {
+ throw new ProcessException("Invalid character encoding", uee); // won't happen because charset has been validated
+ }
final Enumeration<String> headerNames = request.getHeaderNames();
while ( headerNames.hasMoreElements() ) {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7c990541/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
new file mode 100644
index 0000000..85f35e2
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
@@ -0,0 +1,127 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import javax.servlet.AsyncContext;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.http.HttpContextMap;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.stream.io.NullOutputStream;
+import org.apache.nifi.stream.io.StreamUtils;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestHandleHttpRequest {
+
+ @Test
+ public void testRequestAddedToService() throws InitializationException, MalformedURLException, IOException {
+ final TestRunner runner = TestRunners.newTestRunner(HandleHttpRequest.class);
+ runner.setProperty(HandleHttpRequest.PORT, "0");
+
+ final MockHttpContextMap contextMap = new MockHttpContextMap();
+ runner.addControllerService("http-context-map", contextMap);
+ runner.enableControllerService(contextMap);
+ runner.setProperty(HandleHttpRequest.HTTP_CONTEXT_MAP, "http-context-map");
+
+ // trigger processor to stop but not shutdown.
+ runner.run(1, false);
+ try {
+ final Thread httpThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ final int port = ((HandleHttpRequest) runner.getProcessor()).getPort();
+ final HttpURLConnection connection = (HttpURLConnection) new URL("http://localhost:" + port + "/my/path?query=true&value1=value1&value2=&value3&value4=apple=orange").openConnection();
+ connection.setDoOutput(false);
+ connection.setRequestMethod("GET");
+ connection.setRequestProperty("header1", "value1");
+ connection.setRequestProperty("header2", "");
+ connection.setRequestProperty("header3", "apple=orange");
+ connection.setConnectTimeout(3000);
+ connection.setReadTimeout(3000);
+
+ StreamUtils.copy(connection.getInputStream(), new NullOutputStream());
+ } catch (final Throwable t) {
+ t.printStackTrace();
+ Assert.fail(t.toString());
+ }
+ }
+ });
+ httpThread.start();
+
+ try { Thread.sleep(100L); } catch (final InterruptedException ie) {}
+
+ // process the request.
+ runner.run(1, false);
+
+ runner.assertAllFlowFilesTransferred(HandleHttpRequest.REL_SUCCESS, 1);
+ assertEquals(1, contextMap.size());
+
+ final MockFlowFile mff = runner.getFlowFilesForRelationship(HandleHttpRequest.REL_SUCCESS).get(0);
+ mff.assertAttributeEquals("http.query.param.query", "true");
+ mff.assertAttributeEquals("http.query.param.value1", "value1");
+ mff.assertAttributeEquals("http.query.param.value2", "");
+ mff.assertAttributeEquals("http.query.param.value3", "");
+ mff.assertAttributeEquals("http.query.param.value4", "apple=orange");
+ mff.assertAttributeEquals("http.headers.header1", "value1");
+ mff.assertAttributeEquals("http.headers.header3", "apple=orange");
+ } finally {
+ // shut down the server
+ runner.run(1, true);
+ }
+ }
+
+
+ private static class MockHttpContextMap extends AbstractControllerService implements HttpContextMap {
+ private final ConcurrentMap<String, HttpServletResponse> responseMap = new ConcurrentHashMap<>();
+
+ @Override
+ public boolean register(String identifier, HttpServletRequest request, HttpServletResponse response, AsyncContext context) {
+ responseMap.put(identifier, response);
+ return true;
+ }
+
+ @Override
+ public HttpServletResponse getResponse(String identifier) {
+ return responseMap.get(identifier);
+ }
+
+ @Override
+ public void complete(String identifier) {
+ responseMap.remove(identifier);
+ }
+
+ public int size() {
+ return responseMap.size();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7c990541/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpResponse.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpResponse.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpResponse.java
new file mode 100644
index 0000000..7b41809
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpResponse.java
@@ -0,0 +1,172 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.servlet.AsyncContext;
+import javax.servlet.ServletOutputStream;
+import javax.servlet.WriteListener;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.http.HttpContextMap;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+public class TestHandleHttpResponse {
+
+ @Test
+ public void testEnsureCompleted() throws InitializationException {
+ final TestRunner runner = TestRunners.newTestRunner(HandleHttpResponse.class);
+
+ final MockHttpContextMap contextMap = new MockHttpContextMap("my-id");
+ runner.addControllerService("http-context-map", contextMap);
+ runner.enableControllerService(contextMap);
+ runner.setProperty(HandleHttpResponse.HTTP_CONTEXT_MAP, "http-context-map");
+ runner.setProperty(HandleHttpResponse.STATUS_CODE, "${status.code}");
+ runner.setProperty("my-attr", "${my-attr}");
+ runner.setProperty("no-valid-attr", "${no-valid-attr}");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(HandleHttpResponse.HTTP_CONTEXT_ID, "my-id");
+ attributes.put("my-attr", "hello");
+ attributes.put("status.code", "201");
+
+ runner.enqueue("hello".getBytes(), attributes);
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(HandleHttpResponse.REL_SUCCESS, 1);
+
+ assertEquals("hello", contextMap.baos.toString());
+ assertEquals("hello", contextMap.headersSent.get("my-attr"));
+ assertNull(contextMap.headersSent.get("no-valid-attr"));
+ assertEquals(201, contextMap.statusCode);
+ assertEquals(1, contextMap.getCompletionCount());
+ assertTrue(contextMap.headersWithNoValue.isEmpty());
+ }
+
+
+ private static class MockHttpContextMap extends AbstractControllerService implements HttpContextMap {
+ private final String id;
+ private final AtomicInteger completedCount = new AtomicInteger(0);
+ private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ private final ConcurrentMap<String, String> headersSent = new ConcurrentHashMap<>();
+ private volatile int statusCode = -1;
+
+ private final List<String> headersWithNoValue = new CopyOnWriteArrayList<>();
+
+ public MockHttpContextMap(final String expectedIdentifier) {
+ this.id = expectedIdentifier;
+ }
+
+ @Override
+ public boolean register(String identifier, HttpServletRequest request, HttpServletResponse response, AsyncContext context) {
+ return true;
+ }
+
+ @Override
+ public HttpServletResponse getResponse(final String identifier) {
+ if ( !id.equals(identifier) ) {
+ Assert.fail("attempting to respond to wrong request; should have been " + id + " but was " + identifier);
+ }
+
+ try {
+ final HttpServletResponse response = Mockito.mock(HttpServletResponse.class);
+ Mockito.when(response.getOutputStream()).thenReturn(new ServletOutputStream() {
+ @Override
+ public boolean isReady() { return true; }
+
+ @Override
+ public void setWriteListener(WriteListener writeListener) {}
+
+ @Override
+ public void write(int b) throws IOException { baos.write(b); }
+
+ @Override
+ public void write(byte[] b) throws IOException { baos.write(b); }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException { baos.write(b, off, len); }
+ });
+
+
+ Mockito.doAnswer(new Answer<Object>() {
+ @Override
+ public Object answer(final InvocationOnMock invocation) throws Throwable {
+ final String key = invocation.getArgumentAt(0, String.class);
+ final String value = invocation.getArgumentAt(1, String.class);
+ if ( value == null ) {
+ headersWithNoValue.add(key);
+ } else {
+ headersSent.put(key, value);
+ }
+
+ return null;
+ }
+ }).when(response).setHeader(Mockito.any(String.class), Mockito.any(String.class));
+
+ Mockito.doAnswer(new Answer<Object>() {
+ @Override
+ public Object answer(final InvocationOnMock invocation) throws Throwable {
+ statusCode = invocation.getArgumentAt(0, int.class);
+ return null;
+ }
+ }).when(response).setStatus(Mockito.anyInt());
+
+ return response;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ Assert.fail(e.toString());
+ return null;
+ }
+ }
+
+ @Override
+ public void complete(final String identifier) {
+ if ( !id.equals(identifier) ) {
+ Assert.fail("attempting to respond to wrong request; should have been " + id + " but was " + identifier);
+ }
+
+ completedCount.incrementAndGet();
+ }
+
+ public int getCompletionCount() {
+ return completedCount.get();
+ }
+ }
+}
[3/5] incubator-nifi git commit: NIFI-221: Fixed shutdown of
HandleHttpRequest and provided documentation
Posted by ma...@apache.org.
NIFI-221: Fixed shutdown of HandleHttpRequest and provided documentation
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/c53b0f9d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/c53b0f9d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/c53b0f9d
Branch: refs/heads/develop
Commit: c53b0f9d15390ccee1c8011e6a05db82ad02f74c
Parents: 2d48323
Author: Mark Payne <ma...@hotmail.com>
Authored: Sat Feb 28 14:55:56 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Sat Feb 28 14:55:56 2015 -0500
----------------------------------------------------------------------
.../processors/standard/HandleHttpRequest.java | 1 +
.../index.html | 255 +++++++++++++++++++
.../index.html | 112 ++++++++
3 files changed, 368 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/c53b0f9d/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
index 85d7433..98b5fdd 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HandleHttpRequest.java
@@ -397,6 +397,7 @@ public class HandleHttpRequest extends AbstractProcessor {
if ( server != null ) {
getLogger().debug("Shutting down server");
server.stop();
+ server.destroy();
server.join();
getLogger().info("Shut down {}", new Object[] {server});
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/c53b0f9d/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpRequest/index.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpRequest/index.html b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpRequest/index.html
new file mode 100644
index 0000000..d3da666
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpRequest/index.html
@@ -0,0 +1,255 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>HandleHttpRequest</title>
+ <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>
+ This processor starts an HTTP server and creates a FlowFile for each HTTP Request that it receives. The Processor leaves
+ the HTTP Connection open and is intended to be used in conjunction with a
+ <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a> Processor.
+ </p>
+
+ <p>
+ The pairing of this Processor with a <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a> Processor
+ provides the ability to use NiFi to visually construct a web server that can carry out any functionality that is available
+ through the existing Processors. For example, one could construct a Web-based front end to an SFTP Server by constructing a
+ flow such as:
+ </p>
+
+ <p>
+ <a href="index.html">HandleHttpRequest</a> ->
+ <a href="../org.apache.nifi.processors.standard.PutSFTP/index.html">PutSFTP</a> ->
+ <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a>
+ </p>
+
+ <p>
+ The HandleHttpRequest Processor provides several Properties to configure which methods are supported, the paths that are
+ supported, and SSL configuration. The FlowFiles that are generated by this Processor have the following attributes added to
+ them, providing powerful routing capabilities and traceability of all data:
+ </p>
+
+ <table>
+ <thead>
+ <th>Attribute Name</th>
+ <th>Attribute Description</th>
+ </thead>
+ <tbody>
+ <tr>
+ <td>http.context.identifier</td>
+ <td>An identifier that allows the HandleHttpRequest and HandleHttpResponse to coordinate which FlowFile belongs
+ to which HTTP Request/Response.</td>
+ </tr>
+ <tr>
+ <td>mime.type</td>
+ <td>The MIME Type of the data, according to the HTTP Header "Content-Type"</td>
+ </tr>
+ <tr>
+ <td>http.servlet.path</td>
+ <td>The part of the request URL that is considered the Servlet Path</td>
+ </tr>
+ <tr>
+ <td>http.context.path</td>
+ <td>The part of the request URL that is considered to be the Context Path</td>
+ </tr>
+ <tr>
+ <td>http.method</td>
+ <td>The HTTP Method that was used for the request, such as GET or POST</td>
+ </tr>
+ <tr>
+ <td>http.query.string</td>
+ <td>The query string portion of hte Request URL</td>
+ </tr>
+ <tr>
+ <td>http.remote.host</td>
+ <td>The hostname of the requestor</td>
+ </tr>
+ <tr>
+ <td>http.remote.addr</td>
+ <td>The hostname:port combination of the requestor</td>
+ </tr>
+ <tr>
+ <td>http.remote.user</td>
+ <td>The username of the requestor</td>
+ </tr>
+ <tr>
+ <td>http.request.uri</td>
+ <td>The full Request URL</td>
+ </tr>
+ <tr>
+ <td>http.auth.type</td>
+ <td>The type of HTTP Authorization used</td>
+ </tr>
+ <tr>
+ <td>http.principal.name</td>
+ <td>The name of the authenticated user making the request</td>
+ </tr>
+ <tr>
+ <td>http.subject.dn</td>
+ <td>The Distinguished Name of the requestor. This value will not be populated unless the Processor is
+ configured to use an SSLContext Service</td>
+ </tr>
+ <tr>
+ <td>http.issuer.dn</td>
+ <td>The Distinguished Name of the entity that issued the Subject's certificate. This value will not be
+ populated unless the Processor is configured to use an SSLContext Service</td>
+ </tr>
+ <tr>
+ <td>http.headers.XXX</td>
+ <td>Each of the HTTP Headers that is received in the request will be added as an attribute, prefixed
+ with "http.headers." For example, if the request contains an HTTP Header named "x-my-header",
+ then the value will be added to an attribute named "http.headers.x-my-header"</td>
+ </tr>
+ </tbody>
+ </table>
+
+
+ <p>
+ <strong>Properties:</strong>
+ </p>
+ <p>
+ In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. If a property has a default value, it is indicated. If a property supports the use of the NiFi Expression Language (or simply, "expression language"), that is also indicated.
+ </p>
+ <ul>
+ <li><strong>Listening Port</strong>
+ <ul>
+ <li>The port to listen on for incoming HTTP Requests</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li>Hostname
+ <ul>
+ <li>The Hostname to bind to. If not specified, will bind to all hosts</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>HTTP Context Map</strong>
+ <ul>
+ <li>The HTTP Context Map Controller Service to use for caching the HTTP Request Information</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li>SSL Context Service
+ <ul>
+ <li>The Controller Service to use for obtaining an SSL Context. The SSL Context controller service is a mechanism for providing all the security properties that allow for secure communications between NiFi extensions and other systems. See the User Guide or the Controller Services documentation via the "help" link in the upper-right corner of the GUI for more information about the StandardSSLContextService. The value for this property is the identifier name that is configured in the StandardSSLContextService.</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+
+ <li><strong>Allowed Paths</strong>
+ <ul>
+ <li>A Regular Expression that specifies the valid HTTP Paths that are allowed in the incoming URL Requests. If this value is specified and the path of the HTTP Requests does not match this Regular Expression, the Processor will respond with a 404: NotFound</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>Allow GET</strong>
+ <ul>
+ <li>Specifies whether or not to allow HTTP GET Method</li>
+ <li>Default value: true</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>Allow POST</strong>
+ <ul>
+ <li>Specifies whether or not to allow HTTP POST Method</li>
+ <li>Default value: true</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>Allow PUT</strong>
+ <ul>
+ <li>Specifies whether or not to allow HTTP PUT Method</li>
+ <li>Default value: true</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>Allow DELETE</strong>
+ <ul>
+ <li>Specifies whether or not to allow HTTP DELETE Method</li>
+ <li>Default value: true</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>Allow HEAD</strong>
+ <ul>
+ <li>Specifies whether or not to allow HTTP HEAD Method</li>
+ <li>Default value: false</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>Allow OPTIONS</strong>
+ <ul>
+ <li>Specifies whether or not to allow HTTP OPTIONS Method</li>
+ <li>Default value: false</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li>Additional HTTP Methods
+ <ul>
+ <li>A comma-separated list of non-standard HTTP Methods that should be allowed</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ <li><strong>Client Authentication</strong>
+ <ul>
+ <li>Specifies whether or not the Processor should authenticate clients. This value is ignored if the <SSL Context Service> Property is not specified or the SSL Context provided uses only a KeyStore and not a TrustStore.</li>
+ <li>
+ The following values are allowed:
+ <ul>
+ <li><b>No Authentication</b> - Processor will not authenticate clients. Anyone can communicate with this Processor anonymously</li>
+ <li><b>Want Authentication</b> - Processor will try to verify the client but if unable to verify will allow the client to communicate anonymously</li>
+ <li><b>Need Authentication</b> - Processor will reject communications from any client unless the client provides a certificate that is trusted by the TrustStore specified in the SSL Context Service</li>
+ </ul>
+ </li>
+ <li>Default value: No Authentication</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ </ul>
+
+ <p>
+ <strong>Relationships:</strong>
+ </p>
+ <ul>
+ <li>success
+ <ul>
+ <li>All FlowFiles that are created are routed to this relationship.</li>
+ </ul>
+ </li>
+ </ul>
+
+
+ <p>
+ <strong>See Also:</strong><br />
+ <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a><br />
+ <a href="../org.apache.nifi.http.StandardHttpContextMap/index.html">StandardHttpContextMap</a><br />
+ <a href="../org.apache.nifi.ssl.StandardSSLContextService/index.html">StandardSSLContextService</a><br />
+ </p>
+
+ </body>
+</html>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/c53b0f9d/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpResponse/index.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpResponse/index.html b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpResponse/index.html
new file mode 100644
index 0000000..70d76a6
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.HandleHttpResponse/index.html
@@ -0,0 +1,112 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8" />
+ <title>HandleHttpResponse</title>
+ <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
+ </head>
+
+ <body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>
+ This processor responds to an HTTP request that was received by the
+ <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a> Processor.
+ </p>
+
+ <p>
+ The pairing of this Processor with a <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a> Processor
+ provides the ability to use NiFi to visually construct a web server that can carry out any functionality that is available
+ through the existing Processors. For example, one could construct a Web-based front end to an SFTP Server by constructing a
+ flow such as:
+ </p>
+
+ <p>
+ <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a> ->
+ <a href="../org.apache.nifi.processors.standard.PutSFTP/index.html">PutSFTP</a> ->
+ <a href="index.html">HandleHttpResponse</a>
+ </p>
+
+ <p>
+ This Processor must be configured with the same <HTTP Context Map> service as the corresponding HandleHttpRequest Processor.
+ Otherwise, all FlowFiles will be routed to the 'failure' relationship.
+ </p>
+
+ <p>
+ All FlowFiles must have an attribute named <code>http.context.identifier</code>. The value of this attribute is used to lookup
+ the HTTP Response so that the proper message can be sent back to the requestor. If this attribute is missing, the FlowFile
+ will be routed to 'failure.'
+ </p>
+
+
+ <p>
+ <strong>Properties:</strong>
+ </p>
+ <p>
+ In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. If a property has a default value, it is indicated. If a property supports the use of the NiFi Expression Language (or simply, "expression language"), that is also indicated.
+ </p>
+ <ul>
+ <li><strong>HTTP Status Code</strong>
+ <ul>
+ <li>The HTTP Status Code to use when responding to the HTTP Request. See Section 10 of RFC 2616 for more information.</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: true</li>
+ </ul>
+ </li>
+ <li><strong>HTTP Context Map</strong>
+ <ul>
+ <li>The HTTP Context Map Controller Service to use for caching the HTTP Request Information</li>
+ <li>Default value: no default</li>
+ <li>Supports expression language: false</li>
+ </ul>
+ </li>
+ </ul>
+
+ <p>
+ <strong>Relationships:</strong>
+ </p>
+ <ul>
+ <li>success
+ <ul>
+ <li>If a message is successfully sent back to the requestor, the FlowFile is routed to this relationship.</li>
+ </ul>
+ </li>
+
+ <li>failure
+ <ul>
+ <li>
+ A FlowFile will be routed to failure under the following conditions:
+ <ul>
+ <li>The FlowFile does not have an <code>http.context.identifier</code> attribute</li>
+ <li>The <code>http.context.identifier</code> attribute has a value that cannot be found in the HTTP Context Map</li>
+ <li>The HTTP Status Code is not a number</li>
+ <li>There was a communications failure when attempting to write the response to the requestor</li>
+ </ul>
+ </li>
+ </ul>
+ </li>
+ </ul>
+
+ <p>
+ <strong>See Also:</strong><br />
+ <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a><br />
+ <a href="../org.apache.nifi.http.StandardHttpContextMap/index.html">StandardHttpContextMap</a><br />
+ <a href="../org.apache.nifi.ssl.StandardSSLContextService/index.html">StandardSSLContextService</a><br />
+ </p>
+
+ </body>
+</html>