Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-271 more cleanup

This commit is contained in:
joewitt 2015-04-22 23:42:19 -04:00
parent 7f9dff52b0
commit a53cc3d707
58 changed files with 1015 additions and 1118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
nifi-parent/pom.xml
View File

@ -372,6 +372,9 @@
<profiles>
<profile> <!-- will move this up with the always on plugins once we get all checkstyle stuff resolved-->
<id>checkstyle</id>
<activation>
<activeByDefault>false</activeByDefault>
</activation>
<build>
<plugins>
<plugin>

127
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/main/java/org/apache/nifi/web/server/JettyServer.java
View File

@ -102,11 +102,11 @@ public class JettyServer implements NiFiServer {
private final Server server;
private final NiFiProperties props;
private ExtensionMapping extensionMapping;
private WebAppContext webApiContext;
private WebAppContext webDocsContext;
// content viewer and mime type specific extensions
private WebAppContext webContentViewerContext;
private Collection<WebAppContext> contentViewerWebContexts;
@ -114,10 +114,10 @@ public class JettyServer implements NiFiServer {
// component (processor, controller service, reporting task) ui extensions
private UiExtensionMapping componentUiExtensions;
private Collection<WebAppContext> componentUiExtensionWebContexts;
@Deprecated
private Collection<WebAppContext> customUiWebContexts;
/**
* Creates and configures a new Jetty instance.
*
@ -138,11 +138,6 @@ public class JettyServer implements NiFiServer {
loadWars(locateNarWorkingDirectories());
}
/**
* Locates the working directory for each NAR.
*
* @return
*/
private Set<File> locateNarWorkingDirectories() {
final File frameworkWorkingDir = props.getFrameworkWorkingDirectory();
final File extensionsWorkingDir = props.getExtensionsWorkingDirectory();
@ -168,7 +163,7 @@ public class JettyServer implements NiFiServer {
* Loads the WARs in the specified NAR working directories. A WAR file must
* have a ".war" extension.
*
* @param warDir a directory containing WARs to load
* @param narWorkingDirectories dirs
*/
private void loadWars(final Set<File> narWorkingDirectories) {
@ -219,14 +214,14 @@ public class JettyServer implements NiFiServer {
@Deprecated
final Map<String, String> customUiMappings = new HashMap<>();
// deploy the other wars
if (CollectionUtils.isNotEmpty(otherWars)) {
// hold onto to the web contexts for all ui extensions
customUiWebContexts = new ArrayList<>();
componentUiExtensionWebContexts = new ArrayList<>();
contentViewerWebContexts = new ArrayList<>();
// ui extension organized by component type
final Map<String, List<UiExtension>> componentUiExtensionsByType = new HashMap<>();
for (File war : otherWars) {
@ -237,7 +232,7 @@ public class JettyServer implements NiFiServer {
// identify all known extension types in the war
final Map<UiExtensionType, List<String>> uiExtensionInWar = new HashMap<>();
identifyUiExtensionsForComponents(uiExtensionInWar, war);
// only include wars that are for custom processor ui's
if (!customUiProcessorTypes.isEmpty() || !uiExtensionInWar.isEmpty()) {
// get the context path
@ -258,7 +253,7 @@ public class JettyServer implements NiFiServer {
// also store it by type so we can populate the appropriate initialization parameters
if (!customUiProcessorTypes.isEmpty()) {
customUiWebContexts.add(extensionUiContext);
// @Deprecated - supported custom uis as init params to the web api
for (String customUiProcessorType : customUiProcessorTypes) {
// map the processor type to the custom ui path
@ -307,9 +302,9 @@ public class JettyServer implements NiFiServer {
// include custom ui web context in the handlers
handlers.addHandler(extensionUiContext);
}
}
// record all ui extensions to give to the web api
componentUiExtensions = new UiExtensionMapping(componentUiExtensionsByType);
} else {
@ -328,7 +323,7 @@ public class JettyServer implements NiFiServer {
webContentViewerContext = loadWar(webContentViewerWar, "/nifi-content-viewer", frameworkClassLoader);
webContentViewerContext.getInitParams().putAll(mimeMappings);
handlers.addHandler(webContentViewerContext);
// create a web app for the docs
final String docsContextPath = "/nifi-docs";
@ -348,12 +343,6 @@ public class JettyServer implements NiFiServer {
server.setHandler(handlers);
}
/**
* Finds WAR files in the specified NAR working directories.
*
* @param narWorkingDirectories
* @return
*/
private Map<File, File> findWars(final Set<File> narWorkingDirectories) {
final Map<File, File> wars = new HashMap<>();
@ -381,7 +370,7 @@ public class JettyServer implements NiFiServer {
if (jarEntry == null) {
return;
}
// get an input stream for the nifi-processor configuration file
BufferedReader in = new BufferedReader(new InputStreamReader(jarFile.getInputStream(jarEntry)));
@ -392,24 +381,24 @@ public class JettyServer implements NiFiServer {
final String componentType = extractComponentType(rawComponentType);
if (componentType != null) {
List<String> extensions = uiExtensions.get(uiExtensionType);
// if there are currently no extensions for this type create it
if (extensions == null) {
extensions = new ArrayList<>();
uiExtensions.put(uiExtensionType, extensions);
}
// add the specified type
extensions.add(componentType);
}
}
}
/**
* Identifies all known UI extensions and stores them in the specified map.
*
* @param uiExtensions
* @param warFile
*
* @param uiExtensions extensions
* @param warFile war
*/
private void identifyUiExtensionsForComponents(final Map<UiExtensionType, List<String>> uiExtensions, final File warFile) {
try (final JarFile jarFile = new JarFile(warFile)) {
@ -422,12 +411,13 @@ public class JettyServer implements NiFiServer {
logger.warn(String.format("Unable to inspect %s for a UI extensions.", warFile));
}
}
/**
* Extracts the component type. Trims the line and considers comments. Returns null if no type was found.
*
* @param line
* @return
* Extracts the component type. Trims the line and considers comments.
* Returns null if no type was found.
*
* @param line line
* @return type
*/
private String extractComponentType(final String line) {
final String trimmedLine = line.trim();
@ -437,12 +427,13 @@ public class JettyServer implements NiFiServer {
}
return null;
}
/**
* Returns the extension in the specified WAR using the specified path.
*
* @param war
* @return
* @param war war
* @param path path
* @return extensions
*/
private List<String> getWarExtensions(final File war, final String path) {
List<String> processorTypes = new ArrayList<>();
@ -529,10 +520,10 @@ public class JettyServer implements NiFiServer {
final ResourceHandler resourceHandler = new ResourceHandler();
resourceHandler.setDirectoriesListed(false);
// load the docs directory
// load the docs directory
final File docsDir = Paths.get("docs").toRealPath().toFile();
final Resource docsResource = Resource.newResource(docsDir);
// load the component documentation working directory
final String componentDocsDirPath = props.getProperty(NiFiProperties.COMPONENT_DOCS_DIRECTORY, "work/docs/components");
final File workingDocsDirectory = Paths.get(componentDocsDirPath).toRealPath().getParent().toFile();
@ -547,7 +538,7 @@ public class JettyServer implements NiFiServer {
}
}
final Resource webApiDocsResource = Resource.newResource(webApiDocsDir);
// create resources for both docs locations
final ResourceCollection resources = new ResourceCollection(docsResource, workingDocsResource, webApiDocsResource);
resourceHandler.setBaseResource(resources);
@ -660,9 +651,6 @@ public class JettyServer implements NiFiServer {
return contextFactory;
}
/**
* Starts the web server.
*/
@Override
public void start() {
try {
@ -682,21 +670,21 @@ public class JettyServer implements NiFiServer {
}
}
}
// ensure the appropriate wars deployed successfully before injecting the NiFi context and security filters -
// ensure the appropriate wars deployed successfully before injecting the NiFi context and security filters
// this must be done after starting the server (and ensuring there were no start up failures)
if (webApiContext != null) {
// give the web api the component ui extensions
final ServletContext webApiServletContext = webApiContext.getServletHandler().getServletContext();
webApiServletContext.setAttribute("nifi-ui-extensions", componentUiExtensions);
// get the application context
final WebApplicationContext webApplicationContext = WebApplicationContextUtils.getRequiredWebApplicationContext(webApiServletContext);
// @Deprecated
if (CollectionUtils.isNotEmpty(customUiWebContexts)) {
final NiFiWebContext niFiWebContext = webApplicationContext.getBean("nifiWebContext", NiFiWebContext.class);
for (final WebAppContext customUiContext : customUiWebContexts) {
// set the NiFi context in each custom ui servlet context
final ServletContext customUiServletContext = customUiContext.getServletHandler().getServletContext();
@ -709,7 +697,7 @@ public class JettyServer implements NiFiServer {
}
}
}
// component ui extensions
if (CollectionUtils.isNotEmpty(componentUiExtensionWebContexts)) {
final NiFiWebConfigurationContext configurationContext = webApplicationContext.getBean("nifiWebConfigurationContext", NiFiWebConfigurationContext.class);
@ -726,7 +714,7 @@ public class JettyServer implements NiFiServer {
}
}
}
// content viewer extensions
if (CollectionUtils.isNotEmpty(contentViewerWebContexts)) {
for (final WebAppContext contentViewerContext : contentViewerWebContexts) {
@ -737,35 +725,35 @@ public class JettyServer implements NiFiServer {
}
}
}
// content viewer controller
if (webContentViewerContext != null) {
final ContentAccess contentAccess = webApplicationContext.getBean("contentAccess", ContentAccess.class);
// add the content access
final ServletContext webContentViewerServletContext = webContentViewerContext.getServletHandler().getServletContext();
webContentViewerServletContext.setAttribute("nifi-content-access", contentAccess);
final FilterHolder securityFilter = webApiContext.getServletHandler().getFilter("springSecurityFilterChain");
if (securityFilter != null) {
webContentViewerContext.addFilter(securityFilter, "/*", EnumSet.allOf(DispatcherType.class));
}
}
}
// ensure the web document war was loaded and provide the extension mapping
if (webDocsContext != null) {
final ServletContext webDocsServletContext = webDocsContext.getServletHandler().getServletContext();
webDocsServletContext.setAttribute("nifi-extension-mapping", extensionMapping);
}
// if this nifi is a node in a cluster, start the flow service and load the flow - the
// flow service is loaded here for clustered nodes because the loading of the flow will
// initialize the connection between the node and the NCM. if the node connects (starts
// heartbeating, etc), the NCM may issue web requests before the application (wars) have
// finished loading. this results in the node being disconnected since its unable to
// successfully respond to the requests. to resolve this, flow loading was moved to here
// (after the wars have been successfully deployed) when this nifi instance is a node
// if this nifi is a node in a cluster, start the flow service and load the flow - the
// flow service is loaded here for clustered nodes because the loading of the flow will
// initialize the connection between the node and the NCM. if the node connects (starts
// heartbeating, etc), the NCM may issue web requests before the application (wars) have
// finished loading. this results in the node being disconnected since its unable to
// successfully respond to the requests. to resolve this, flow loading was moved to here
// (after the wars have been successfully deployed) when this nifi instance is a node
// in a cluster
if (props.isNode()) {
@ -799,11 +787,6 @@ public class JettyServer implements NiFiServer {
}
}
/**
* Dump each applicable url.
*
* @throws SocketException
*/
private void dumpUrls() throws SocketException {
final List<String> urls = new ArrayList<>();
@ -853,11 +836,6 @@ public class JettyServer implements NiFiServer {
}
}
/**
* Handles when a start up failure occurs.
*
* @param t
*/
private void startUpFailure(Throwable t) {
System.err.println("Failed to start web server: " + t.getMessage());
System.err.println("Shutting down...");
@ -870,9 +848,6 @@ public class JettyServer implements NiFiServer {
this.extensionMapping = extensionMapping;
}
/**
* Stops the web server.
*/
@Override
public void stop() {
try {
@ -881,4 +856,4 @@ public class JettyServer implements NiFiServer {
logger.warn("Failed to stop web server", ex);
}
}
}
}

8
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-ui-extension/src/main/java/org/apache/nifi/ui/extension/UiExtension.java
View File

@ -32,18 +32,14 @@ public class UiExtension {
}
/**
* The type of this UI extension.
*
* @return
* @return type of this UI extension
*/
public UiExtensionType getExtensionType() {
return extensionType;
}
/**
* The context path of this UI extenion.
*
* @return
* @return The context path of this UI extension
*/
public String getContextPath() {
return contextPath;

17
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-ui-extension/src/main/java/org/apache/nifi/ui/extension/UiExtensionMapping.java
View File

@ -31,22 +31,21 @@ public class UiExtensionMapping {
}
/**
* Returns whether there are any UI extensions for the specified component type.
*
* @param type
* @return
* @param type type
* @return whether there are any UI extensions for the specified component
* type
*/
public boolean hasUiExtension(final String type) {
return uiExtensions.containsKey(type);
}
/**
* Gets the listing of all discovered UI extensions for the specified component type.
* @param type
* @return
* @param type type
* @return the listing of all discovered UI extensions for the specified
* component type
*/
public List<UiExtension> getUiExtension(final String type) {
return uiExtensions.get(type);
}
}

25
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/ControllerServiceAuditor.java
View File

@ -230,35 +230,35 @@ public class ControllerServiceAuditor extends NiFiAuditor {
public Object updateControllerServiceReferenceAdvice(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
// update the controller service references
final ControllerServiceReference controllerServiceReference = (ControllerServiceReference) proceedingJoinPoint.proceed();
// get the current user
final NiFiUser user = NiFiUserUtils.getNiFiUser();
if (user != null) {
final Collection<Action> actions = new ArrayList<>();
final Collection<String> visitedServices = new ArrayList<>();
visitedServices.add(controllerServiceReference.getReferencedComponent().getIdentifier());
// get all applicable actions
getUpdateActionsForReferencingComponents(user, actions, visitedServices, controllerServiceReference.getReferencingComponents());
// ensure there are actions to record
if (!actions.isEmpty()) {
// save the actions
saveActions(actions, logger);
}
}
return controllerServiceReference;
}
/**
* Gets the update actions for all specified referencing components.
*
*
* @param user
* @param actions
* @param visitedServices
* @param referencingComponents
* @param referencingComponents
*/
private void getUpdateActionsForReferencingComponents(final NiFiUser user, final Collection<Action> actions, final Collection<String> visitedServices, final Set<ConfiguredComponent> referencingComponents) {
// consider each component updates
@ -325,7 +325,7 @@ public class ControllerServiceAuditor extends NiFiAuditor {
}
}
}
/**
* Audits the removal of a controller service via deleteControllerService().
*
@ -464,10 +464,11 @@ public class ControllerServiceAuditor extends NiFiAuditor {
}
/**
* Returns whether the specified controller service is disabled (or disabling).
*
* Returns whether the specified controller service is disabled (or
* disabling).
*
* @param controllerService
* @return
* @return
*/
private boolean isDisabled(final ControllerServiceNode controllerService) {
return ControllerServiceState.DISABLED.equals(controllerService.getState()) || ControllerServiceState.DISABLING.equals(controllerService.getState());

4
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/FunnelAuditor.java
View File

@ -99,7 +99,7 @@ public class FunnelAuditor extends NiFiAuditor {
*
* @param funnel
* @param operation
* @return
* @return
*/
public Action generateAuditRecord(Funnel funnel, Operation operation) {
return generateAuditRecord(funnel, operation, null);
@ -111,7 +111,7 @@ public class FunnelAuditor extends NiFiAuditor {
* @param funnel
* @param operation
* @param actionDetails
* @return
* @return
*/
public Action generateAuditRecord(Funnel funnel, Operation operation, ActionDetails actionDetails) {
Action action = null;

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/NiFiAuditor.java
View File

@ -58,7 +58,7 @@ public abstract class NiFiAuditor {
*/
protected void saveActions(Collection<Action> actions, Logger logger) {
ClusterContext ctx = ClusterContextThreadLocal.getContext();
// if we're a connected node, then put audit actions on threadlocal to propagate back to manager
if (ctx != null) {
ctx.getActions().addAll(actions);

4
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/PortAuditor.java
View File

@ -295,7 +295,7 @@ public class PortAuditor extends NiFiAuditor {
*
* @param port
* @param operation
* @return
* @return
*/
public Action generateAuditRecord(Port port, Operation operation) {
return generateAuditRecord(port, operation, null);
@ -307,7 +307,7 @@ public class PortAuditor extends NiFiAuditor {
* @param port
* @param operation
* @param actionDetails
* @return
* @return
*/
public Action generateAuditRecord(Port port, Operation operation, ActionDetails actionDetails) {
Action action = null;

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/ProcessGroupAuditor.java
View File

@ -55,7 +55,7 @@ public class ProcessGroupAuditor extends NiFiAuditor {
* alleviate this issue.
*
* @param proceedingJoinPoint
* @return
* @return
* @throws java.lang.Throwable
*/
@Around("within(org.apache.nifi.web.dao.ProcessGroupDAO+) && "

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/ProcessorAuditor.java
View File

@ -79,7 +79,7 @@ public class ProcessorAuditor extends NiFiAuditor {
* alleviate this issue.
*
* @param proceedingJoinPoint
* @return
* @return
* @throws java.lang.Throwable
*/
@Around("within(org.apache.nifi.web.dao.ProcessorDAO+) && "

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/RelationshipAuditor.java
View File

@ -77,7 +77,7 @@ public class RelationshipAuditor extends NiFiAuditor {
* alleviate this issue.
*
* @param proceedingJoinPoint
* @return
* @return
* @throws java.lang.Throwable
*/
@Around("within(org.apache.nifi.web.dao.ConnectionDAO+) && "

8
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/RemoteProcessGroupAuditor.java
View File

@ -61,8 +61,8 @@ public class RemoteProcessGroupAuditor extends NiFiAuditor {
* alleviate this issue.
*
* @param proceedingJoinPoint
* @return
* @throws java.lang.Throwable
* @return
* @throws java.lang.Throwable
*/
@Around("within(org.apache.nifi.web.dao.RemoteProcessGroupDAO+) && "
+ "execution(org.apache.nifi.groups.RemoteProcessGroup createRemoteProcessGroup(java.lang.String, org.apache.nifi.web.api.dto.RemoteProcessGroupDTO))")
@ -331,7 +331,7 @@ public class RemoteProcessGroupAuditor extends NiFiAuditor {
*
* @param remoteProcessGroup
* @param operation
* @return
* @return
*/
public Action generateAuditRecord(RemoteProcessGroup remoteProcessGroup, Operation operation) {
return generateAuditRecord(remoteProcessGroup, operation, null);
@ -343,7 +343,7 @@ public class RemoteProcessGroupAuditor extends NiFiAuditor {
* @param remoteProcessGroup
* @param operation
* @param actionDetails
* @return
* @return
*/
public Action generateAuditRecord(RemoteProcessGroup remoteProcessGroup, Operation operation, ActionDetails actionDetails) {
Action action = null;

3
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/ReportingTaskAuditor.java
View File

@ -297,7 +297,8 @@ public class ReportingTaskAuditor extends NiFiAuditor {
}
/**
* Extracts the values for the configured properties from the specified ReportingTask.
* Extracts the values for the configured properties from the specified
* ReportingTask.
*
* @param reportingTask
* @param reportingTaskDTO

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/audit/SnippetAuditor.java
View File

@ -258,7 +258,7 @@ public class SnippetAuditor extends NiFiAuditor {
* @param proceedingJoinPoint
* @param snippetDTO
* @param snippetDAO
* @return
* @return
* @throws Throwable
*/
@Around("within(org.apache.nifi.web.dao.SnippetDAO+) && "

496
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
View File

File diff suppressed because it is too large Load Diff

18
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiContentAccess.java
View File

@ -50,11 +50,11 @@ public class StandardNiFiContentAccess implements ContentAccess {
private static final Logger logger = LoggerFactory.getLogger(StandardNiFiContentAccess.class);
public static final String CLIENT_ID_PARAM = "clientId";
private NiFiProperties properties;
private NiFiServiceFacade serviceFacade;
private WebClusterManager clusterManager;
@Override
@PreAuthorize("hasRole('ROLE_PROVENANCE')")
public DownloadableContent getContent(final ContentRequestContext request) {
@ -67,11 +67,11 @@ public class StandardNiFiContentAccess implements ContentAccess {
} catch (final URISyntaxException use) {
throw new ClusterRequestException(use);
}
// set the request parameters
final MultivaluedMap<String, String> parameters = new MultivaluedMapImpl();
parameters.add(CLIENT_ID_PARAM, request.getClientId());
// set the headers
final Map<String, String> headers = new HashMap<>();
if (StringUtils.isNotBlank(request.getProxiedEntitiesChain())) {
@ -90,7 +90,7 @@ public class StandardNiFiContentAccess implements ContentAccess {
headers.put("X-ProxiedEntityUserDetails", hexEncodedUserDetails);
}
}
// get the target node and ensure it exists
final Node targetNode = clusterManager.getNode(request.getClusterNodeId());
if (targetNode == null) {
@ -104,14 +104,14 @@ public class StandardNiFiContentAccess implements ContentAccess {
final NodeResponse nodeResponse = clusterManager.applyRequest(HttpMethod.GET, dataUri, parameters, headers, targetNodes);
final ClientResponse clientResponse = nodeResponse.getClientResponse();
final MultivaluedMap<String, String> responseHeaders = clientResponse.getHeaders();
// get the file name
final String contentDisposition = responseHeaders.getFirst("Content-Disposition");
final String filename = StringUtils.substringAfterLast(contentDisposition, "filename=");
// get the content type
final String contentType = responseHeaders.getFirst("Content-Type");
// create the downloadable content
return new DownloadableContent(filename, contentType, clientResponse.getEntityInputStream());
} else {
@ -119,7 +119,7 @@ public class StandardNiFiContentAccess implements ContentAccess {
final String eventDetails = StringUtils.substringAfterLast(request.getDataUri(), "events/");
final String rawEventId = StringUtils.substringBefore(eventDetails, "/content/");
final String rawDirection = StringUtils.substringAfterLast(eventDetails, "/content/");
// get the content type
final Long eventId;
final ContentDirection direction;

119
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
View File

@ -211,7 +211,6 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// -----------------------------------------
// Verification Operations
// -----------------------------------------
@Override
public void verifyCreateConnection(String groupId, ConnectionDTO connectionDTO) {
connectionDAO.verifyCreate(groupId, connectionDTO);
@ -365,21 +364,20 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// -----------------------------------------
// Write Operations
// -----------------------------------------
@Override
public ConfigurationSnapshot<ConnectionDTO> updateConnection(final Revision revision, final String groupId, final ConnectionDTO connectionDTO) {
// if connection does not exist, then create new connection
if (connectionDAO.hasConnection(groupId, connectionDTO.getId()) == false) {
return createConnection(revision, groupId, connectionDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<ConnectionDTO>() {
@Override
public ConnectionDTO execute() {
final Connection connection = connectionDAO.updateConnection(groupId, connectionDTO);
controllerFacade.save();
return dtoFactory.createConnectionDto(connection);
}
});
@ -391,7 +389,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (processorDAO.hasProcessor(groupId, processorDTO.getId()) == false) {
return createProcessor(revision, groupId, processorDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<ProcessorDTO>() {
@Override
public ProcessorDTO execute() {
@ -400,7 +398,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createProcessorDto(processor);
}
});
@ -412,7 +410,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (labelDAO.hasLabel(groupId, labelDTO.getId()) == false) {
return createLabel(revision, groupId, labelDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<LabelDTO>() {
@Override
public LabelDTO execute() {
@ -421,7 +419,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createLabelDto(label);
}
});
@ -433,7 +431,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (funnelDAO.hasFunnel(groupId, funnelDTO.getId()) == false) {
return createFunnel(revision, groupId, funnelDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<FunnelDTO>() {
@Override
public FunnelDTO execute() {
@ -442,7 +440,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createFunnelDto(funnel);
}
});
@ -463,7 +461,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (snippetDAO.hasSnippet(snippetDto.getId()) == false) {
return createSnippet(revision, snippetDto);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<SnippetDTO>() {
@Override
public SnippetDTO execute() {
@ -478,7 +476,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (snippetDto.getParentGroupId() != null && snippet.isLinked()) {
controllerFacade.save();
}
return responseSnippetDto;
}
});
@ -490,7 +488,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (inputPortDAO.hasPort(groupId, inputPortDTO.getId()) == false) {
return createInputPort(revision, groupId, inputPortDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<PortDTO>() {
@Override
public PortDTO execute() {
@ -498,8 +496,8 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createPortDto(inputPort);
return dtoFactory.createPortDto(inputPort);
}
});
}
@ -510,7 +508,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (outputPortDAO.hasPort(groupId, outputPortDTO.getId()) == false) {
return createOutputPort(revision, groupId, outputPortDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<PortDTO>() {
@Override
public PortDTO execute() {
@ -518,7 +516,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createPortDto(outputPort);
}
});
@ -530,7 +528,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (remoteProcessGroupDAO.hasRemoteProcessGroup(groupId, remoteProcessGroupDTO.getId()) == false) {
return createRemoteProcessGroup(revision, groupId, remoteProcessGroupDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<RemoteProcessGroupDTO>() {
@Override
public RemoteProcessGroupDTO execute() {
@ -538,7 +536,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createRemoteProcessGroupDto(remoteProcessGroup);
}
});
@ -554,7 +552,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createRemoteProcessGroupPortDto(remoteGroupPort);
}
});
@ -570,7 +568,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createRemoteProcessGroupPortDto(remoteGroupPort);
}
});
@ -586,7 +584,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
return createProcessGroup(parentGroupId, revision, processGroupDTO);
}
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<ProcessGroupDTO>() {
@Override
public ProcessGroupDTO execute() {
@ -595,7 +593,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save updated controller
controllerFacade.save();
return dtoFactory.createProcessGroupDto(processGroup);
}
});
@ -625,7 +623,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return controllerConfig;
}
});
@ -662,14 +660,14 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
@Override
public ConfigurationSnapshot<Void> deleteConnection(final Revision revision, final String groupId, final String connectionId) {
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<Void>(){
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<Void>() {
@Override
public Void execute() {
connectionDAO.deleteConnection(groupId, connectionId);
// save the flow
controllerFacade.save();
return null;
}
});
@ -685,7 +683,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return null;
}
});
@ -701,7 +699,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return null;
}
});
@ -717,7 +715,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return null;
}
});
@ -744,7 +742,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (linked) {
controllerFacade.save();
}
return null;
}
});
@ -759,7 +757,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return null;
}
});
@ -774,7 +772,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return null;
}
});
@ -789,7 +787,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return null;
}
});
@ -804,7 +802,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return null;
}
});
@ -830,7 +828,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createConnectionDto(connection);
}
});
@ -851,7 +849,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createProcessorDto(processor);
}
});
@ -872,7 +870,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createLabelDto(label);
}
});
@ -893,7 +891,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createFunnelDto(funnel);
}
});
@ -960,7 +958,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
@Override
public FlowSnippetDTO execute() {
String id = snippetId;
// ensure id is set
if (StringUtils.isBlank(id)) {
id = UUID.randomUUID().toString();
@ -974,7 +972,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return flowSnippet;
}
});
@ -994,7 +992,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
final Snippet snippet = snippetDAO.createSnippet(snippetDTO);
final SnippetDTO responseSnippetDTO = dtoFactory.createSnippetDto(snippet);
responseSnippetDTO.setContents(snippetUtils.populateFlowSnippet(snippet, false));
return responseSnippetDTO;
}
});
@ -1014,7 +1012,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createPortDto(inputPort);
}
});
@ -1034,7 +1032,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createPortDto(outputPort);
}
});
@ -1054,7 +1052,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createProcessGroupDto(processGroup);
}
});
@ -1074,7 +1072,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return dtoFactory.createRemoteProcessGroupDto(remoteProcessGroup);
}
});
@ -1136,7 +1134,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// save the flow
controllerFacade.save();
return flowSnippet;
}
});
@ -1206,7 +1204,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
} else {
controllerFacade.save();
}
return dtoFactory.createControllerServiceDto(controllerService);
}
});
@ -1218,7 +1216,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (controllerServiceDAO.hasControllerService(controllerServiceDTO.getId()) == false) {
return createControllerService(revision, controllerServiceDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<ControllerServiceDTO>() {
@Override
public ControllerServiceDTO execute() {
@ -1261,7 +1259,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
} else {
controllerFacade.save();
}
return null;
}
});
@ -1286,7 +1284,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
} else {
controllerFacade.save();
}
return dtoFactory.createReportingTaskDto(reportingTask);
}
});
@ -1298,7 +1296,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (reportingTaskDAO.hasReportingTask(reportingTaskDTO.getId()) == false) {
return createReportingTask(revision, reportingTaskDTO);
}
return optimisticLockingManager.configureFlow(revision, new ConfigurationRequest<ReportingTaskDTO>() {
@Override
public ReportingTaskDTO execute() {
@ -1330,7 +1328,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
} else {
controllerFacade.save();
}
return null;
}
});
@ -1488,7 +1486,6 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
// -----------------------------------------
// Read Operations
// -----------------------------------------
@Override
public RevisionDTO getRevision() {
return dtoFactory.createRevisionDTO(optimisticLockingManager.getLastModification());
@ -1738,12 +1735,12 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
public PropertyDescriptorDTO getProcessorPropertyDescriptor(String groupId, String id, String property) {
final ProcessorNode processor = processorDAO.getProcessor(groupId, id);
PropertyDescriptor descriptor = processor.getPropertyDescriptor(property);
// return an invalid descriptor if the processor doesn't suppor this property
if (descriptor == null) {
descriptor = new PropertyDescriptor.Builder().name(property).addValidator(Validator.INVALID).dynamic(true).build();
}
return dtoFactory.createPropertyDescriptorDto(descriptor);
}
@ -2057,15 +2054,15 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
public PropertyDescriptorDTO getControllerServicePropertyDescriptor(String id, String property) {
final ControllerServiceNode controllerService = controllerServiceDAO.getControllerService(id);
PropertyDescriptor descriptor = controllerService.getControllerServiceImplementation().getPropertyDescriptor(property);
// return an invalid descriptor if the controller service doesn't support this property
if (descriptor == null) {
descriptor = new PropertyDescriptor.Builder().name(property).addValidator(Validator.INVALID).dynamic(true).build();
}
return dtoFactory.createPropertyDescriptorDto(descriptor);
}
@Override
public Set<ControllerServiceReferencingComponentDTO> getControllerServiceReferencingComponents(String controllerServiceId) {
final ControllerServiceNode service = controllerServiceDAO.getControllerService(controllerServiceId);
@ -2090,15 +2087,15 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
public PropertyDescriptorDTO getReportingTaskPropertyDescriptor(String id, String property) {
final ReportingTaskNode reportingTask = reportingTaskDAO.getReportingTask(id);
PropertyDescriptor descriptor = reportingTask.getReportingTask().getPropertyDescriptor(property);
// return an invalid descriptor if the reporting task doesn't support this property
if (descriptor == null) {
descriptor = new PropertyDescriptor.Builder().name(property).addValidator(Validator.INVALID).dynamic(true).build();
}
return dtoFactory.createPropertyDescriptorDto(descriptor);
}
@Override
public StatusHistoryDTO getProcessGroupStatusHistory(String groupId) {
return controllerFacade.getProcessGroupStatusHistory(groupId);

104
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebConfigurationContext.java
View File

@ -68,8 +68,8 @@ import org.apache.nifi.web.api.entity.ReportingTaskEntity;
import org.apache.nifi.web.util.ClientResponseUtils;
/**
* Implements the NiFiWebConfigurationContext interface to support a context in both
* standalone and clustered environments.
* Implements the NiFiWebConfigurationContext interface to support a context in
* both standalone and clustered environments.
*/
public class StandardNiFiWebConfigurationContext implements NiFiWebConfigurationContext {
@ -99,7 +99,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (requestContext.getExtensionType() == null) {
throw new IllegalArgumentException("The UI extension type must be specified.");
}
Component componentType = null;
switch (requestContext.getExtensionType()) {
case ProcessorConfiguration:
@ -116,7 +116,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (componentType == null) {
throw new IllegalArgumentException("UI extension type must support Processor, ControllerService, or ReportingTask configuration.");
}
// - when running standalone or cluster ncm - actions from custom UIs are stored locally
// - clustered nodes do not serve custom UIs directly to users so they should never be invoking this method
final Date now = new Date();
@ -192,25 +192,25 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (requestContext.getExtensionType() == null) {
throw new IllegalArgumentException("The UI extension type must be specified.");
}
// get the component facade for interacting directly with that type of object
ComponentFacade componentFacade = null;
switch (requestContext.getExtensionType()) {
case ProcessorConfiguration:
componentFacade = new ProcessorFacade();
break;
case ControllerServiceConfiguration:
case ControllerServiceConfiguration:
componentFacade = new ControllerServiceFacade();
break;
case ReportingTaskConfiguration:
componentFacade = new ReportingTaskFacade();
break;
}
if (componentFacade == null) {
throw new IllegalArgumentException("UI extension type must support Processor, ControllerService, or ReportingTask configuration.");
}
return componentFacade.getComponentDetails(requestContext);
}
@ -224,7 +224,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (StringUtils.isBlank(id)) {
throw new ResourceNotFoundException(String.format("Configuration request context did not have a component ID."));
}
// ensure the path could be
if (requestContext.getExtensionType() == null) {
throw new IllegalArgumentException("The UI extension type must be specified.");
@ -236,18 +236,18 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
case ProcessorConfiguration:
componentFacade = new ProcessorFacade();
break;
case ControllerServiceConfiguration:
case ControllerServiceConfiguration:
componentFacade = new ControllerServiceFacade();
break;
case ReportingTaskConfiguration:
componentFacade = new ReportingTaskFacade();
break;
}
if (componentFacade == null) {
throw new IllegalArgumentException("UI extension type must support Processor, ControllerService, or ReportingTask configuration.");
}
return componentFacade.setAnnotationData(requestContext, annotationData);
}
@ -255,32 +255,34 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
* Facade over accessing different types of NiFi components.
*/
private interface ComponentFacade {
/**
* Gets the component details using the specified request context.
*
*
* @param requestContext
* @return
* @return
*/
ComponentDetails getComponentDetails(NiFiWebRequestContext requestContext);
/**
* Sets the annotation data using the specified request context.
*
*
* @param requestContext
* @param annotationData
* @return
* @return
*/
ComponentDetails setAnnotationData(NiFiWebConfigurationRequestContext requestContext, String annotationData);
}
/**
* Interprets the request/response with the underlying Processor model.
*/
private class ProcessorFacade implements ComponentFacade {
@Override
public ComponentDetails getComponentDetails(final NiFiWebRequestContext requestContext) {
final String id = requestContext.getId();
final ProcessorDTO processor;
if (properties.isClusterManager()) {
// create the request URL
@ -320,7 +322,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
public ComponentDetails setAnnotationData(final NiFiWebConfigurationRequestContext requestContext, final String annotationData) {
final Revision revision = requestContext.getRevision();
final String id = requestContext.getId();
final ProcessorDTO processor;
if (properties.isClusterManager()) {
// create the request URL
@ -360,7 +362,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// check for issues replicating request
checkResponse(nodeResponse, id);
// return processor
ProcessorEntity entity = (ProcessorEntity) nodeResponse.getUpdatedEntity();
if (entity == null) {
@ -371,11 +373,11 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
final ConfigurationSnapshot<ProcessorDTO> response = serviceFacade.setProcessorAnnotationData(revision, id, annotationData);
processor = response.getConfiguration();
}
// return the processor info
return getComponentConfiguration(processor);
}
private ComponentDetails getComponentConfiguration(final ProcessorDTO processor) {
final ProcessorConfigDTO processorConfig = processor.getConfig();
return new ComponentDetails.Builder()
@ -388,16 +390,18 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
.validateErrors(processor.getValidationErrors()).build();
}
}
/**
* Interprets the request/response with the underlying ControllerService model.
* Interprets the request/response with the underlying ControllerService
* model.
*/
private class ControllerServiceFacade implements ComponentFacade {
@Override
public ComponentDetails getComponentDetails(final NiFiWebRequestContext requestContext) {
final String id = requestContext.getId();
final ControllerServiceDTO controllerService;
// if the lookup has the service that means we are either a node or
// the ncm and the service is available there only
if (controllerServiceLookup.getControllerService(id) != null) {
@ -408,7 +412,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (!properties.isClusterManager()) {
throw new ResourceNotFoundException(String.format("Controller service[%s] could not be found on this NiFi.", id));
}
// create the request URL
URI requestUrl;
try {
@ -443,13 +447,13 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
public ComponentDetails setAnnotationData(final NiFiWebConfigurationRequestContext requestContext, final String annotationData) {
final Revision revision = requestContext.getRevision();
final String id = requestContext.getId();
final ControllerServiceDTO controllerService;
if (controllerServiceLookup.getControllerService(id) != null) {
final ControllerServiceDTO controllerServiceDto = new ControllerServiceDTO();
controllerServiceDto.setId(id);
controllerServiceDto.setAnnotationData(annotationData);
final ConfigurationSnapshot<ControllerServiceDTO> response = serviceFacade.updateControllerService(revision, controllerServiceDto);
controllerService = response.getConfiguration();
} else {
@ -458,13 +462,13 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (!properties.isClusterManager()) {
throw new ResourceNotFoundException(String.format("Controller service[%s] could not be found on this NiFi.", id));
}
// since this PUT request can be interpreted as a request to create a controller service
// we need to be sure that this service exists on the node before the request is replicated.
// this is done by attempting to get the details. if the service doesn't exist it will
// throw a ResourceNotFoundException
getComponentDetails(requestContext);
// create the request URL
URI requestUrl;
try {
@ -498,7 +502,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// check for issues replicating request
checkResponse(nodeResponse, id);
// return controller service
ControllerServiceEntity entity = (ControllerServiceEntity) nodeResponse.getUpdatedEntity();
if (entity == null) {
@ -506,11 +510,11 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
}
controllerService = entity.getControllerService();
}
// return the controller service info
return getComponentConfiguration(controllerService);
}
private ComponentDetails getComponentConfiguration(final ControllerServiceDTO controllerService) {
return new ComponentDetails.Builder()
.id(controllerService.getId())
@ -522,16 +526,18 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
.validateErrors(controllerService.getValidationErrors()).build();
}
}
/**
* Interprets the request/response with the underlying ControllerService model.
* Interprets the request/response with the underlying ControllerService
* model.
*/
private class ReportingTaskFacade implements ComponentFacade {
@Override
public ComponentDetails getComponentDetails(final NiFiWebRequestContext requestContext) {
final String id = requestContext.getId();
final ReportingTaskDTO reportingTask;
// if the provider has the service that means we are either a node or
// the ncm and the service is available there only
if (reportingTaskProvider.getReportingTaskNode(id) != null) {
@ -542,7 +548,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (!properties.isClusterManager()) {
throw new ResourceNotFoundException(String.format("Reporting task[%s] could not be found on this NiFi.", id));
}
// create the request URL
URI requestUrl;
try {
@ -577,13 +583,13 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
public ComponentDetails setAnnotationData(final NiFiWebConfigurationRequestContext requestContext, final String annotationData) {
final Revision revision = requestContext.getRevision();
final String id = requestContext.getId();
final ReportingTaskDTO reportingTask;
if (reportingTaskProvider.getReportingTaskNode(id) != null) {
final ReportingTaskDTO reportingTaskDto = new ReportingTaskDTO();
reportingTaskDto.setId(id);
reportingTaskDto.setAnnotationData(annotationData);
final ConfigurationSnapshot<ReportingTaskDTO> response = serviceFacade.updateReportingTask(revision, reportingTaskDto);
reportingTask = response.getConfiguration();
} else {
@ -592,13 +598,13 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
if (!properties.isClusterManager()) {
throw new ResourceNotFoundException(String.format("Reporting task[%s] could not be found on this NiFi.", id));
}
// since this PUT request can be interpreted as a request to create a reporting task
// we need to be sure that this task exists on the node before the request is replicated.
// this is done by attempting to get the details. if the service doesn't exist it will
// throw a ResourceNotFoundException
getComponentDetails(requestContext);
// create the request URL
URI requestUrl;
try {
@ -632,7 +638,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
// check for issues replicating request
checkResponse(nodeResponse, id);
// return reporting task
ReportingTaskEntity entity = (ReportingTaskEntity) nodeResponse.getUpdatedEntity();
if (entity == null) {
@ -640,11 +646,11 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
}
reportingTask = entity.getReportingTask();
}
// return the processor info
return getComponentConfiguration(reportingTask);
}
private ComponentDetails getComponentConfiguration(final ReportingTaskDTO reportingTask) {
return new ComponentDetails.Builder()
.id(reportingTask.getId())
@ -656,7 +662,7 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
.validateErrors(reportingTask.getValidationErrors()).build();
}
}
/**
* Gets the headers for the request to replicate to each node while
* clustered.
@ -685,13 +691,13 @@ public class StandardNiFiWebConfigurationContext implements NiFiWebConfiguration
}
return headers;
}
/**
* Checks the specified response and drains the stream appropriately.
*
*
* @param nodeResponse
* @param revision
* @param id
* @param id
*/
private void checkResponse(final NodeResponse nodeResponse, final String id) {
if (nodeResponse.hasThrowable()) {

4
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiWebContext.java
View File

@ -288,8 +288,8 @@ public class StandardNiFiWebContext implements NiFiWebContext {
* Gets the headers for the request to replicate to each node while
* clustered.
*
* @param config
* @return
* @param config config
* @return headers
*/
private Map<String, String> getHeaders(final NiFiWebContextConfig config) {
final Map<String, String> headers = new HashMap<>();

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ApplicationResource.java
View File

@ -179,7 +179,7 @@ public abstract class ApplicationResource {
// get cluster context from threadlocal
ClusterContext clusterCtx = ClusterContextThreadLocal.getContext();
if (clusterCtx != null) {
// serialize cluster context
String serializedClusterContext = WebUtils.serializeObjectToHex(clusterCtx);
if (serializedClusterContext.length() > CLUSTER_CONTEXT_HEADER_VALUE_MAX_BYTES) {

8
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ControllerResource.java
View File

@ -159,7 +159,7 @@ public class ControllerResource extends ApplicationResource {
public SnippetResource getSnippetResource() {
return resourceContext.getResource(SnippetResource.class);
}
/**
* Locates the Controller Services sub-resource.
*
@ -169,7 +169,7 @@ public class ControllerResource extends ApplicationResource {
public ControllerServiceResource getControllerServiceResource() {
return resourceContext.getResource(ControllerServiceResource.class);
}
/**
* Locates the Reporting Tasks sub-resource.
*
@ -730,7 +730,7 @@ public class ControllerResource extends ApplicationResource {
// generate the response
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Retrieves the types of controller services that this NiFi supports.
*
@ -766,7 +766,7 @@ public class ControllerResource extends ApplicationResource {
// generate the response
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Retrieves the types of reporting tasks that this NiFi supports.
*

174
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ControllerServiceResource.java
View File

@ -84,12 +84,12 @@ public class ControllerServiceResource extends ApplicationResource {
@Context
private ServletContext servletContext;
/**
* Populates the uri for the specified controller service.
*
*
* @param controllerServices
* @return
* @return
*/
private Set<ControllerServiceDTO> populateRemainingControllerServicesContent(final String availability, final Set<ControllerServiceDTO> controllerServices) {
for (ControllerServiceDTO controllerService : controllerServices) {
@ -97,7 +97,7 @@ public class ControllerServiceResource extends ApplicationResource {
}
return controllerServices;
}
/**
* Populates the uri for the specified controller service.
*/
@ -105,7 +105,7 @@ public class ControllerServiceResource extends ApplicationResource {
// populate the controller service href
controllerService.setUri(generateResourceUri("controller", "controller-services", availability, controllerService.getId()));
controllerService.setAvailability(availability);
// see if this processor has any ui extensions
final UiExtensionMapping uiExtensionMapping = (UiExtensionMapping) servletContext.getAttribute("nifi-ui-extensions");
if (uiExtensionMapping.hasUiExtension(controllerService.getType())) {
@ -116,16 +116,16 @@ public class ControllerServiceResource extends ApplicationResource {
}
}
}
return controllerService;
}
/**
* Parses the availability and ensure that the specified availability makes sense for the
* given NiFi instance.
*
* Parses the availability and ensure that the specified availability makes
* sense for the given NiFi instance.
*
* @param availability
* @return
* @return
*/
private Availability parseAvailability(final String availability) {
final Availability avail;
@ -134,23 +134,24 @@ public class ControllerServiceResource extends ApplicationResource {
} catch (IllegalArgumentException iae) {
throw new IllegalArgumentException(String.format("Availability: Value must be one of [%s]", StringUtils.join(Availability.values(), ", ")));
}
// ensure this nifi is an NCM is specifying NCM availability
if (!properties.isClusterManager() && Availability.NCM.equals(avail)) {
throw new IllegalArgumentException("Availability of NCM is only applicable when the NiFi instance is the cluster manager.");
}
return avail;
}
/**
* Retrieves all the of controller services in this NiFi.
*
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @return A controllerServicesEntity.
*/
@GET
@ -160,7 +161,7 @@ public class ControllerServiceResource extends ApplicationResource {
@TypeHint(ControllerServicesEntity.class)
public Response getControllerServices(@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId, @PathParam("availability") String availability) {
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -168,7 +169,7 @@ public class ControllerServiceResource extends ApplicationResource {
// get all the controller services
final Set<ControllerServiceDTO> controllerServices = populateRemainingControllerServicesContent(availability, serviceFacade.getControllerServices());
// create the revision
final RevisionDTO revision = new RevisionDTO();
revision.setClientId(clientId.getClientId());
@ -191,8 +192,9 @@ public class ControllerServiceResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param type The type of controller service to create.
* @return A controllerServiceEntity.
*/
@ -206,9 +208,9 @@ public class ControllerServiceResource extends ApplicationResource {
@Context HttpServletRequest httpServletRequest,
@FormParam(VERSION) LongParameter version,
@FormParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability,
@PathParam("availability") String availability,
@FormParam("type") String type) {
// create the controller service DTO
final ControllerServiceDTO controllerServiceDTO = new ControllerServiceDTO();
controllerServiceDTO.setType(type);
@ -232,8 +234,9 @@ public class ControllerServiceResource extends ApplicationResource {
* Creates a new Controller Service.
*
* @param httpServletRequest
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param controllerServiceEntity A controllerServiceEntity.
* @return A controllerServiceEntity.
*/
@ -245,19 +248,19 @@ public class ControllerServiceResource extends ApplicationResource {
@TypeHint(ControllerServiceEntity.class)
public Response createControllerService(
@Context HttpServletRequest httpServletRequest,
@PathParam("availability") String availability,
@PathParam("availability") String availability,
ControllerServiceEntity controllerServiceEntity) {
final Availability avail = parseAvailability(availability);
if (controllerServiceEntity == null || controllerServiceEntity.getControllerService()== null) {
if (controllerServiceEntity == null || controllerServiceEntity.getControllerService() == null) {
throw new IllegalArgumentException("Controller service details must be specified.");
}
if (controllerServiceEntity.getRevision() == null) {
throw new IllegalArgumentException("Revision must be specified.");
}
if (controllerServiceEntity.getControllerService().getId() != null) {
throw new IllegalArgumentException("Controller service ID cannot be specified.");
}
@ -265,10 +268,10 @@ public class ControllerServiceResource extends ApplicationResource {
if (StringUtils.isBlank(controllerServiceEntity.getControllerService().getType())) {
throw new IllegalArgumentException("The type of controller service to create must be specified.");
}
// get the revision
final RevisionDTO revision = controllerServiceEntity.getRevision();
// if cluster manager, convert POST to PUT (to maintain same ID across nodes) and replicate
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
// create ID for resource
@ -324,8 +327,9 @@ public class ControllerServiceResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param id The id of the controller service to retrieve
* @return A controllerServiceEntity.
*/
@ -334,11 +338,11 @@ public class ControllerServiceResource extends ApplicationResource {
@Path("/{availability}/{id}")
@PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
@TypeHint(ControllerServiceEntity.class)
public Response getControllerService(@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
public Response getControllerService(@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability, @PathParam("id") String id) {
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -358,10 +362,10 @@ public class ControllerServiceResource extends ApplicationResource {
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Returns the descriptor for the specified property.
*
*
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
@ -376,46 +380,47 @@ public class ControllerServiceResource extends ApplicationResource {
@PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
@TypeHint(PropertyDescriptorEntity.class)
public Response getPropertyDescriptor(
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability, @PathParam("id") String id,
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability, @PathParam("id") String id,
@QueryParam("propertyName") String propertyName) {
final Availability avail = parseAvailability(availability);
// ensure the property name is specified
if (propertyName == null) {
throw new IllegalArgumentException("The property name must be specified.");
}
// replicate if cluster manager and service is on node
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
}
// get the property descriptor
final PropertyDescriptorDTO descriptor = serviceFacade.getControllerServicePropertyDescriptor(id, propertyName);
// create the revision
final RevisionDTO revision = new RevisionDTO();
revision.setClientId(clientId.getClientId());
// generate the response entity
final PropertyDescriptorEntity entity = new PropertyDescriptorEntity();
entity.setRevision(revision);
entity.setPropertyDescriptor(descriptor);
// generate the response
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Retrieves the references of the specified controller service.
*
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param id The id of the controller service to retrieve
* @return A controllerServiceEntity.
*/
@ -429,7 +434,7 @@ public class ControllerServiceResource extends ApplicationResource {
@PathParam("availability") String availability, @PathParam("id") String id) {
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -449,7 +454,7 @@ public class ControllerServiceResource extends ApplicationResource {
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Updates the references of the specified controller service.
*
@ -459,12 +464,14 @@ public class ControllerServiceResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param id The id of the controller service to retrieve
* @param state Sets the state of referencing components. A value of RUNNING or STOPPED will update
* referencing schedulable components (Processors and Reporting Tasks). A value of ENABLED or
* DISABLED will update referencing controller services.
* @param state Sets the state of referencing components. A value of RUNNING
* or STOPPED will update referencing schedulable components (Processors and
* Reporting Tasks). A value of ENABLED or DISABLED will update referencing
* controller services.
* @return A controllerServiceEntity.
*/
@PUT
@ -481,7 +488,6 @@ public class ControllerServiceResource extends ApplicationResource {
@FormParam("state") @DefaultValue(StringUtils.EMPTY) String state) {
// parse the state to determine the desired action
// need to consider controller service state first as it shares a state with
// scheduled state (disabled) which is applicable for referencing services
// but not referencing schedulable components
@ -491,29 +497,29 @@ public class ControllerServiceResource extends ApplicationResource {
} catch (final IllegalArgumentException iae) {
// ignore
}
ScheduledState scheduledState = null;
try {
scheduledState = ScheduledState.valueOf(state);
} catch (final IllegalArgumentException iae) {
// ignore
}
// ensure an action has been specified
if (scheduledState == null && controllerServiceState == null) {
throw new IllegalArgumentException("Must specify the updated state. To update referencing Processors "
+ "and Reporting Tasks the state should be RUNNING or STOPPED. To update the referencing Controller Services the "
+ "state should be ENABLED or DISABLED.");
}
// ensure the controller service state is not ENABLING or DISABLING
if (controllerServiceState != null && (ControllerServiceState.ENABLING.equals(controllerServiceState) || ControllerServiceState.DISABLING.equals(controllerServiceState))) {
throw new IllegalArgumentException("Cannot set the referencing services to ENABLING or DISABLING");
}
// determine the availability
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.PUT, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -525,16 +531,16 @@ public class ControllerServiceResource extends ApplicationResource {
serviceFacade.verifyUpdateControllerServiceReferencingComponents(id, scheduledState, controllerServiceState);
return generateContinueResponse().build();
}
// determine the specified version
Long clientVersion = null;
if (version != null) {
clientVersion = version.getLong();
}
// get the controller service
final ConfigurationSnapshot<Set<ControllerServiceReferencingComponentDTO>> response =
serviceFacade.updateControllerServiceReferencingComponents(new Revision(clientVersion, clientId.getClientId()), id, scheduledState, controllerServiceState);
final ConfigurationSnapshot<Set<ControllerServiceReferencingComponentDTO>> response
= serviceFacade.updateControllerServiceReferencingComponents(new Revision(clientVersion, clientId.getClientId()), id, scheduledState, controllerServiceState);
// create the revision
final RevisionDTO revision = new RevisionDTO();
@ -558,14 +564,17 @@ public class ControllerServiceResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param id The id of the controller service to update.
* @param name The name of the controller service
* @param annotationData The annotation data for the controller service
* @param comments The comments for the controller service
* @param state The state of this controller service. Should be ENABLED or DISABLED.
* @param markedForDeletion Array of property names whose value should be removed.
* @param state The state of this controller service. Should be ENABLED or
* DISABLED.
* @param markedForDeletion Array of property names whose value should be
* removed.
* @param formParams Additionally, the processor properties and styles are
* specified in the form parameters. Because the property names and styles
* differ from processor to processor they are specified in a map-like
@ -599,7 +608,7 @@ public class ControllerServiceResource extends ApplicationResource {
// create collections for holding the controller service properties
final Map<String, String> updatedProperties = new LinkedHashMap<>();
// go through each parameter and look for processor properties
for (String parameterName : formParams.keySet()) {
if (StringUtils.isNotBlank(parameterName)) {
@ -615,12 +624,12 @@ public class ControllerServiceResource extends ApplicationResource {
}
}
}
// set the properties to remove
for (String propertyToDelete : markedForDeletion) {
updatedProperties.put(propertyToDelete, null);
}
// create the controller service DTO
final ControllerServiceDTO controllerServiceDTO = new ControllerServiceDTO();
controllerServiceDTO.setId(id);
@ -633,7 +642,7 @@ public class ControllerServiceResource extends ApplicationResource {
if (!updatedProperties.isEmpty()) {
controllerServiceDTO.setProperties(updatedProperties);
}
// create the revision
final RevisionDTO revision = new RevisionDTO();
revision.setClientId(clientId.getClientId());
@ -654,8 +663,9 @@ public class ControllerServiceResource extends ApplicationResource {
* Updates the specified a new Controller Service.
*
* @param httpServletRequest
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param id The id of the controller service to update.
* @param controllerServiceEntity A controllerServiceEntity.
* @return A controllerServiceEntity.
@ -668,13 +678,13 @@ public class ControllerServiceResource extends ApplicationResource {
@TypeHint(ControllerServiceEntity.class)
public Response updateControllerService(
@Context HttpServletRequest httpServletRequest,
@PathParam("availability") String availability,
@PathParam("availability") String availability,
@PathParam("id") String id,
ControllerServiceEntity controllerServiceEntity) {
final Availability avail = parseAvailability(availability);
if (controllerServiceEntity == null || controllerServiceEntity.getControllerService()== null) {
if (controllerServiceEntity == null || controllerServiceEntity.getControllerService() == null) {
throw new IllegalArgumentException("Controller service details must be specified.");
}
@ -698,7 +708,7 @@ public class ControllerServiceResource extends ApplicationResource {
// replicate the request
return clusterManager.applyRequest(HttpMethod.PUT, getAbsolutePath(), updateClientId(controllerServiceEntity), getHeaders(headersToOverride)).getResponse();
}
// handle expects request (usually from the cluster manager)
final String expects = httpServletRequest.getHeader(WebClusterManager.NCM_EXPECTS_HTTP_HEADER);
if (expects != null) {
@ -736,8 +746,9 @@ public class ControllerServiceResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the controller service is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all services should use the node availability.
* @param availability Whether the controller service is available on the
* NCM only (ncm) or on the nodes only (node). If this instance is not
* clustered all services should use the node availability.
* @param id The id of the controller service to remove.
* @return A entity containing the client id and an updated revision.
*/
@ -753,7 +764,7 @@ public class ControllerServiceResource extends ApplicationResource {
@PathParam("availability") String availability, @PathParam("id") String id) {
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.DELETE, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -788,7 +799,6 @@ public class ControllerServiceResource extends ApplicationResource {
}
// setters
public void setServiceFacade(NiFiServiceFacade serviceFacade) {
this.serviceFacade = serviceFacade;
}

3
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FunnelResource.java
View File

@ -72,8 +72,9 @@ public class FunnelResource extends ApplicationResource {
/**
* Populates the uri for the specified funnels.
*
* @param funnels
* @return
* @return
*/
public Set<FunnelDTO> populateRemainingFunnelsContent(Set<FunnelDTO> funnels) {
for (FunnelDTO funnel : funnels) {

6
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/HistoryResource.java
View File

@ -262,7 +262,7 @@ public class HistoryResource extends ApplicationResource {
// generate the response
return generateOkResponse(entity).build();
}
/**
* Gets the actions for the specified controller service.
*
@ -293,7 +293,7 @@ public class HistoryResource extends ApplicationResource {
// generate the response
return generateOkResponse(entity).build();
}
/**
* Gets the actions for the specified reporting task.
*
@ -324,7 +324,7 @@ public class HistoryResource extends ApplicationResource {
// generate the response
return generateOkResponse(entity).build();
}
/* setters */
public void setServiceFacade(NiFiServiceFacade serviceFacade) {
this.serviceFacade = serviceFacade;

3
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/InputPortResource.java
View File

@ -74,8 +74,9 @@ public class InputPortResource extends ApplicationResource {
/**
* Populates the uri for the specified input ports.
*
* @param inputPorts
* @return
* @return
*/
public Set<PortDTO> populateRemainingInputPortsContent(Set<PortDTO> inputPorts) {
for (PortDTO inputPort : inputPorts) {

3
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/LabelResource.java
View File

@ -74,8 +74,9 @@ public class LabelResource extends ApplicationResource {
/**
* Populates the uri for the specified labels.
*
* @param labels
* @return
* @return
*/
public Set<LabelDTO> populateRemainingLabelsContent(Set<LabelDTO> labels) {
for (LabelDTO label : labels) {

3
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/OutputPortResource.java
View File

@ -74,8 +74,9 @@ public class OutputPortResource extends ApplicationResource {
/**
* Populates the uri for the specified output ports.
*
* @param outputPorts
* @return
* @return
*/
public Set<PortDTO> populateRemainingOutputPortsContent(Set<PortDTO> outputPorts) {
for (PortDTO outputPort : outputPorts) {

20
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessorResource.java
View File

@ -262,7 +262,7 @@ public class ProcessorResource extends ApplicationResource {
if (processorEntity.getProcessor().getId() != null) {
throw new IllegalArgumentException("Processor ID cannot be specified.");
}
if (StringUtils.isBlank(processorEntity.getProcessor().getType())) {
throw new IllegalArgumentException("The type of processor to create must be specified.");
}
@ -394,10 +394,10 @@ public class ProcessorResource extends ApplicationResource {
// generate the response
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Returns the descriptor for the specified property.
*
*
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
@ -411,31 +411,31 @@ public class ProcessorResource extends ApplicationResource {
@PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
@TypeHint(PropertyDescriptorEntity.class)
public Response getPropertyDescriptor(
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("id") String id, @QueryParam("propertyName") String propertyName) {
// ensure the property name is specified
if (propertyName == null) {
throw new IllegalArgumentException("The property name must be specified.");
}
// replicate if cluster manager
if (properties.isClusterManager()) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
}
// get the property descriptor
final PropertyDescriptorDTO descriptor = serviceFacade.getProcessorPropertyDescriptor(groupId, id, propertyName);
// create the revision
final RevisionDTO revision = new RevisionDTO();
revision.setClientId(clientId.getClientId());
// generate the response entity
final PropertyDescriptorEntity entity = new PropertyDescriptorEntity();
entity.setRevision(revision);
entity.setPropertyDescriptor(descriptor);
// generate the response
return clusterContext(generateOkResponse(entity)).build();
}

4
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/RemoteProcessGroupResource.java
View File

@ -370,7 +370,7 @@ public class RemoteProcessGroupResource extends ApplicationResource {
if (uri.getScheme() == null || uri.getHost() == null) {
throw new IllegalArgumentException("The specified remote process group URL is malformed: " + requestProcessGroupDTO.getTargetUri());
}
if (!(uri.getScheme().equalsIgnoreCase("http") || uri.getScheme().equalsIgnoreCase("https"))) {
throw new IllegalArgumentException("The specified remote process group URL is invalid because it is not http or https: " + requestProcessGroupDTO.getTargetUri());
}
@ -864,7 +864,7 @@ public class RemoteProcessGroupResource extends ApplicationResource {
if (uri.getScheme() == null || uri.getHost() == null) {
throw new IllegalArgumentException("The specified remote process group URL is malformed: " + requestRemoteProcessGroup.getTargetUri());
}
if (!(uri.getScheme().equalsIgnoreCase("http") || uri.getScheme().equalsIgnoreCase("https"))) {
throw new IllegalArgumentException("The specified remote process group URL is invalid because it is not http or https: " + requestRemoteProcessGroup.getTargetUri());
}

131
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ReportingTaskResource.java
View File

@ -77,15 +77,15 @@ public class ReportingTaskResource extends ApplicationResource {
private NiFiServiceFacade serviceFacade;
private WebClusterManager clusterManager;
private NiFiProperties properties;
@Context
private ServletContext servletContext;
/**
* Populates the uri for the specified reporting task.
*
*
* @param reportingTasks
* @return
* @return
*/
private Set<ReportingTaskDTO> populateRemainingReportingTasksContent(final String availability, final Set<ReportingTaskDTO> reportingTasks) {
for (ReportingTaskDTO reportingTask : reportingTasks) {
@ -93,7 +93,7 @@ public class ReportingTaskResource extends ApplicationResource {
}
return reportingTasks;
}
/**
* Populates the uri for the specified reporting task.
*/
@ -101,7 +101,7 @@ public class ReportingTaskResource extends ApplicationResource {
// populate the reporting task href
reportingTask.setUri(generateResourceUri("controller", "reporting-tasks", availability, reportingTask.getId()));
reportingTask.setAvailability(availability);
// see if this processor has any ui extensions
final UiExtensionMapping uiExtensionMapping = (UiExtensionMapping) servletContext.getAttribute("nifi-ui-extensions");
if (uiExtensionMapping.hasUiExtension(reportingTask.getType())) {
@ -112,16 +112,16 @@ public class ReportingTaskResource extends ApplicationResource {
}
}
}
return reportingTask;
}
/**
* Parses the availability and ensure that the specified availability makes sense for the
* given NiFi instance.
*
* Parses the availability and ensure that the specified availability makes
* sense for the given NiFi instance.
*
* @param availability
* @return
* @return
*/
private Availability parseAvailability(final String availability) {
final Availability avail;
@ -130,23 +130,24 @@ public class ReportingTaskResource extends ApplicationResource {
} catch (IllegalArgumentException iae) {
throw new IllegalArgumentException(String.format("Availability: Value must be one of [%s]", StringUtils.join(Availability.values(), ", ")));
}
// ensure this nifi is an NCM is specifying NCM availability
if (!properties.isClusterManager() && Availability.NCM.equals(avail)) {
throw new IllegalArgumentException("Availability of NCM is only applicable when the NiFi instance is the cluster manager.");
}
return avail;
}
/**
* Retrieves all the of reporting tasks in this NiFi.
*
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the reporting task is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all tasks should use the node availability.
* @param availability Whether the reporting task is available on the NCM
* only (ncm) or on the nodes only (node). If this instance is not clustered
* all tasks should use the node availability.
* @return A reportingTasksEntity.
*/
@GET
@ -156,7 +157,7 @@ public class ReportingTaskResource extends ApplicationResource {
@TypeHint(ReportingTasksEntity.class)
public Response getReportingTasks(@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId, @PathParam("availability") String availability) {
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -164,7 +165,7 @@ public class ReportingTaskResource extends ApplicationResource {
// get all the reporting tasks
final Set<ReportingTaskDTO> reportingTasks = populateRemainingReportingTasksContent(availability, serviceFacade.getReportingTasks());
// create the revision
final RevisionDTO revision = new RevisionDTO();
revision.setClientId(clientId.getClientId());
@ -187,8 +188,9 @@ public class ReportingTaskResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the reporting task is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all tasks should use the node availability.
* @param availability Whether the reporting task is available on the NCM
* only (ncm) or on the nodes only (node). If this instance is not clustered
* all tasks should use the node availability.
* @param type The type of reporting task to create.
* @return A reportingTaskEntity.
*/
@ -202,9 +204,9 @@ public class ReportingTaskResource extends ApplicationResource {
@Context HttpServletRequest httpServletRequest,
@FormParam(VERSION) LongParameter version,
@FormParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability,
@PathParam("availability") String availability,
@FormParam("type") String type) {
// create the reporting task DTO
final ReportingTaskDTO reportingTaskDTO = new ReportingTaskDTO();
reportingTaskDTO.setType(type);
@ -228,8 +230,9 @@ public class ReportingTaskResource extends ApplicationResource {
* Creates a new Reporting Task.
*
* @param httpServletRequest
* @param availability Whether the reporting task is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all tasks should use the node availability.
* @param availability Whether the reporting task is available on the NCM
* only (ncm) or on the nodes only (node). If this instance is not clustered
* all tasks should use the node availability.
* @param reportingTaskEntity A reportingTaskEntity.
* @return A reportingTaskEntity.
*/
@ -241,27 +244,27 @@ public class ReportingTaskResource extends ApplicationResource {
@TypeHint(ReportingTaskEntity.class)
public Response createReportingTask(
@Context HttpServletRequest httpServletRequest,
@PathParam("availability") String availability,
@PathParam("availability") String availability,
ReportingTaskEntity reportingTaskEntity) {
final Availability avail = parseAvailability(availability);
if (reportingTaskEntity == null || reportingTaskEntity.getReportingTask()== null) {
if (reportingTaskEntity == null || reportingTaskEntity.getReportingTask() == null) {
throw new IllegalArgumentException("Reporting task details must be specified.");
}
if (reportingTaskEntity.getRevision() == null) {
throw new IllegalArgumentException("Revision must be specified.");
}
if (reportingTaskEntity.getReportingTask().getId() != null) {
throw new IllegalArgumentException("Reporting task ID cannot be specified.");
}
if (StringUtils.isBlank(reportingTaskEntity.getReportingTask().getType())) {
throw new IllegalArgumentException("The type of reporting task to create must be specified.");
}
// get the revision
final RevisionDTO revision = reportingTaskEntity.getRevision();
@ -320,8 +323,9 @@ public class ReportingTaskResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the reporting task is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all tasks should use the node availability.
* @param availability Whether the reporting task is available on the NCM
* only (ncm) or on the nodes only (node). If this instance is not clustered
* all tasks should use the node availability.
* @param id The id of the reporting task to retrieve
* @return A reportingTaskEntity.
*/
@ -330,11 +334,11 @@ public class ReportingTaskResource extends ApplicationResource {
@Path("/{availability}/{id}")
@PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
@TypeHint(ReportingTaskEntity.class)
public Response getReportingTask(@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
public Response getReportingTask(@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability, @PathParam("id") String id) {
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -354,10 +358,10 @@ public class ReportingTaskResource extends ApplicationResource {
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Returns the descriptor for the specified property.
*
*
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
@ -372,38 +376,38 @@ public class ReportingTaskResource extends ApplicationResource {
@PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
@TypeHint(PropertyDescriptorEntity.class)
public Response getPropertyDescriptor(
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability, @PathParam("id") String id,
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@PathParam("availability") String availability, @PathParam("id") String id,
@QueryParam("propertyName") String propertyName) {
final Availability avail = parseAvailability(availability);
// ensure the property name is specified
if (propertyName == null) {
throw new IllegalArgumentException("The property name must be specified.");
}
// replicate if cluster manager and task is on node
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
}
// get the property descriptor
final PropertyDescriptorDTO descriptor = serviceFacade.getReportingTaskPropertyDescriptor(id, propertyName);
// create the revision
final RevisionDTO revision = new RevisionDTO();
revision.setClientId(clientId.getClientId());
// generate the response entity
final PropertyDescriptorEntity entity = new PropertyDescriptorEntity();
entity.setRevision(revision);
entity.setPropertyDescriptor(descriptor);
// generate the response
return clusterContext(generateOkResponse(entity)).build();
}
/**
* Updates the specified reporting task.
*
@ -413,12 +417,14 @@ public class ReportingTaskResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the reporting task is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all tasks should use the node availability.
* @param availability Whether the reporting task is available on the NCM
* only (ncm) or on the nodes only (node). If this instance is not clustered
* all tasks should use the node availability.
* @param id The id of the reporting task to update.
* @param name The name of the reporting task
* @param annotationData The annotation data for the reporting task
* @param markedForDeletion Array of property names whose value should be removed.
* @param markedForDeletion Array of property names whose value should be
* removed.
* @param state The updated scheduled state
* @param schedulingStrategy The scheduling strategy for this reporting task
* @param schedulingPeriod The scheduling period for this reporting task
@ -457,7 +463,7 @@ public class ReportingTaskResource extends ApplicationResource {
// create collections for holding the reporting task properties
final Map<String, String> updatedProperties = new LinkedHashMap<>();
// go through each parameter and look for processor properties
for (String parameterName : formParams.keySet()) {
if (StringUtils.isNotBlank(parameterName)) {
@ -473,12 +479,12 @@ public class ReportingTaskResource extends ApplicationResource {
}
}
}
// set the properties to remove
for (String propertyToDelete : markedForDeletion) {
updatedProperties.put(propertyToDelete, null);
}
// create the reporting task DTO
final ReportingTaskDTO reportingTaskDTO = new ReportingTaskDTO();
reportingTaskDTO.setId(id);
@ -493,7 +499,7 @@ public class ReportingTaskResource extends ApplicationResource {
if (!updatedProperties.isEmpty()) {
reportingTaskDTO.setProperties(updatedProperties);
}
// create the revision
final RevisionDTO revision = new RevisionDTO();
revision.setClientId(clientId.getClientId());
@ -514,8 +520,9 @@ public class ReportingTaskResource extends ApplicationResource {
* Updates the specified a Reporting Task.
*
* @param httpServletRequest
* @param availability Whether the reporting task is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all tasks should use the node availability.
* @param availability Whether the reporting task is available on the NCM
* only (ncm) or on the nodes only (node). If this instance is not clustered
* all tasks should use the node availability.
* @param id The id of the reporting task to update.
* @param reportingTaskEntity A reportingTaskEntity.
* @return A reportingTaskEntity.
@ -528,12 +535,12 @@ public class ReportingTaskResource extends ApplicationResource {
@TypeHint(ReportingTaskEntity.class)
public Response updateReportingTask(
@Context HttpServletRequest httpServletRequest,
@PathParam("availability") String availability,
@PathParam("availability") String availability,
@PathParam("id") String id,
ReportingTaskEntity reportingTaskEntity) {
final Availability avail = parseAvailability(availability);
if (reportingTaskEntity == null || reportingTaskEntity.getReportingTask() == null) {
throw new IllegalArgumentException("Reporting task details must be specified.");
}
@ -558,7 +565,7 @@ public class ReportingTaskResource extends ApplicationResource {
// replicate the request
return clusterManager.applyRequest(HttpMethod.PUT, getAbsolutePath(), updateClientId(reportingTaskEntity), getHeaders(headersToOverride)).getResponse();
}
// handle expects request (usually from the cluster manager)
final String expects = httpServletRequest.getHeader(WebClusterManager.NCM_EXPECTS_HTTP_HEADER);
if (expects != null) {
@ -596,8 +603,9 @@ public class ReportingTaskResource extends ApplicationResource {
* @param clientId Optional client id. If the client id is not specified, a
* new one will be generated. This value (whether specified or generated) is
* included in the response.
* @param availability Whether the reporting task is available on the NCM only (ncm) or on the
* nodes only (node). If this instance is not clustered all tasks should use the node availability.
* @param availability Whether the reporting task is available on the NCM
* only (ncm) or on the nodes only (node). If this instance is not clustered
* all tasks should use the node availability.
* @param id The id of the reporting task to remove.
* @return A entity containing the client id and an updated revision.
*/
@ -613,7 +621,7 @@ public class ReportingTaskResource extends ApplicationResource {
@PathParam("availability") String availability, @PathParam("id") String id) {
final Availability avail = parseAvailability(availability);
// replicate if cluster manager
if (properties.isClusterManager() && Availability.NODE.equals(avail)) {
return clusterManager.applyRequest(HttpMethod.DELETE, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
@ -648,7 +656,6 @@ public class ReportingTaskResource extends ApplicationResource {
}
// setters
public void setServiceFacade(NiFiServiceFacade serviceFacade) {
this.serviceFacade = serviceFacade;
}

3
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/TemplateResource.java
View File

@ -71,8 +71,9 @@ public class TemplateResource extends ApplicationResource {
/**
* Populates the uri for the specified templates.
*
* @param templates
* @return
* @return
*/
public Set<TemplateDTO> populateRemainingTemplatesContent(Set<TemplateDTO> templates) {
for (TemplateDTO template : templates) {

70
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
View File

@ -854,7 +854,7 @@ public final class DtoFactory {
defaultSchedulingPeriod.put(SchedulingStrategy.TIMER_DRIVEN.name(), SchedulingStrategy.TIMER_DRIVEN.getDefaultSchedulingPeriod());
defaultSchedulingPeriod.put(SchedulingStrategy.CRON_DRIVEN.name(), SchedulingStrategy.CRON_DRIVEN.getDefaultSchedulingPeriod());
dto.setDefaultSchedulingPeriod(defaultSchedulingPeriod);
// sort a copy of the properties
final Map<PropertyDescriptor, String> sortedProperties = new TreeMap<>(new Comparator<PropertyDescriptor>() {
@Override
@ -874,7 +874,7 @@ public final class DtoFactory {
}
}
orderedProperties.putAll(sortedProperties);
// build the descriptor and property dtos
dto.setDescriptors(new LinkedHashMap<String, PropertyDescriptorDTO>());
dto.setProperties(new LinkedHashMap<String, String>());
@ -893,7 +893,7 @@ public final class DtoFactory {
// set the property value
dto.getProperties().put(descriptor.getName(), propertyValue);
}
// add the validation errors
final Collection<ValidationResult> validationErrors = reportingTaskNode.getValidationErrors();
if (validationErrors != null && !validationErrors.isEmpty()) {
@ -904,10 +904,10 @@ public final class DtoFactory {
dto.setValidationErrors(errors);
}
return dto;
}
public ControllerServiceDTO createControllerServiceDto(final ControllerServiceNode controllerServiceNode) {
final ControllerServiceDTO dto = new ControllerServiceDTO();
dto.setId(controllerServiceNode.getIdentifier());
@ -916,7 +916,7 @@ public final class DtoFactory {
dto.setState(controllerServiceNode.getState().name());
dto.setAnnotationData(controllerServiceNode.getAnnotationData());
dto.setComments(controllerServiceNode.getComments());
// sort a copy of the properties
final Map<PropertyDescriptor, String> sortedProperties = new TreeMap<>(new Comparator<PropertyDescriptor>() {
@Override
@ -936,7 +936,7 @@ public final class DtoFactory {
}
}
orderedProperties.putAll(sortedProperties);
// build the descriptor and property dtos
dto.setDescriptors(new LinkedHashMap<String, PropertyDescriptorDTO>());
dto.setProperties(new LinkedHashMap<String, String>());
@ -955,7 +955,7 @@ public final class DtoFactory {
// set the property value
dto.getProperties().put(descriptor.getName(), propertyValue);
}
// create the reference dto's
dto.setReferencingComponents(createControllerServiceReferencingComponentsDto(controllerServiceNode.getReferences()));
@ -969,23 +969,23 @@ public final class DtoFactory {
dto.setValidationErrors(errors);
}
return dto;
}
public Set<ControllerServiceReferencingComponentDTO> createControllerServiceReferencingComponentsDto(final ControllerServiceReference reference) {
return createControllerServiceReferencingComponentsDto(reference, new HashSet<ControllerServiceNode>());
}
private Set<ControllerServiceReferencingComponentDTO> createControllerServiceReferencingComponentsDto(final ControllerServiceReference reference, final Set<ControllerServiceNode> visited) {
final Set<ControllerServiceReferencingComponentDTO> referencingComponents = new LinkedHashSet<>();
// get all references
for (final ConfiguredComponent component : reference.getReferencingComponents()) {
final ControllerServiceReferencingComponentDTO dto = new ControllerServiceReferencingComponentDTO();
dto.setId(component.getIdentifier());
dto.setName(component.getName());
List<PropertyDescriptor> propertyDescriptors = null;
Collection<ValidationResult> validationErrors = null;
if (component instanceof ProcessorNode) {
@ -995,7 +995,7 @@ public final class DtoFactory {
dto.setActiveThreadCount(node.getActiveThreadCount());
dto.setType(node.getProcessor().getClass().getName());
dto.setReferenceType(Processor.class.getSimpleName());
propertyDescriptors = node.getProcessor().getPropertyDescriptors();
validationErrors = node.getValidationErrors();
} else if (component instanceof ControllerServiceNode) {
@ -1004,12 +1004,12 @@ public final class DtoFactory {
dto.setType(node.getControllerServiceImplementation().getClass().getName());
dto.setReferenceType(ControllerService.class.getSimpleName());
dto.setReferenceCycle(visited.contains(node));
// if we haven't encountered this service before include it's referencing components
if (!dto.getReferenceCycle()) {
dto.setReferencingComponents(createControllerServiceReferencingComponentsDto(node.getReferences(), visited));
}
propertyDescriptors = node.getControllerServiceImplementation().getPropertyDescriptors();
validationErrors = node.getValidationErrors();
} else if (component instanceof ReportingTaskNode) {
@ -1018,11 +1018,11 @@ public final class DtoFactory {
dto.setActiveThreadCount(node.getActiveThreadCount());
dto.setType(node.getReportingTask().getClass().getName());
dto.setReferenceType(ReportingTask.class.getSimpleName());
propertyDescriptors = node.getReportingTask().getPropertyDescriptors();
validationErrors = node.getValidationErrors();
}
if (propertyDescriptors != null && !propertyDescriptors.isEmpty()) {
final Map<PropertyDescriptor, String> sortedProperties = new TreeMap<>(new Comparator<PropertyDescriptor>() {
@Override
@ -1057,7 +1057,7 @@ public final class DtoFactory {
dto.getProperties().put(descriptor.getName(), propertyValue);
}
}
if (validationErrors != null && !validationErrors.isEmpty()) {
final List<String> errors = new ArrayList<>();
for (final ValidationResult validationResult : validationErrors) {
@ -1066,13 +1066,13 @@ public final class DtoFactory {
dto.setValidationErrors(errors);
}
referencingComponents.add(dto);
}
return referencingComponents;
}
public RemoteProcessGroupPortDTO createRemoteProcessGroupPortDto(final RemoteGroupPort port) {
if (port == null) {
return null;
@ -1324,13 +1324,13 @@ public final class DtoFactory {
@SuppressWarnings("deprecation")
private String getCapabilityDescription(final Class<?> cls) {
final CapabilityDescription capabilityDesc = cls.getAnnotation(CapabilityDescription.class);
if ( capabilityDesc != null ) {
if (capabilityDesc != null) {
return capabilityDesc.value();
}
final org.apache.nifi.processor.annotation.CapabilityDescription deprecatedCapabilityDesc =
cls.getAnnotation(org.apache.nifi.processor.annotation.CapabilityDescription.class);
final org.apache.nifi.processor.annotation.CapabilityDescription deprecatedCapabilityDesc
= cls.getAnnotation(org.apache.nifi.processor.annotation.CapabilityDescription.class);
return (deprecatedCapabilityDesc == null) ? null : deprecatedCapabilityDesc.value();
}
@ -1350,8 +1350,8 @@ public final class DtoFactory {
}
} else {
final org.apache.nifi.processor.annotation.Tags deprecatedTagsAnnotation = cls.getAnnotation(org.apache.nifi.processor.annotation.Tags.class);
if ( deprecatedTagsAnnotation != null ) {
for ( final String tag : deprecatedTagsAnnotation.value() ) {
if (deprecatedTagsAnnotation != null) {
for (final String tag : deprecatedTagsAnnotation.value()) {
tags.add(tag);
}
}
@ -1382,7 +1382,7 @@ public final class DtoFactory {
return types;
}
/**
* Creates a ProcessorDTO from the specified ProcessorNode.
*
@ -1829,7 +1829,7 @@ public final class DtoFactory {
dto.setDescription(propertyDescriptor.getDescription());
dto.setDefaultValue(propertyDescriptor.getDefaultValue());
dto.setSupportsEl(propertyDescriptor.isExpressionLanguageSupported());
// set the identifies controller service is applicable
if (propertyDescriptor.getControllerServiceDefinition() != null) {
dto.setIdentifiesControllerService(propertyDescriptor.getControllerServiceDefinition().getName());
@ -1842,7 +1842,7 @@ public final class DtoFactory {
} else {
final List<AllowableValueDTO> allowableValues = new ArrayList<>();
for (final String serviceIdentifier : controllerServiceLookup.getControllerServiceIdentifiers(serviceDefinition)) {
final String displayName = controllerServiceLookup.getControllerServiceName(serviceIdentifier);
final String displayName = controllerServiceLookup.getControllerServiceName(serviceIdentifier);
final AllowableValueDTO allowableValue = new AllowableValueDTO();
allowableValue.setDisplayName(displayName);
@ -1883,7 +1883,6 @@ public final class DtoFactory {
return copy;
}
public ControllerServiceDTO copy(final ControllerServiceDTO original) {
final ControllerServiceDTO copy = new ControllerServiceDTO();
copy.setAnnotationData(original.getAnnotationData());
@ -1901,7 +1900,7 @@ public final class DtoFactory {
copy.setValidationErrors(copy(original.getValidationErrors()));
return copy;
}
public FunnelDTO copy(final FunnelDTO original) {
final FunnelDTO copy = new FunnelDTO();
copy.setId(original.getId());
@ -2294,7 +2293,7 @@ public final class DtoFactory {
*/
public RevisionDTO createRevisionDTO(FlowModification lastMod) {
final Revision revision = lastMod.getRevision();
// create the dto
final RevisionDTO revisionDTO = new RevisionDTO();
revisionDTO.setVersion(revision.getVersion());
@ -2409,7 +2408,6 @@ public final class DtoFactory {
}
/* setters */
public void setControllerServiceLookup(ControllerServiceLookup lookup) {
this.controllerServiceLookup = lookup;
}

17
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/contextlistener/ApplicationStartupContextListener.java
View File

@ -39,7 +39,6 @@ import org.springframework.web.context.support.WebApplicationContextUtils;
* clustered environment. In this case, the cluster manager is created and
* managed.
*
* @author unattributed
*/
public class ApplicationStartupContextListener implements ServletContextListener {
@ -68,14 +67,14 @@ public class ApplicationStartupContextListener implements ServletContextListener
try {
flowService = ctx.getBean("flowService", FlowService.class);
// start and load the flow if we're not clustered (clustered flow loading should
// happen once the application (wars) is fully loaded and initialized). non clustered
// nifi instance need to load the flow before the application (wars) are fully loaded.
// during the flow loading (below) the flow controller is lazily initialized. when the
// flow is loaded after the application is completely initialized (wars deploy), as
// required with a clustered node, users are able to make web requests before the flow
// is loaded (and the flow controller is initialized) which shouldn't be allowed. moving
// the flow loading here when not clustered resolves this.
// start and load the flow if we're not clustered (clustered flow loading should
// happen once the application (wars) is fully loaded and initialized). non clustered
// nifi instance need to load the flow before the application (wars) are fully loaded.
// during the flow loading (below) the flow controller is lazily initialized. when the
// flow is loaded after the application is completely initialized (wars deploy), as
// required with a clustered node, users are able to make web requests before the flow
// is loaded (and the flow controller is initialized) which shouldn't be allowed. moving
// the flow loading here when not clustered resolves this.
if (!properties.isNode()) {
logger.info("Starting Flow Controller...");

47
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
View File

@ -347,10 +347,10 @@ public class ControllerFacade {
public Set<DocumentedTypeDTO> getFlowFileComparatorTypes() {
return dtoFactory.fromDocumentedTypes(ExtensionManager.getExtensions(FlowFilePrioritizer.class));
}
/**
* Returns whether the specified type implements the specified serviceType.
*
*
* @param baseType
* @param type
* @return
@ -362,24 +362,24 @@ public class ControllerFacade {
return true;
}
}
return false;
}
/**
* Gets the ControllerService types that this controller supports.
*
*
* @param serviceType
* @return
* @return
*/
public Set<DocumentedTypeDTO> getControllerServiceTypes(final String serviceType) {
public Set<DocumentedTypeDTO> getControllerServiceTypes(final String serviceType) {
final Set<Class> serviceImplementations = ExtensionManager.getExtensions(ControllerService.class);
// identify the controller services that implement the specified serviceType if applicable
final Set<Class> matchingServiceImplementions;
if (serviceType != null) {
matchingServiceImplementions = new HashSet<>();
// check each type and remove those that aren't in the specified ancestry
for (final Class type : serviceImplementations) {
if (implementsServiceType(serviceType, type)) {
@ -389,14 +389,14 @@ public class ControllerFacade {
} else {
matchingServiceImplementions = serviceImplementations;
}
return dtoFactory.fromDocumentedTypes(matchingServiceImplementions);
}
/**
* Gets the ReportingTask types that this controller supports.
*
* @return
*
* @return
*/
public Set<DocumentedTypeDTO> getReportingTaskTypes() {
return dtoFactory.fromDocumentedTypes(ExtensionManager.getExtensions(ReportingTask.class));
@ -413,8 +413,9 @@ public class ControllerFacade {
/**
* Resets the counter with the specified id.
*
* @param id
* @return
* @return
*/
public Counter resetCounter(final String id) {
final Counter counter = flowController.resetCounter(id);
@ -425,7 +426,7 @@ public class ControllerFacade {
return counter;
}
/**
* Gets the status of this controller.
*
@ -831,7 +832,7 @@ public class ControllerFacade {
if (!downloadAuthorization.isApproved()) {
throw new AccessDeniedException(downloadAuthorization.getExplanation());
}
// get the filename and fall back to the idnetifier (should never happen)
String filename = event.getAttributes().get(CoreAttributes.FILENAME.key());
if (filename == null) {
@ -1193,22 +1194,22 @@ public class ControllerFacade {
for (final Map.Entry<PropertyDescriptor, String> entry : procNode.getProperties().entrySet()) {
final PropertyDescriptor descriptor = entry.getKey();
addIfAppropriate(searchStr, descriptor.getName(), "Property name", matches);
addIfAppropriate(searchStr, descriptor.getDescription(), "Property description", matches);
// never include sensitive properties values in search results
if (descriptor.isSensitive()) {
continue;
}
String value = entry.getValue();
// if unset consider default value
if (value == null) {
value = descriptor.getDefaultValue();
}
// evaluate if the value matches the search criteria
if (StringUtils.containsIgnoreCase(value, searchStr)) {
matches.add("Property value: " + descriptor.getName() + " - " + value);
@ -1286,7 +1287,7 @@ public class ControllerFacade {
for (final FlowFilePrioritizer comparator : queue.getPriorities()) {
addIfAppropriate(searchStr, comparator.getClass().getName(), "Prioritizer", matches);
}
// search expiration
if (StringUtils.containsIgnoreCase("expires", searchStr) || StringUtils.containsIgnoreCase("expiration", searchStr)) {
final int expirationMillis = connection.getFlowFileQueue().getFlowFileExpiration(TimeUnit.MILLISECONDS);
@ -1294,7 +1295,7 @@ public class ControllerFacade {
matches.add("FlowFile expiration: " + connection.getFlowFileQueue().getFlowFileExpiration());
}
}
// search back pressure
if (StringUtils.containsIgnoreCase("back pressure", searchStr) || StringUtils.containsIgnoreCase("pressure", searchStr)) {
final String backPressureDataSize = connection.getFlowFileQueue().getBackPressureDataSizeThreshold();

37
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/ConnectionDAO.java
View File

@ -20,15 +20,12 @@ import java.util.Set;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.web.api.dto.ConnectionDTO;
/**
*
*/
public interface ConnectionDAO {
/**
* Gets the specified Connection.
*
* @param groupId
* @param groupId group id
* @param id The connection id
* @return The connection
*/
@ -37,25 +34,25 @@ public interface ConnectionDAO {
/**
* Gets the connections for the specified source processor.
*
* @param groupId
* @param processorId
* @return
* @param groupId group id
* @param processorId processor id
* @return connections
*/
Set<Connection> getConnectionsForSource(String groupId, String processorId);
/**
* Determines if the specified connection exists.
*
* @param groupId
* @param id
* @return
* @param groupId group id
* @param id id
* @return true if connection exists
*/
boolean hasConnection(String groupId, String id);
/**
* Gets all of the connections.
*
* @param groupId
* @param groupId group identifier
* @return The connections
*/
Set<Connection> getConnections(String groupId);
@ -63,7 +60,7 @@ public interface ConnectionDAO {
/**
* Creates a new Connection.
*
* @param groupId
* @param groupId group id
* @param connectionDTO The connection DTO
* @return The connection
*/
@ -72,23 +69,23 @@ public interface ConnectionDAO {
/**
* Verifies the create request can be processed.
*
* @param groupId
* @param connectionDTO
* @param groupId group id
* @param connectionDTO connection
*/
void verifyCreate(String groupId, ConnectionDTO connectionDTO);
/**
* Verifies the update request can be processed.
*
* @param groupId
* @param connectionDTO
* @param groupId group id
* @param connectionDTO connection
*/
void verifyUpdate(String groupId, ConnectionDTO connectionDTO);
/**
* Updates the specified Connection.
*
* @param groupId
* @param groupId group id
* @param connectionDTO The connection DTO
* @return The connection
*/
@ -97,15 +94,15 @@ public interface ConnectionDAO {
/**
* Verifies the delete request can be processed.
*
* @param groupId
* @param id
* @param groupId group id
* @param id id
*/
void verifyDelete(String groupId, String id);
/**
* Deletes the specified Connection.
*
* @param groupId
* @param groupId group id
* @param id The id of the connection
*/
void deleteConnection(String groupId, String id);

40
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/ControllerServiceDAO.java
View File

@ -24,16 +24,11 @@ import org.apache.nifi.controller.service.ControllerServiceReference;
import org.apache.nifi.controller.service.ControllerServiceState;
import org.apache.nifi.web.api.dto.ControllerServiceDTO;
/**
*
*/
public interface ControllerServiceDAO {
/**
* Determines if the specified controller service exists.
*
* @param controllerServiceId
* @return
* @param controllerServiceId service id
* @return Determines if the specified controller service exists
*/
boolean hasControllerService(String controllerServiceId);
@ -70,34 +65,35 @@ public interface ControllerServiceDAO {
/**
* Updates the referencing components for the specified controller service.
*
* @param controllerServiceId
* @param scheduledState
* @param controllerServiceState the value of state
* @return the org.apache.nifi.controller.service.ControllerServiceReference
*
* @param controllerServiceId service id
* @param scheduledState scheduled state
* @param controllerServiceState the value of state
* @return the org.apache.nifi.controller.service.ControllerServiceReference
*/
ControllerServiceReference updateControllerServiceReferencingComponents(String controllerServiceId, ScheduledState scheduledState, ControllerServiceState controllerServiceState);
/**
* Determines whether this controller service can be updated.
*
* @param controllerServiceDTO
* @param controllerServiceDTO service
*/
void verifyUpdate(ControllerServiceDTO controllerServiceDTO);
/**
* Determines whether the referencing component of the specified controller service can be updated.
*
* @param controllerServiceId
* @param scheduledState
* @param controllerServiceState
* Determines whether the referencing component of the specified controller
* service can be updated.
*
* @param controllerServiceId service id
* @param scheduledState scheduled state
* @param controllerServiceState service state
*/
void verifyUpdateReferencingComponents(String controllerServiceId, ScheduledState scheduledState, ControllerServiceState controllerServiceState);
/**
* Determines whether this controller service can be removed.
*
* @param controllerServiceId
* @param controllerServiceId service id
*/
void verifyDelete(String controllerServiceId);

25
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/FunnelDAO.java
View File

@ -21,24 +21,19 @@ import java.util.Set;
import org.apache.nifi.connectable.Funnel;
import org.apache.nifi.web.api.dto.FunnelDTO;
/**
*
*/
public interface FunnelDAO {
/**
* Determines if the specified funnel exists in the specified group.
*
* @param groupId
* @param funnelId
* @return
* @param groupId group id
* @param funnelId funnel id
* @return Determines if the specified funnel exists in the specified group
*/
boolean hasFunnel(String groupId, String funnelId);
/**
* Creates a funnel in the specified group.
*
* @param groupId
* @param groupId group id
* @param funnelDTO The funnel DTO
* @return The funnel
*/
@ -47,7 +42,7 @@ public interface FunnelDAO {
/**
* Gets the specified funnel in the specified group.
*
* @param groupId
* @param groupId group id
* @param funnelId The funnel id
* @return The funnel
*/
@ -56,7 +51,7 @@ public interface FunnelDAO {
/**
* Gets all of the funnels in the specified group.
*
* @param groupId
* @param groupId group id
* @return The funnels
*/
Set<Funnel> getFunnels(String groupId);
@ -64,7 +59,7 @@ public interface FunnelDAO {
/**
* Updates the specified funnel in the specified group.
*
* @param groupId
* @param groupId group id
* @param funnelDTO The funnel DTO
* @return The funnel
*/
@ -73,15 +68,15 @@ public interface FunnelDAO {
/**
* Determines whether this funnel can be removed.
*
* @param groupId
* @param funnelId
* @param groupId group id
* @param funnelId funnel id
*/
void verifyDelete(String groupId, String funnelId);
/**
* Deletes the specified Funnel in the specified group.
*
* @param groupId
* @param groupId group id
* @param funnelId The funnel id
*/
void deleteFunnel(String groupId, String funnelId);

21
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/LabelDAO.java
View File

@ -20,24 +20,19 @@ import java.util.Set;
import org.apache.nifi.controller.label.Label;
import org.apache.nifi.web.api.dto.LabelDTO;
/**
*
*/
public interface LabelDAO {
/**
* Determines if the specified label exists in the specified group.
*
* @param groupId
* @param labelId
* @return
* @param groupId group id
* @param labelId label id
* @return Determines if the specified label exists in the specified group
*/
boolean hasLabel(String groupId, String labelId);
/**
* Creates a label in the specified group.
*
* @param groupId
* @param groupId group id
* @param labelDTO The label DTO
* @return The label
*/
@ -46,7 +41,7 @@ public interface LabelDAO {
/**
* Gets the specified label in the specified group.
*
* @param groupId
* @param groupId group id
* @param labelId The label id
* @return The label
*/
@ -55,7 +50,7 @@ public interface LabelDAO {
/**
* Gets all of the labels in the specified group.
*
* @param groupId
* @param groupId group id
* @return The labels
*/
Set<Label> getLabels(String groupId);
@ -63,7 +58,7 @@ public interface LabelDAO {
/**
* Updates the specified label in the specified group.
*
* @param groupId
* @param groupId group id
* @param labelDTO The label DTO
* @return The label
*/
@ -72,7 +67,7 @@ public interface LabelDAO {
/**
* Deletes the specified label in the specified group.
*
* @param groupId
* @param groupId group id
* @param labelId The label id
*/
void deleteLabel(String groupId, String labelId);

29
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/PortDAO.java
View File

@ -21,24 +21,19 @@ import java.util.Set;
import org.apache.nifi.connectable.Port;
import org.apache.nifi.web.api.dto.PortDTO;
/**
*
*/
public interface PortDAO {
/**
* Determines if the specified port exists in the specified group.
*
* @param groupId
* @param portId
* @return
* @param groupId group id
* @param portId port id
* @return Determines if the specified port exists in the specified group
*/
boolean hasPort(String groupId, String portId);
/**
* Creates a port in the specified group.
*
* @param groupId
* @param groupId group id
* @param portDTO The port DTO
* @return The port
*/
@ -47,7 +42,7 @@ public interface PortDAO {
/**
* Gets the specified port in the specified group.
*
* @param groupId
* @param groupId group id
* @param portId The port id
* @return The port
*/
@ -56,7 +51,7 @@ public interface PortDAO {
/**
* Gets all of the ports in the specified group.
*
* @param groupId
* @param groupId group id
* @return The ports
*/
Set<Port> getPorts(String groupId);
@ -64,15 +59,15 @@ public interface PortDAO {
/**
* Verifies the specified port can be updated per the specified request.
*
* @param groupId
* @param portDTO
* @param groupId group id
* @param portDTO port
*/
void verifyUpdate(String groupId, PortDTO portDTO);
/**
* Updates the specified port in the specified group.
*
* @param groupId
* @param groupId group
* @param portDTO The port DTO
* @return The port
*/
@ -81,15 +76,15 @@ public interface PortDAO {
/**
* Verifies the specified port can be removed.
*
* @param groupId
* @param portId
* @param groupId group id
* @param portId port id
*/
void verifyDelete(String groupId, String portId);
/**
* Deletes the specified label in the specified group.
*
* @param groupId
* @param groupId group id
* @param portId The port id
*/
void deletePort(String groupId, String portId);

33
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/ProcessorDAO.java
View File

@ -20,24 +20,19 @@ import java.util.Set;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.web.api.dto.ProcessorDTO;
/**
*
*/
public interface ProcessorDAO {
/**
* Determines if the specified processor is loaded.
*
* @param groupId
* @param id
* @return
* @param groupId group id
* @param id id
* @return Determines if the specified processor is loaded
*/
boolean hasProcessor(String groupId, String id);
/**
* Creates a new Processor.
*
* @param groupId
* @param groupId group id
* @param processorDTO The processor DTO
* @return The new Processor
*/
@ -46,7 +41,7 @@ public interface ProcessorDAO {
/**
* Gets the Processor transfer object for the specified id.
*
* @param groupId
* @param groupId group id
* @param id Id of the processor to return
* @return The Processor
*/
@ -55,7 +50,7 @@ public interface ProcessorDAO {
/**
* Gets all the Processor transfer objects for this controller.
*
* @param groupId
* @param groupId group id
* @return List of all the Processors
*/
Set<ProcessorNode> getProcessors(String groupId);
@ -63,8 +58,8 @@ public interface ProcessorDAO {
/**
* Verifies the specified processor can be updated.
*
* @param groupId
* @param processorDTO
* @param groupId group id
* @param processorDTO processor
*/
void verifyUpdate(String groupId, ProcessorDTO processorDTO);
@ -72,24 +67,24 @@ public interface ProcessorDAO {
* Updates the configuration for the processor using the specified
* processorDTO.
*
* @param groupId
* @param processorDTO
* @return
* @param groupId group id
* @param processorDTO processor
* @return updated processor
*/
ProcessorNode updateProcessor(String groupId, ProcessorDTO processorDTO);
/**
* Verifies the specified processor can be removed.
*
* @param groupId
* @param processorId
* @param groupId group id
* @param processorId processor id
*/
void verifyDelete(String groupId, String processorId);
/**
* Deletes the specified processor.
*
* @param groupId
* @param groupId group id
* @param processorId The processor id to delete
*/
void deleteProcessor(String groupId, String processorId);

51
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/RemoteProcessGroupDAO.java
View File

@ -22,24 +22,21 @@ import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
/**
*
*/
public interface RemoteProcessGroupDAO {
/**
* Determines if the specified remote process group exists.
*
* @param groupId
* @param remoteProcessGroupId
* @return
* @param groupId group id
* @param remoteProcessGroupId group id
* @return true if the specified remote process group exists
*/
boolean hasRemoteProcessGroup(String groupId, String remoteProcessGroupId);
/**
* Creates a remote process group reference.
*
* @param groupId
* @param groupId group id
* @param remoteProcessGroup The remote process group
* @return The remote process group
*/
@ -48,7 +45,7 @@ public interface RemoteProcessGroupDAO {
/**
* Gets the specified remote process group.
*
* @param groupId
* @param groupId group id
* @param remoteProcessGroupId The remote process group id
* @return The remote process group
*/
@ -57,7 +54,7 @@ public interface RemoteProcessGroupDAO {
/**
* Gets all of the remote process groups.
*
* @param groupId
* @param groupId group id
* @return The remote process groups
*/
Set<RemoteProcessGroup> getRemoteProcessGroups(String groupId);
@ -65,33 +62,33 @@ public interface RemoteProcessGroupDAO {
/**
* Verifies the specified remote process group can be updated.
*
* @param groupId
* @param remoteProcessGroup
* @param groupId group id
* @param remoteProcessGroup group
*/
void verifyUpdate(String groupId, RemoteProcessGroupDTO remoteProcessGroup);
/**
* Verifies the specified remote process group input port can be updated.
*
* @param groupId
* @param remoteProcessGroupId
* @param remoteProcessGroupPort
* @param groupId group id
* @param remoteProcessGroupId process group id
* @param remoteProcessGroupPort port
*/
void verifyUpdateInputPort(String groupId, String remoteProcessGroupId, RemoteProcessGroupPortDTO remoteProcessGroupPort);
/**
* Verifies the specified remote process group input port can be updated.
*
* @param groupId
* @param remoteProcessGroupId
* @param remoteProcessGroupPort
* @param groupId group id
* @param remoteProcessGroupId group id
* @param remoteProcessGroupPort group port
*/
void verifyUpdateOutputPort(String groupId, String remoteProcessGroupId, RemoteProcessGroupPortDTO remoteProcessGroupPort);
/**
* Updates the specified remote process group.
*
* @param groupId
* @param groupId id
* @param remoteProcessGroup The remote process group
* @return The remote process group
*/
@ -100,9 +97,9 @@ public interface RemoteProcessGroupDAO {
/**
* Updates the specified remote process group input port.
*
* @param groupId
* @param remoteProcessGroupId
* @param remoteProcessGroupPort
* @param groupId id
* @param remoteProcessGroupId id
* @param remoteProcessGroupPort port
* @return
*/
RemoteGroupPort updateRemoteProcessGroupInputPort(String groupId, String remoteProcessGroupId, RemoteProcessGroupPortDTO remoteProcessGroupPort);
@ -110,9 +107,9 @@ public interface RemoteProcessGroupDAO {
/**
* Updates the specified remote process group output port.
*
* @param groupId
* @param remoteProcessGroupId
* @param remoteProcessGroupPort
* @param groupId group id
* @param remoteProcessGroupId group id
* @param remoteProcessGroupPort port
* @return
*/
RemoteGroupPort updateRemoteProcessGroupOutputPort(String groupId, String remoteProcessGroupId, RemoteProcessGroupPortDTO remoteProcessGroupPort);
@ -120,15 +117,15 @@ public interface RemoteProcessGroupDAO {
/**
* Verifies the specified remote process group can be removed.
*
* @param groupId
* @param remoteProcessGroupId
* @param groupId group id
* @param remoteProcessGroupId group id
*/
void verifyDelete(String groupId, String remoteProcessGroupId);
/**
* Deletes the specified remote process group.
*
* @param groupId
* @param groupId group id
* @param remoteProcessGroupId The remote process group id
*/
void deleteRemoteProcessGroup(String groupId, String remoteProcessGroupId);

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/ReportingTaskDAO.java
View File

@ -71,7 +71,7 @@ public interface ReportingTaskDAO {
* @param reportingTaskDTO
*/
void verifyUpdate(ReportingTaskDTO reportingTaskDTO);
/**
* Determines whether this reporting task can be removed.
*

25
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/SnippetDAO.java
View File

@ -20,27 +20,24 @@ import org.apache.nifi.controller.Snippet;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.SnippetDTO;
/**
*
*/
public interface SnippetDAO {
/**
* Copies the specified snippet and added the copy to the flow in the
* specified group.
*
* @param groupId
* @param snippetId
* @param originX
* @param originY
* @return
* @param groupId group id
* @param snippetId snippet id
* @param originX x
* @param originY y
* @return snippet
*/
FlowSnippetDTO copySnippet(String groupId, String snippetId, Double originX, Double originY);
/**
* Creates a snippet.
*
* @param snippetDTO
* @param snippetDTO snippet
* @return The snippet
*/
Snippet createSnippet(SnippetDTO snippetDTO);
@ -48,8 +45,8 @@ public interface SnippetDAO {
/**
* Determines if the specified snippet exists.
*
* @param snippetId
* @return
* @param snippetId snippet id
* @return true if the snippet exists
*/
boolean hasSnippet(String snippetId);
@ -64,14 +61,14 @@ public interface SnippetDAO {
/**
* Verifies the specified snippet can be updated.
*
* @param snippetDTO
* @param snippetDTO snippet
*/
void verifyUpdate(SnippetDTO snippetDTO);
/**
* Updates the specified snippet.
*
* @param snippetDTO
* @param snippetDTO snippet
* @return The snippet
*/
Snippet updateSnippet(SnippetDTO snippetDTO);
@ -79,7 +76,7 @@ public interface SnippetDAO {
/**
* Verifies the specified snippet can be removed.
*
* @param snippetId
* @param snippetId snippet id
*/
void verifyDelete(String snippetId);

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/TemplateDAO.java
View File

@ -49,7 +49,7 @@ public interface TemplateDAO {
* @param originX
* @param originY
* @param templateId
* @return
* @return
*/
FlowSnippetDTO instantiateTemplate(String groupId, Double originX, Double originY, String templateId);

49
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardControllerServiceDAO.java
View File

@ -67,7 +67,7 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
if (controllerServiceDTO.getType() == null) {
throw new IllegalArgumentException("The controller service type must be specified.");
}
try {
// create the controller service
final ControllerServiceNode controllerService = serviceProvider.createControllerService(controllerServiceDTO.getType(), controllerServiceDTO.getId(), true);
@ -126,10 +126,10 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
public ControllerServiceNode updateControllerService(final ControllerServiceDTO controllerServiceDTO) {
// get the controller service
final ControllerServiceNode controllerService = locateControllerService(controllerServiceDTO.getId());
// ensure we can perform the update
verifyUpdate(controllerService, controllerServiceDTO);
// perform the update
configureControllerService(controllerService, controllerServiceDTO);
@ -146,7 +146,7 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
}
}
}
return controllerService;
}
@ -154,7 +154,7 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
public ControllerServiceReference updateControllerServiceReferencingComponents(final String controllerServiceId, final ScheduledState scheduledState, final ControllerServiceState controllerServiceState) {
// get the controller service
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
// this request is either acting upon referncing services or schedulable components
if (controllerServiceState != null) {
if (ControllerServiceState.ENABLED.equals(controllerServiceState)) {
@ -169,22 +169,23 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
serviceProvider.unscheduleReferencingComponents(controllerService);
}
}
return controllerService.getReferences();
}
/**
* Validates the specified configuration for the specified controller service.
*
* Validates the specified configuration for the specified controller
* service.
*
* @param controllerService
* @param controllerServiceDTO
* @return
* @return
*/
private List<String> validateProposedConfiguration(final ControllerServiceNode controllerService, final ControllerServiceDTO controllerServiceDTO) {
final List<String> validationErrors = new ArrayList<>();
return validationErrors;
}
@Override
public void verifyDelete(final String controllerServiceId) {
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
@ -200,7 +201,7 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
@Override
public void verifyUpdateReferencingComponents(String controllerServiceId, ScheduledState scheduledState, ControllerServiceState controllerServiceState) {
final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);
if (controllerServiceState != null) {
if (ControllerServiceState.ENABLED.equals(controllerServiceState)) {
serviceProvider.verifyCanEnableReferencingServices(controllerService);
@ -215,12 +216,12 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
}
}
}
/**
* Verifies the controller service can be updated.
*
*
* @param controllerService
* @param controllerServiceDTO
* @param controllerServiceDTO
*/
private void verifyUpdate(final ControllerServiceNode controllerService, final ControllerServiceDTO controllerServiceDTO) {
// validate the new controller service state if appropriate
@ -228,12 +229,12 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
try {
// attempt to parse the service state
final ControllerServiceState purposedControllerServiceState = ControllerServiceState.valueOf(controllerServiceDTO.getState());
// ensure the state is valid
if (ControllerServiceState.ENABLING.equals(purposedControllerServiceState) || ControllerServiceState.DISABLING.equals(purposedControllerServiceState)) {
throw new IllegalArgumentException();
}
// only attempt an action if it is changing
if (!purposedControllerServiceState.equals(controllerService.getState())) {
if (ControllerServiceState.ENABLED.equals(purposedControllerServiceState)) {
@ -246,14 +247,14 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
throw new IllegalArgumentException("Controller Service state: Value must be one of [ENABLED, DISABLED]");
}
}
boolean modificationRequest = false;
if (isAnyNotNull(controllerServiceDTO.getName(),
controllerServiceDTO.getAnnotationData(),
controllerServiceDTO.getComments(),
controllerServiceDTO.getProperties())) {
modificationRequest = true;
// validate the request
final List<String> requestValidation = validateProposedConfiguration(controllerService, controllerServiceDTO);
@ -262,24 +263,24 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
throw new ValidationException(requestValidation);
}
}
if (modificationRequest) {
controllerService.verifyCanUpdate();
}
}
/**
* Configures the specified controller service.
*
*
* @param controllerService
* @param controllerServiceDTO
* @param controllerServiceDTO
*/
private void configureControllerService(final ControllerServiceNode controllerService, final ControllerServiceDTO controllerServiceDTO) {
final String name = controllerServiceDTO.getName();
final String annotationData = controllerServiceDTO.getAnnotationData();
final String comments = controllerServiceDTO.getComments();
final Map<String, String> properties = controllerServiceDTO.getProperties();
if (isNotNull(name)) {
controllerService.setName(name);
}
@ -301,7 +302,7 @@ public class StandardControllerServiceDAO extends ComponentDAO implements Contro
}
}
}
/**
* Deletes the specified controller service.
*

2
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardProcessorDAO.java
View File

@ -99,7 +99,7 @@ public class StandardProcessorDAO extends ComponentDAO implements ProcessorDAO {
if (processorDTO.getParentGroupId() != null && !flowController.areGroupsSame(groupId, processorDTO.getParentGroupId())) {
throw new IllegalArgumentException("Cannot specify a different Parent Group ID than the Group to which the Processor is being added.");
}
// ensure the type is specified
if (processorDTO.getType() == null) {
throw new IllegalArgumentException("The processor type must be specified.");

1
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardRemoteProcessGroupDAO.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.nifi.web.dao.impl;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

4
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardReportingTaskDAO.java
View File

@ -73,7 +73,7 @@ public class StandardReportingTaskDAO extends ComponentDAO implements ReportingT
if (reportingTaskDTO.getType() == null) {
throw new IllegalArgumentException("The reporting task type must be specified.");
}
try {
// create the reporting task
final ReportingTaskNode reportingTask = reportingTaskProvider.createReportingTask(reportingTaskDTO.getType(), reportingTaskDTO.getId(), true);
@ -321,7 +321,7 @@ public class StandardReportingTaskDAO extends ComponentDAO implements ReportingT
if (isNotNull(schedulingStrategy)) {
reportingTask.setSchedulingStrategy(SchedulingStrategy.valueOf(schedulingStrategy));
}
if (isNotNull(name)) {
reportingTask.setName(name);
}

8
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardSnippetDAO.java
View File

@ -289,10 +289,10 @@ public class StandardSnippetDAO implements SnippetDAO {
lookupSensitiveProcessorProperties(snippet.getProcessors());
}
if ( snippet.getControllerServices() != null ) {
if (snippet.getControllerServices() != null) {
lookupSensitiveControllerServiceProperties(snippet.getControllerServices());
}
// go through each process group if specified
if (snippet.getProcessGroups() != null) {
for (final ProcessGroupDTO group : snippet.getProcessGroups()) {
@ -336,11 +336,11 @@ public class StandardSnippetDAO implements SnippetDAO {
}
}
}
private void lookupSensitiveControllerServiceProperties(final Set<ControllerServiceDTO> controllerServices) {
// go through each service
for (final ControllerServiceDTO serviceDTO : controllerServices) {
// ensure that some property configuration have been specified
final Map<String, String> serviceProperties = serviceDTO.getProperties();
if (serviceProperties != null) {

1
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/spring/ReportingTaskProviderFactoryBean.java
View File

@ -20,7 +20,6 @@ import org.apache.nifi.cluster.manager.impl.WebClusterManager;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.reporting.ReportingTaskProvider;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.dao.ControllerServiceDAO;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.context.ApplicationContext;

1
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/Availability.java
View File

@ -25,7 +25,6 @@ public enum Availability {
* Service or reporting task will run only on the NiFi Cluster Manager (NCM)
*/
NCM,
/**
* Service or reporting task will run only on NiFi Nodes (or standalone
* instance, if not clustered)

137
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/SnippetUtils.java
View File

@ -189,84 +189,84 @@ public final class SnippetUtils {
}
addControllerServicesToSnippet(snippetDto);
return snippetDto;
}
private void addControllerServicesToSnippet(final FlowSnippetDTO snippetDto) {
final Set<ProcessorDTO> processors = snippetDto.getProcessors();
if ( processors != null ) {
for ( final ProcessorDTO processorDto : processors ) {
addControllerServicesToSnippet(snippetDto, processorDto);
}
if (processors != null) {
for (final ProcessorDTO processorDto : processors) {
addControllerServicesToSnippet(snippetDto, processorDto);
}
}
final Set<ProcessGroupDTO> childGroups = snippetDto.getProcessGroups();
if ( childGroups != null ) {
for ( final ProcessGroupDTO processGroupDto : childGroups ) {
final FlowSnippetDTO childGroupDto = processGroupDto.getContents();
if ( childGroupDto != null ) {
addControllerServicesToSnippet(childGroupDto);
}
}
if (childGroups != null) {
for (final ProcessGroupDTO processGroupDto : childGroups) {
final FlowSnippetDTO childGroupDto = processGroupDto.getContents();
if (childGroupDto != null) {
addControllerServicesToSnippet(childGroupDto);
}
}
}
}
private void addControllerServicesToSnippet(final FlowSnippetDTO snippet, final ProcessorDTO processorDto) {
final ProcessorConfigDTO configDto = processorDto.getConfig();
if ( configDto == null ) {
if (configDto == null) {
return;
}
final Map<String, PropertyDescriptorDTO> descriptors = configDto.getDescriptors();
final Map<String, String> properties = configDto.getProperties();
if ( properties != null && descriptors != null ) {
for ( final Map.Entry<String, String> entry : properties.entrySet() ) {
if (properties != null && descriptors != null) {
for (final Map.Entry<String, String> entry : properties.entrySet()) {
final String propName = entry.getKey();
final String propValue = entry.getValue();
if ( propValue == null ) {
if (propValue == null) {
continue;
}
final PropertyDescriptorDTO propertyDescriptorDto = descriptors.get(propName);
if ( propertyDescriptorDto != null && propertyDescriptorDto.getIdentifiesControllerService() != null ) {
if (propertyDescriptorDto != null && propertyDescriptorDto.getIdentifiesControllerService() != null) {
final ControllerServiceNode serviceNode = flowController.getControllerServiceNode(propValue);
if ( serviceNode != null ) {
if (serviceNode != null) {
addControllerServicesToSnippet(snippet, serviceNode);
}
}
}
}
}
private void addControllerServicesToSnippet(final FlowSnippetDTO snippet, final ControllerServiceNode serviceNode) {
if ( isServicePresent(serviceNode.getIdentifier(), snippet.getControllerServices()) ) {
if (isServicePresent(serviceNode.getIdentifier(), snippet.getControllerServices())) {
return;
}
final ControllerServiceDTO serviceNodeDto = dtoFactory.createControllerServiceDto(serviceNode);
Set<ControllerServiceDTO> existingServiceDtos = snippet.getControllerServices();
if ( existingServiceDtos == null ) {
if (existingServiceDtos == null) {
existingServiceDtos = new HashSet<>();
snippet.setControllerServices(existingServiceDtos);
}
existingServiceDtos.add(serviceNodeDto);
for ( final Map.Entry<PropertyDescriptor, String> entry : serviceNode.getProperties().entrySet() ) {
for (final Map.Entry<PropertyDescriptor, String> entry : serviceNode.getProperties().entrySet()) {
final PropertyDescriptor descriptor = entry.getKey();
final String propertyValue = entry.getValue();
if ( descriptor.getControllerServiceDefinition() != null ) {
if (descriptor.getControllerServiceDefinition() != null) {
final ControllerServiceNode referencedNode = flowController.getControllerServiceNode(propertyValue);
if ( referencedNode == null ) {
if (referencedNode == null) {
throw new IllegalStateException("Controller Service with ID " + propertyValue + " is referenced in template but cannot be found");
}
final String referencedNodeId = referencedNode.getIdentifier();
final boolean alreadyPresent = isServicePresent(referencedNodeId, snippet.getControllerServices());
if ( !alreadyPresent ) {
if (!alreadyPresent) {
addControllerServicesToSnippet(snippet, referencedNode);
}
}
@ -274,20 +274,19 @@ public final class SnippetUtils {
}
private boolean isServicePresent(final String serviceId, final Collection<ControllerServiceDTO> services) {
if ( services == null ) {
if (services == null) {
return false;
}
for ( final ControllerServiceDTO existingService : services ) {
if ( serviceId.equals(existingService.getId()) ) {
for (final ControllerServiceDTO existingService : services) {
if (serviceId.equals(existingService.getId())) {
return true;
}
}
return false;
}
public FlowSnippetDTO copy(final FlowSnippetDTO snippetContents, final ProcessGroup group) {
final FlowSnippetDTO snippetCopy = copyContentsForGroup(snippetContents, group.getIdentifier(), null, null);
resolveNameConflicts(snippetCopy, group);
@ -350,33 +349,33 @@ public final class SnippetUtils {
//
// Copy the Controller Services
//
if ( serviceIdMap == null ) {
if (serviceIdMap == null) {
serviceIdMap = new HashMap<>();
final Set<ControllerServiceDTO> services = new HashSet<>();
if ( snippetContents.getControllerServices() != null ) {
for (final ControllerServiceDTO serviceDTO : snippetContents.getControllerServices() ) {
if (snippetContents.getControllerServices() != null) {
for (final ControllerServiceDTO serviceDTO : snippetContents.getControllerServices()) {
final ControllerServiceDTO service = dtoFactory.copy(serviceDTO);
service.setId(generateId(serviceDTO.getId()));
service.setState(ControllerServiceState.DISABLED.name());
services.add(service);
// Map old service ID to new service ID so that we can make sure that we reference the new ones.
serviceIdMap.put(serviceDTO.getId(), service.getId());
}
}
// if there is any controller service that maps to another controller service, update the id's
for ( final ControllerServiceDTO serviceDTO : services ) {
for (final ControllerServiceDTO serviceDTO : services) {
final Map<String, String> properties = serviceDTO.getProperties();
final Map<String, PropertyDescriptorDTO> descriptors = serviceDTO.getDescriptors();
if ( properties != null && descriptors != null ) {
for ( final PropertyDescriptorDTO descriptor : descriptors.values() ) {
if ( descriptor.getIdentifiesControllerService() != null ) {
if (properties != null && descriptors != null) {
for (final PropertyDescriptorDTO descriptor : descriptors.values()) {
if (descriptor.getIdentifiesControllerService() != null) {
final String currentServiceId = properties.get(descriptor.getName());
if ( currentServiceId == null ) {
if (currentServiceId == null) {
continue;
}
final String newServiceId = serviceIdMap.get(currentServiceId);
properties.put(descriptor.getName(), newServiceId);
}
@ -385,7 +384,7 @@ public final class SnippetUtils {
}
snippetContentsCopy.setControllerServices(services);
}
//
// Copy the labels
//
@ -477,7 +476,7 @@ public final class SnippetUtils {
// if there is any controller service that maps to another controller service, update the id's
updateControllerServiceIdentifiers(snippetContentsCopy, serviceIdMap);
//
// Copy ProcessGroups
//
@ -542,43 +541,41 @@ public final class SnippetUtils {
return snippetContentsCopy;
}
private void updateControllerServiceIdentifiers(final FlowSnippetDTO snippet, final Map<String, String> serviceIdMap) {
final Set<ProcessorDTO> processors = snippet.getProcessors();
if ( processors != null ) {
for ( final ProcessorDTO processor : processors ) {
if (processors != null) {
for (final ProcessorDTO processor : processors) {
updateControllerServiceIdentifiers(processor.getConfig(), serviceIdMap);
}
}
for ( final ProcessGroupDTO processGroupDto : snippet.getProcessGroups() ) {
for (final ProcessGroupDTO processGroupDto : snippet.getProcessGroups()) {
updateControllerServiceIdentifiers(processGroupDto.getContents(), serviceIdMap);
}
}
private void updateControllerServiceIdentifiers(final ProcessorConfigDTO configDto, final Map<String, String> serviceIdMap) {
if ( configDto == null ) {
if (configDto == null) {
return;
}
final Map<String, String> properties = configDto.getProperties();
final Map<String, PropertyDescriptorDTO> descriptors = configDto.getDescriptors();
if ( properties != null && descriptors != null ) {
for ( final PropertyDescriptorDTO descriptor : descriptors.values() ) {
if ( descriptor.getIdentifiesControllerService() != null ) {
if (properties != null && descriptors != null) {
for (final PropertyDescriptorDTO descriptor : descriptors.values()) {
if (descriptor.getIdentifiesControllerService() != null) {
final String currentServiceId = properties.get(descriptor.getName());
if ( currentServiceId == null ) {
if (currentServiceId == null) {
continue;
}
final String newServiceId = serviceIdMap.get(currentServiceId);
properties.put(descriptor.getName(), newServiceId);
}
}
}
}
/**
* Generates a new id for the current id that is specified. If no seed is

58
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/util/NiFiTestServer.java
View File

@ -46,16 +46,9 @@ public class NiFiTestServer {
private static final Logger logger = LoggerFactory.getLogger(NiFiTestServer.class);
private Server jetty;
private NiFiProperties properties;
private final NiFiProperties properties;
private WebAppContext webappContext;
/**
* Creates the NiFi test server.
*
* @param webappRoot
* @param contextPath
* @param props
*/
public NiFiTestServer(String webappRoot, String contextPath) {
// load the configuration
properties = NiFiProperties.getInstance();
@ -64,13 +57,6 @@ public class NiFiTestServer {
createServer();
}
/**
* Creates the webapp context.
*
* @param webappRoot
* @param contextPath
* @return
*/
private WebAppContext createWebAppContext(String webappRoot, String contextPath) {
webappContext = new WebAppContext();
webappContext.setContextPath(contextPath);
@ -80,12 +66,6 @@ public class NiFiTestServer {
return webappContext;
}
/**
* Creates the server.
*
* @param webappContext
* @return
*/
private Server createServer() {
jetty = new Server(0);
@ -95,11 +75,6 @@ public class NiFiTestServer {
return jetty;
}
/**
* Creates the connector.
*
* @return
*/
private void createSecureConnector() {
org.eclipse.jetty.util.ssl.SslContextFactory contextFactory = new org.eclipse.jetty.util.ssl.SslContextFactory();
@ -155,21 +130,14 @@ public class NiFiTestServer {
jetty.addConnector(https);
}
/**
* Starts the server.
*
* @throws Exception
*/
public void startServer() throws Exception {
jetty.start();
// ensure the ui extensions are set
webappContext.getServletContext().setAttribute("nifi-ui-extensions", new UiExtensionMapping(Collections.EMPTY_MAP));
}
/**
* Loads the flow.
*/
public void loadFlow() throws Exception {
logger.info("Loading Flow...");
@ -180,20 +148,10 @@ public class NiFiTestServer {
logger.info("Flow loaded successfully.");
}
/**
* Stops the server.
*
* @throws Exception
*/
public void shutdownServer() throws Exception {
jetty.stop();
}
/**
* Returns the port for the server.
*
* @return
*/
public int getPort() {
if (!jetty.isStarted()) {
throw new IllegalStateException("Jetty server not started");
@ -201,20 +159,10 @@ public class NiFiTestServer {
return ((ServerConnector) jetty.getConnectors()[1]).getLocalPort();
}
/**
* Returns the url for the server.
*
* @return
*/
public String getBaseUrl() {
return "https://localhost:" + getPort();
}
/**
* Get a client for accessing the resources.
*
* @return
*/
public Client getClient() {
// create the client
return WebUtils.createClient(null, SslContextFactory.createSslContext(properties));

55
nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-content-viewer/src/main/java/org/apache/nifi/web/ContentViewerController.java
View File

@ -42,7 +42,7 @@ import org.springframework.security.access.AccessDeniedException;
/**
* Controller servlet for viewing content. This is responsible for generating
* the markup for the header and footer of the page. Included in that is the
* the markup for the header and footer of the page. Included in that is the
* combo that allows the user to choose how they wait to view the data
* (original, formatted, hex). If a data viewer is registered for the detected
* content type, it will include the markup it generates in the response.
@ -50,13 +50,13 @@ import org.springframework.security.access.AccessDeniedException;
public class ContentViewerController extends HttpServlet {
private static final Logger logger = LoggerFactory.getLogger(ContentViewerController.class);
// 1.5kb - multiple of 12 (3 bytes = 4 base 64 encoded chars)
private final static int BUFFER_LENGTH = 1536;
/**
* Gets the content and defers to registered viewers to generate the markup.
*
*
* @param request servlet request
* @param response servlet response
* @throws ServletException if a servlet-specific error occurs
@ -67,18 +67,18 @@ public class ContentViewerController extends HttpServlet {
// get the content
final ServletContext servletContext = request.getServletContext();
final ContentAccess contentAccess = (ContentAccess) servletContext.getAttribute("nifi-content-access");
final ContentRequestContext contentRequest = getContentRequest(request);
if (contentRequest.getDataUri() == null) {
request.setAttribute("title", "Error");
request.setAttribute("messages", "The data reference must be specified.");
// forward to the error page
final ServletContext viewerContext = servletContext.getContext("/nifi");
viewerContext.getRequestDispatcher("/message").forward(request, response);
return;
}
// get the content
final DownloadableContent downloadableContent;
try {
@ -86,7 +86,7 @@ public class ContentViewerController extends HttpServlet {
} catch (final ResourceNotFoundException rnfe) {
request.setAttribute("title", "Error");
request.setAttribute("messages", "Unable to find the specified content");
// forward to the error page
final ServletContext viewerContext = servletContext.getContext("/nifi");
viewerContext.getRequestDispatcher("/message").forward(request, response);
@ -94,7 +94,7 @@ public class ContentViewerController extends HttpServlet {
} catch (final AccessDeniedException ade) {
request.setAttribute("title", "Acess Denied");
request.setAttribute("messages", "Unable to approve access to the specified content: " + ade.getMessage());
// forward to the error page
final ServletContext viewerContext = servletContext.getContext("/nifi");
viewerContext.getRequestDispatcher("/message").forward(request, response);
@ -102,7 +102,7 @@ public class ContentViewerController extends HttpServlet {
} catch (final Exception e) {
request.setAttribute("title", "Error");
request.setAttribute("messages", "An unexcepted error has occurred: " + e.getMessage());
// forward to the error page
final ServletContext viewerContext = servletContext.getContext("/nifi");
viewerContext.getRequestDispatcher("/message").forward(request, response);
@ -111,12 +111,12 @@ public class ContentViewerController extends HttpServlet {
// determine how we want to view the data
String mode = request.getParameter("mode");
// if the name isn't set, use original
if (mode == null) {
mode = DisplayMode.Original.name();
}
// determine the display mode
final DisplayMode displayMode;
try {
@ -124,16 +124,16 @@ public class ContentViewerController extends HttpServlet {
} catch (final IllegalArgumentException iae) {
request.setAttribute("title", "Error");
request.setAttribute("messages", "Invalid display mode: " + mode);
// forward to the error page
final ServletContext viewerContext = servletContext.getContext("/nifi");
viewerContext.getRequestDispatcher("/message").forward(request, response);
return;
}
// buffer the content to support reseting in case we need to detect the content type or char encoding
final BufferedInputStream bis = new BufferedInputStream(downloadableContent.getContent());
// detect the content type
final DefaultDetector detector = new DefaultDetector();
@ -147,33 +147,33 @@ public class ContentViewerController extends HttpServlet {
// Get mime type
final MediaType mediatype = detector.detect(tikaStream, metadata);
final String mimeType = mediatype.toString();
// add attributes needed for the header
request.setAttribute("filename", downloadableContent.getFilename());
request.setAttribute("contentType", mimeType);
// generate the header
request.getRequestDispatcher("/WEB-INF/jsp/header.jsp").include(request, response);
// remove the attributes needed for the header
request.removeAttribute("filename");
request.removeAttribute("contentType");
// generate the markup for the content based on the display mode
if (DisplayMode.Hex.equals(displayMode)) {
final byte[] buffer = new byte[BUFFER_LENGTH];
final int read = StreamUtils.fillBuffer(bis, buffer, false);
// trim the byte array if necessary
byte[] bytes = buffer;
if (read != buffer.length) {
bytes = new byte[read];
System.arraycopy(buffer, 0, bytes, 0, read);
}
// convert bytes into the base 64 bytes
final String base64 = Base64.encodeBase64String(bytes);
// defer to the jsp
request.setAttribute("content", base64);
request.getRequestDispatcher("/WEB-INF/jsp/hexview.jsp").include(request, response);
@ -232,13 +232,13 @@ public class ContentViewerController extends HttpServlet {
} catch (final Exception e) {
String message = e.getMessage() != null ? e.getMessage() : e.toString();
message = "Unable to generate view of data: " + message;
// log the error
logger.error(message);
if (logger.isDebugEnabled()) {
logger.error(StringUtils.EMPTY, e);
}
// populate the request attributes
request.setAttribute("title", "Error");
request.setAttribute("messages", message);
@ -248,7 +248,7 @@ public class ContentViewerController extends HttpServlet {
viewerContext.getRequestDispatcher("/message").forward(request, response);
return;
}
// remove the request attribute
request.removeAttribute(ViewableContent.CONTENT_REQUEST_ATTRIBUTE);
}
@ -259,9 +259,8 @@ public class ContentViewerController extends HttpServlet {
}
/**
* Get the content request context based on the specified request.
* @param request
* @return
* @param request request
* @return Get the content request context based on the specified request
*/
private ContentRequestContext getContentRequest(final HttpServletRequest request) {
return new ContentRequestContext() {

8
nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/Deserializer.java vendored
View File

@ -23,18 +23,18 @@ import org.apache.nifi.distributed.cache.client.exception.DeserializationExcepti
/**
* Provides an interface for deserializing an array of bytes into an Object
*
* @param <T>
* @param <T> type
*/
public interface Deserializer<T> {
/**
* Deserializes the given byte array input an Object and returns that value.
*
* @param input
* @return
* @param input input
* @return returns deserialized value
* @throws DeserializationException if a valid object cannot be deserialized
* from the given byte array
* @throws java.io.IOException
* @throws java.io.IOException ex
*/
T deserialize(byte[] input) throws DeserializationException, IOException;

64
nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClient.java vendored
View File

@ -37,12 +37,12 @@ public interface DistributedMapCacheClient extends ControllerService {
* present, serializing the key and value with the given
* {@link Serializer}s.
*
* @param <K>
* @param <V>
* @param <K> type of key
* @param <V> type of value
* @param key the key for into the map
* @param value the value to add to the map if and only if the key is absent
* @param keySerializer
* @param valueSerializer
* @param keySerializer key serializer
* @param valueSerializer value serializer
* @return true if the value was added to the cache, false if the value
* already existed in the cache
*
@ -57,15 +57,17 @@ public interface DistributedMapCacheClient extends ControllerService {
* key, the value associated with the key is returned, after being
* deserialized with the given valueDeserializer.
*
* @param <K>
* @param <V>
* @param key
* @param value
* @param keySerializer
* @param valueSerializer
* @param valueDeserializer
* @return
* @throws IOException
* @param <K> type of key
* @param <V> type of value
* @param key key
* @param value value
* @param keySerializer key serializer
* @param valueSerializer key serializer
* @param valueDeserializer value deserializer
* @return If a value already exists in the cache for the given
* key, the value associated with the key is returned, after being
* deserialized with the given valueDeserializer
* @throws IOException ex
*/
<K, V> V getAndPutIfAbsent(K key, V value, Serializer<K> keySerializer, Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) throws IOException;
@ -73,46 +75,46 @@ public interface DistributedMapCacheClient extends ControllerService {
* Determines if the given value is present in the cache and if so returns
* <code>true</code>, else returns <code>false</code>
*
* @param <K>
* @param key
* @param keySerializer
* @return
* @param <K> type of key
* @param key key
* @param keySerializer key serializer
* @return Determines if the given value is present in the cache and if so returns
* <code>true</code>, else returns <code>false</code>
*
* @throws IOException if unable to communicate with the remote instance
*/
<K> boolean containsKey(K key, Serializer<K> keySerializer) throws IOException;
/**
* Returns the value in the cache for the given key, if one exists;
* otherwise returns <code>null</code>
*
* @param <K>
* @param <V>
* @param <K> type of key
* @param <V> type of value
* @param key the key to lookup in the map
* @param keySerializer
* @param valueDeserializer
* @param keySerializer key serializer
* @param valueDeserializer value serializer
*
* @return
* @throws IOException
* @return the value in the cache for the given key, if one exists;
* otherwise returns <code>null</code>
* @throws IOException ex
*/
<K, V> V get(K key, Serializer<K> keySerializer, Deserializer<V> valueDeserializer) throws IOException;
/**
* Attempts to notify the server that we are finished communicating with it
* and cleans up resources
* @throws java.io.IOException
*
* @throws java.io.IOException ex
*/
void close() throws IOException;
/**
* Removes the entry with the given key from the cache, if it is present.
*
* @param <K>
* @param key
* @param serializer
* @param <K> type of key
* @param key key
* @param serializer serializer
* @return <code>true</code> if the entry is removed, <code>false</code> if
* the key did not exist in the cache
* @throws IOException
* @throws IOException ex
*/
<K> boolean remove(K key, Serializer<K> serializer) throws IOException;

27
nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClient.java vendored
View File

@ -35,9 +35,9 @@ public interface DistributedSetCacheClient extends ControllerService {
* Adds the specified value to the cache, serializing the value with the
* given {@link Serializer}.
*
* @param <T>
* @param value
* @param serializer
* @param <T> type
* @param value value
* @param serializer serializer
* @return true if the value was added to the cache, false if the value
* already existed in the cache
*
@ -46,14 +46,12 @@ public interface DistributedSetCacheClient extends ControllerService {
<T> boolean addIfAbsent(T value, Serializer<T> serializer) throws IOException;
/**
* Returns if the given value is present in the cache and if so returns
* @param <T> type
* @param value value
* @param serializer serializer
* @return if the given value is present in the cache and if so returns
* <code>true</code>, else returns <code>false</code>
*
* @param <T>
* @param value
* @param serializer
* @return
*
* @throws IOException if unable to communicate with the remote instance
*/
<T> boolean contains(T value, Serializer<T> serializer) throws IOException;
@ -61,19 +59,20 @@ public interface DistributedSetCacheClient extends ControllerService {
/**
* Removes the given value from the cache, if it is present.
*
* @param <T>
* @param value
* @param serializer
* @param <T> type
* @param value value
* @param serializer serializer
* @return <code>true</code> if the value is removed, <code>false</code> if
* the value did not exist in the cache
* @throws IOException
* @throws IOException ex
*/
<T> boolean remove(T value, Serializer<T> serializer) throws IOException;
/**
* Attempts to notify the server that we are finished communicating with it
* and cleans up resources
* @throws java.io.IOException
*
* @throws java.io.IOException ex
*/
void close() throws IOException;
}

9
nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/Serializer.java vendored
View File

@ -23,17 +23,18 @@ import org.apache.nifi.distributed.cache.client.exception.SerializationException
/**
* Provides a mechanism by which a value can be serialized to a stream of bytes
* @param <T>
*
* @param <T> type to serialize
*/
public interface Serializer<T> {
/**
* Serializes the given value to the {@link OutputStream}
*
* @param value
* @param output
* @param value value
* @param output stream
* @throws SerializationException If unable to serialize the given value
* @throws java.io.IOException
* @throws java.io.IOException ex
*/
void serialize(T value, OutputStream output) throws SerializationException, IOException;