You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2016/06/17 17:21:46 UTC
[5/5] nifi git commit: NIFI-2007: - Restoring bulletin functionality.
- Ensuring appropriate merging of bulletins in clustered responses.
NIFI-2007:
- Restoring bulletin functionality.
- Ensuring appropriate merging of bulletins in clustered responses.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/0b437e09
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/0b437e09
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/0b437e09
Branch: refs/heads/master
Commit: 0b437e09a7a6a47ac0f717ffe695ed036d840f1c
Parents: 548561e
Author: Matt Gilman <ma...@gmail.com>
Authored: Wed Jun 15 21:42:05 2016 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Jun 17 13:20:20 2016 -0400
----------------------------------------------------------------------
.../controller/status/ProcessGroupStatus.java | 13 --
.../status/RemoteProcessGroupStatus.java | 15 --
.../nifi/reporting/BulletinRepository.java | 10 ++
.../nifi/util/MockBulletinRepository.java | 10 +-
.../nifi/web/api/dto/RemoteProcessGroupDTO.java | 13 +-
.../nifi/web/api/entity/ComponentEntity.java | 19 +++
.../http/StandardHttpResponseMerger.java | 29 +++-
.../endpoints/AbstractSingleDTOEndpoint.java | 50 +-----
.../endpoints/AbstractSingleEntityEndpoint.java | 49 +-----
.../endpoints/BulletinBoardEndpointMerger.java | 43 ++---
.../endpoints/ConnectionEndpointMerger.java | 58 +++++++
.../endpoints/ConnectionsEndpointMerger.java | 72 ++++++++
.../ControllerServiceEndpointMerger.java | 108 ++----------
...ontrollerServiceReferenceEndpointMerger.java | 15 +-
.../ControllerServicesEndpointMerger.java | 58 ++++---
.../coordination/http/endpoints/FlowMerger.java | 167 ++++---------------
.../endpoints/InputPortsEndpointMerger.java | 72 ++++++++
.../endpoints/OutputPortsEndpointMerger.java | 72 ++++++++
.../http/endpoints/PortEndpointMerger.java | 75 +++++++++
.../endpoints/ProcessGroupEndpointMerger.java | 85 ++--------
.../endpoints/ProcessGroupsEndpointMerger.java | 72 ++++++++
.../http/endpoints/ProcessorEndpointMerger.java | 54 +-----
.../endpoints/ProcessorsEndpointMerger.java | 32 ++--
.../RemoteProcessGroupEndpointMerger.java | 92 ++--------
.../RemoteProcessGroupsEndpointMerger.java | 32 ++--
.../endpoints/ReportingTaskEndpointMerger.java | 39 +----
.../endpoints/ReportingTasksEndpointMerger.java | 54 +++---
.../nifi/cluster/manager/BulletinMerger.java | 85 ++++++++++
.../cluster/manager/ComponentEntityMerger.java | 51 ++++++
.../cluster/manager/ConnectionEntityMerger.java | 41 +++++
.../manager/ConnectionsEntityMerger.java | 38 +++++
.../manager/ControllerServiceEntityMerger.java | 146 ++++++++++++++++
.../manager/ControllerServicesEntityMerger.java | 38 +++++
.../nifi/cluster/manager/ErrorMerger.java | 66 ++++++++
.../nifi/cluster/manager/PortEntityMerger.java | 78 +++++++++
.../nifi/cluster/manager/PortsEntityMerger.java | 38 +++++
.../manager/ProcessGroupEntityMerger.java | 43 +++++
.../manager/ProcessGroupsEntityMerger.java | 38 +++++
.../cluster/manager/ProcessorEntityMerger.java | 78 +++++++++
.../cluster/manager/ProcessorsEntityMerger.java | 38 +++++
.../manager/RemoteProcessGroupEntityMerger.java | 119 +++++++++++++
.../RemoteProcessGroupsEntityMerger.java | 38 +++++
.../manager/ReportingTaskEntityMerger.java | 80 +++++++++
.../manager/ReportingTasksEntityMerger.java | 38 +++++
.../endpoints/TestProcessorEndpointMerger.java | 13 +-
.../apache/nifi/controller/FlowController.java | 71 ++++----
.../nifi/events/VolatileBulletinRepository.java | 19 ++-
.../nifi/web/StandardNiFiServiceFacade.java | 111 +++++++-----
.../org/apache/nifi/web/api/dto/DtoFactory.java | 37 ++--
.../apache/nifi/web/api/dto/EntityFactory.java | 26 ++-
.../nifi/web/controller/ControllerFacade.java | 56 +++----
.../src/main/resources/nifi-web-api-context.xml | 3 +
.../src/main/webapp/css/flow-status.css | 4 +
.../nf-ng-canvas-flow-status-controller.js | 10 +-
.../main/webapp/js/nf/canvas/nf-canvas-utils.js | 15 +-
.../js/nf/canvas/nf-controller-services.js | 4 +-
.../webapp/js/nf/canvas/nf-process-group.js | 2 +-
.../main/webapp/js/nf/canvas/nf-processor.js | 4 +-
.../js/nf/canvas/nf-remote-process-group.js | 24 +--
.../src/main/webapp/js/nf/canvas/nf-settings.js | 4 +-
.../webapp/js/nf/summary/nf-summary-table.js | 17 +-
61 files changed, 1990 insertions(+), 891 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java b/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java
index f60bda8..b890c85 100644
--- a/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java
+++ b/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java
@@ -19,7 +19,6 @@ package org.apache.nifi.controller.status;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
/**
@@ -554,18 +553,6 @@ public class ProcessGroupStatus implements Cloneable {
merged.setSentContentSize(merged.getSentContentSize() + statusToMerge.getSentContentSize());
merged.setSentCount(merged.getSentCount() + statusToMerge.getSentCount());
merged.setActiveThreadCount(merged.getActiveThreadCount() + statusToMerge.getActiveThreadCount());
-
- List<String> mergedAuthenticationIssues = merged.getAuthorizationIssues();
- if (mergedAuthenticationIssues == null) {
- mergedAuthenticationIssues = new ArrayList<>();
- }
-
- final List<String> nodeAuthorizationIssues = statusToMerge.getAuthorizationIssues();
- if (nodeAuthorizationIssues != null && !nodeAuthorizationIssues.isEmpty()) {
- mergedAuthenticationIssues.addAll(nodeAuthorizationIssues);
- }
-
- merged.setAuthorizationIssues(mergedAuthenticationIssues);
}
target.setRemoteProcessGroupStatus(mergedRemoteGroupMap.values());
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java b/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java
index 110972e..2b0efff 100644
--- a/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java
+++ b/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java
@@ -16,9 +16,6 @@
*/
package org.apache.nifi.controller.status;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
import java.util.concurrent.TimeUnit;
/**
@@ -40,7 +37,6 @@ public class RemoteProcessGroupStatus implements Cloneable {
private Integer inactiveRemotePortCount;
private long averageLineageDuration;
- private List<String> authorizationIssues = new ArrayList<>();
public String getTargetUri() {
return uri;
@@ -138,14 +134,6 @@ public class RemoteProcessGroupStatus implements Cloneable {
this.inactiveRemotePortCount = inactiveRemotePortCount;
}
- public List<String> getAuthorizationIssues() {
- return new ArrayList<>(authorizationIssues);
- }
-
- public void setAuthorizationIssues(List<String> authorizationIssues) {
- this.authorizationIssues = new ArrayList<>(Objects.requireNonNull(authorizationIssues));
- }
-
public long getAverageLineageDuration() {
return averageLineageDuration;
}
@@ -178,7 +166,6 @@ public class RemoteProcessGroupStatus implements Cloneable {
clonedObj.activeRemotePortCount = activeRemotePortCount;
clonedObj.inactiveRemotePortCount = inactiveRemotePortCount;
clonedObj.averageLineageDuration = averageLineageDuration;
- clonedObj.authorizationIssues = getAuthorizationIssues();
return clonedObj;
}
@@ -209,8 +196,6 @@ public class RemoteProcessGroupStatus implements Cloneable {
builder.append(activeRemotePortCount);
builder.append(", inactiveRemotePortCount=");
builder.append(inactiveRemotePortCount);
- builder.append(", authenticationIssues=");
- builder.append(authorizationIssues);
builder.append("]");
return builder.toString();
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java b/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java
index 2679099..abc0595 100644
--- a/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java
+++ b/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java
@@ -25,6 +25,8 @@ import java.util.List;
*/
public interface BulletinRepository {
+ public static final int MAX_BULLETINS_PER_COMPONENT = 5;
+
/**
* Adds a Bulletin to the repository.
*
@@ -51,6 +53,14 @@ public interface BulletinRepository {
List<Bulletin> findBulletins(BulletinQuery bulletinQuery);
/**
+ * Finds all bulletins for the specified source component.
+ *
+ * @param sourceId id of the source
+ * @return bulletins for the given source
+ */
+ List<Bulletin> findBulletinsForSource(String sourceId);
+
+ /**
* Finds all bulletins for the specified group.
*
* @param groupId id of the group
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java
index 04fc496..a52853a 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java
@@ -16,12 +16,12 @@
*/
package org.apache.nifi.util;
-import java.util.List;
-
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinQuery;
import org.apache.nifi.reporting.BulletinRepository;
+import java.util.List;
+
public class MockBulletinRepository implements BulletinRepository {
@Override
@@ -49,6 +49,12 @@ public class MockBulletinRepository implements BulletinRepository {
}
@Override
+ public List<Bulletin> findBulletinsForSource(String sourceId) {
+ // TODO: Implement
+ return null;
+ }
+
+ @Override
public List<Bulletin> findBulletinsForGroupBySource(String groupId) {
// TODO: Implement
return null;
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java
index 340e9b8..8a2f9e3 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java
@@ -17,11 +17,12 @@
package org.apache.nifi.web.api.dto;
import com.wordnik.swagger.annotations.ApiModelProperty;
-import java.util.Date;
-import java.util.List;
+import org.apache.nifi.web.api.dto.util.DateTimeAdapter;
+
import javax.xml.bind.annotation.XmlType;
import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
-import org.apache.nifi.web.api.dto.util.DateTimeAdapter;
+import java.util.Collection;
+import java.util.Date;
/**
* Details of a remote process group in this NiFi.
@@ -42,7 +43,7 @@ public class RemoteProcessGroupDTO extends ComponentDTO {
private String proxyUser;
private String proxyPassword;
- private List<String> authorizationIssues;
+ private Collection<String> authorizationIssues;
private Boolean transmitting;
private Integer inputPortCount;
@@ -116,11 +117,11 @@ public class RemoteProcessGroupDTO extends ComponentDTO {
@ApiModelProperty(
value = "Any remote authorization issues for the remote process group."
)
- public List<String> getAuthorizationIssues() {
+ public Collection<String> getAuthorizationIssues() {
return authorizationIssues;
}
- public void setAuthorizationIssues(List<String> authorizationIssues) {
+ public void setAuthorizationIssues(Collection<String> authorizationIssues) {
this.authorizationIssues = authorizationIssues;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java
index 231ce53..fbeeb7f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java
@@ -18,10 +18,12 @@ package org.apache.nifi.web.api.entity;
import com.wordnik.swagger.annotations.ApiModelProperty;
import org.apache.nifi.web.api.dto.AccessPolicyDTO;
+import org.apache.nifi.web.api.dto.BulletinDTO;
import org.apache.nifi.web.api.dto.PositionDTO;
import org.apache.nifi.web.api.dto.RevisionDTO;
import javax.xml.bind.annotation.XmlRootElement;
+import java.util.List;
/**
* A base type for request/response entities.
@@ -33,6 +35,7 @@ public class ComponentEntity extends Entity {
private String id;
private PositionDTO position;
private AccessPolicyDTO accessPolicy;
+ private List<BulletinDTO> bulletins;
/**
* @return revision for this request/response
@@ -96,6 +99,22 @@ public class ComponentEntity extends Entity {
this.accessPolicy = accessPolicy;
}
+ /**
+ * The bulletins for this component.
+ *
+ * @return The bulletins
+ */
+ @ApiModelProperty(
+ value = "The bulletins for this component."
+ )
+ public List<BulletinDTO> getBulletins() {
+ return bulletins;
+ }
+
+ public void setBulletins(List<BulletinDTO> bulletins) {
+ this.bulletins = bulletins;
+ }
+
@Override
public int hashCode() {
return id.hashCode();
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java
index ce2e32a..63a2895 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java
@@ -17,18 +17,11 @@
package org.apache.nifi.cluster.coordination.http;
-import java.io.IOException;
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import javax.ws.rs.core.StreamingOutput;
-
import org.apache.nifi.cluster.coordination.http.endpoints.BulletinBoardEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ComponentStateEndpointMerger;
+import org.apache.nifi.cluster.coordination.http.endpoints.ConnectionEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ConnectionStatusEndpiontMerger;
+import org.apache.nifi.cluster.coordination.http.endpoints.ConnectionsEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ControllerServiceEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ControllerServiceReferenceEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ControllerServicesEndpointMerger;
@@ -38,9 +31,13 @@ import org.apache.nifi.cluster.coordination.http.endpoints.DropRequestEndpiontMe
import org.apache.nifi.cluster.coordination.http.endpoints.FlowMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.FlowSnippetEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.GroupStatusEndpointMerger;
+import org.apache.nifi.cluster.coordination.http.endpoints.InputPortsEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ListFlowFilesEndpointMerger;
+import org.apache.nifi.cluster.coordination.http.endpoints.OutputPortsEndpointMerger;
+import org.apache.nifi.cluster.coordination.http.endpoints.PortEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.PortStatusEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ProcessGroupEndpointMerger;
+import org.apache.nifi.cluster.coordination.http.endpoints.ProcessGroupsEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ProcessorEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ProcessorStatusEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.ProcessorsEndpointMerger;
@@ -59,6 +56,14 @@ import org.apache.nifi.stream.io.NullOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.ws.rs.core.StreamingOutput;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
public class StandardHttpResponseMerger implements HttpResponseMerger {
private Logger logger = LoggerFactory.getLogger(StandardHttpResponseMerger.class);
@@ -72,9 +77,15 @@ public class StandardHttpResponseMerger implements HttpResponseMerger {
endpointMergers.add(new RemoteProcessGroupStatusEndpointMerger());
endpointMergers.add(new ProcessorEndpointMerger());
endpointMergers.add(new ProcessorsEndpointMerger());
+ endpointMergers.add(new ConnectionEndpointMerger());
+ endpointMergers.add(new ConnectionsEndpointMerger());
+ endpointMergers.add(new PortEndpointMerger());
+ endpointMergers.add(new InputPortsEndpointMerger());
+ endpointMergers.add(new OutputPortsEndpointMerger());
endpointMergers.add(new RemoteProcessGroupEndpointMerger());
endpointMergers.add(new RemoteProcessGroupsEndpointMerger());
endpointMergers.add(new ProcessGroupEndpointMerger());
+ endpointMergers.add(new ProcessGroupsEndpointMerger());
endpointMergers.add(new FlowSnippetEndpointMerger());
endpointMergers.add(new ProvenanceQueryEndpointMerger());
endpointMergers.add(new ProvenanceEventEndpointMerger());
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java
index db3cfd1..2ef3129 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java
@@ -17,18 +17,16 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
-import java.net.URI;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.web.api.entity.Entity;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
public abstract class AbstractSingleDTOEndpoint<EntityType extends Entity, DtoType> implements EndpointResponseMerger {
@Override
@@ -51,44 +49,6 @@ public abstract class AbstractSingleDTOEndpoint<EntityType extends Entity, DtoTy
return new NodeResponse(clientResponse, responseEntity);
}
-
- /**
- * Normalizes the validation errors by prepending the corresponding nodes when the error does not exist across all nodes.
- *
- * @param validationErrorMap map
- * @param totalNodes total
- * @return normalized errors
- */
- protected Set<String> normalizedMergedValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, int totalNodes) {
- final Set<String> normalizedValidationErrors = new HashSet<>();
- for (final Map.Entry<String, Set<NodeIdentifier>> validationEntry : validationErrorMap.entrySet()) {
- final String msg = validationEntry.getKey();
- final Set<NodeIdentifier> nodeIds = validationEntry.getValue();
-
- if (nodeIds.size() == totalNodes) {
- normalizedValidationErrors.add(msg);
- } else {
- nodeIds.forEach(id -> normalizedValidationErrors.add(id.getApiAddress() + ":" + id.getApiPort() + " -- " + msg));
- }
- }
- return normalizedValidationErrors;
- }
-
- /**
- * Merges the validation errors into the specified map, recording the corresponding node identifier.
- *
- * @param validationErrorMap map
- * @param nodeId id
- * @param nodeValidationErrors errors
- */
- protected void mergeValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, final NodeIdentifier nodeId, final Collection<String> nodeValidationErrors) {
- if (nodeValidationErrors != null) {
- nodeValidationErrors.stream().forEach(
- err -> validationErrorMap.computeIfAbsent(err, k -> new HashSet<NodeIdentifier>())
- .add(nodeId));
- }
- }
-
/**
* @return the class that represents the type of Entity that is expected by this response mapper
*/
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java
index 026ecec..a7ea4a1 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java
@@ -17,18 +17,16 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
-import java.net.URI;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.web.api.entity.Entity;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
public abstract class AbstractSingleEntityEndpoint<EntityType extends Entity> implements EndpointResponseMerger {
@Override
public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
@@ -49,43 +47,6 @@ public abstract class AbstractSingleEntityEndpoint<EntityType extends Entity> im
}
/**
- * Merges the validation errors into the specified map, recording the corresponding node identifier.
- *
- * @param validationErrorMap map
- * @param nodeId id
- * @param nodeValidationErrors errors
- */
- protected void mergeValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, final NodeIdentifier nodeId, final Collection<String> nodeValidationErrors) {
- if (nodeValidationErrors != null) {
- nodeValidationErrors.stream().forEach(
- err -> validationErrorMap.computeIfAbsent(err, k -> new HashSet<NodeIdentifier>())
- .add(nodeId));
- }
- }
-
- /**
- * Normalizes the validation errors by prepending the corresponding nodes when the error does not exist across all nodes.
- *
- * @param validationErrorMap map
- * @param totalNodes total
- * @return normalized errors
- */
- protected Set<String> normalizedMergedValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, int totalNodes) {
- final Set<String> normalizedValidationErrors = new HashSet<>();
- for (final Map.Entry<String, Set<NodeIdentifier>> validationEntry : validationErrorMap.entrySet()) {
- final String msg = validationEntry.getKey();
- final Set<NodeIdentifier> nodeIds = validationEntry.getValue();
-
- if (nodeIds.size() == totalNodes) {
- normalizedValidationErrors.add(msg);
- } else {
- nodeIds.forEach(id -> normalizedValidationErrors.add(id.getApiAddress() + ":" + id.getApiPort() + " -- " + msg));
- }
- }
- return normalizedValidationErrors;
- }
-
- /**
* @return the class that represents the type of Entity that is expected by this response mapper
*/
protected abstract Class<EntityType> getEntityClass();
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java
index 8da6353..8f0f88d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java
@@ -17,21 +17,21 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
+import org.apache.nifi.cluster.manager.BulletinMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.dto.BulletinBoardDTO;
+import org.apache.nifi.web.api.dto.BulletinDTO;
+import org.apache.nifi.web.api.entity.BulletinBoardEntity;
+
import java.net.URI;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
-import org.apache.nifi.cluster.manager.NodeResponse;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.dto.BulletinBoardDTO;
-import org.apache.nifi.web.api.dto.BulletinDTO;
-import org.apache.nifi.web.api.entity.BulletinBoardEntity;
-
public class BulletinBoardEndpointMerger extends AbstractSingleDTOEndpoint<BulletinBoardEntity, BulletinBoardDTO> {
public static final Pattern BULLETIN_BOARD_URI_PATTERN = Pattern.compile("/nifi-api/flow/bulletin-board");
@@ -52,31 +52,16 @@ public class BulletinBoardEndpointMerger extends AbstractSingleDTOEndpoint<Bulle
@Override
protected void mergeResponses(BulletinBoardDTO clientDto, Map<NodeIdentifier, BulletinBoardDTO> dtoMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
- final List<BulletinDTO> bulletinDtos = new ArrayList<>();
+ final Map<NodeIdentifier, List<BulletinDTO>> bulletinDtos = new HashMap<>();
for (final Map.Entry<NodeIdentifier, BulletinBoardDTO> entry : dtoMap.entrySet()) {
- final NodeIdentifier nodeId = entry.getKey();
+ final NodeIdentifier nodeIdentifier = entry.getKey();
final BulletinBoardDTO boardDto = entry.getValue();
- final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort();
-
- for (final BulletinDTO bulletin : boardDto.getBulletins()) {
- bulletin.setNodeAddress(nodeAddress);
- bulletinDtos.add(bulletin);
- }
+ boardDto.getBulletins().forEach(bulletin -> {
+ bulletinDtos.computeIfAbsent(nodeIdentifier, nodeId -> new ArrayList<>()).add(bulletin);
+ });
}
- Collections.sort(bulletinDtos, new Comparator<BulletinDTO>() {
- @Override
- public int compare(final BulletinDTO o1, final BulletinDTO o2) {
- final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp());
- if (timeComparison != 0) {
- return timeComparison;
- }
-
- return o1.getNodeAddress().compareTo(o2.getNodeAddress());
- }
- });
-
- clientDto.setBulletins(bulletinDtos);
+ clientDto.setBulletins(BulletinMerger.mergeBulletins(bulletinDtos));
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java
new file mode 100644
index 0000000..30c6f77
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.ConnectionEntityMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.ConnectionEntity;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class ConnectionEndpointMerger extends AbstractSingleEntityEndpoint<ConnectionEntity> implements EndpointResponseMerger {
+ public static final Pattern PROCESSORS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/connections");
+ public static final Pattern PROCESSOR_URI_PATTERN = Pattern.compile("/nifi-api/connections/[a-f0-9\\-]{36}");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && (PROCESSOR_URI_PATTERN.matcher(uri.getPath()).matches())) {
+ return true;
+ } else if ("POST".equalsIgnoreCase(method) && PROCESSORS_URI_PATTERN.matcher(uri.getPath()).matches()) {
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ protected Class<ConnectionEntity> getEntityClass() {
+ return ConnectionEntity.class;
+ }
+
+
+ @Override
+ protected void mergeResponses(final ConnectionEntity clientEntity, final Map<NodeIdentifier, ConnectionEntity> entityMap, final Set<NodeResponse> successfulResponses,
+ final Set<NodeResponse> problematicResponses) {
+
+ ConnectionEntityMerger.mergeConnections(clientEntity, entityMap);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java
new file mode 100644
index 0000000..661350d
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.ConnectionsEntityMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.ConnectionEntity;
+import org.apache.nifi.web.api.entity.ConnectionsEntity;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class ConnectionsEndpointMerger implements EndpointResponseMerger {
+ public static final Pattern CONNECTIONS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/connections");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && CONNECTIONS_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ @Override
+ public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ConnectionsEntity responseEntity = clientResponse.getClientResponse().getEntity(ConnectionsEntity.class);
+ final Set<ConnectionEntity> connectionEntities = responseEntity.getConnections();
+
+ final Map<String, Map<NodeIdentifier, ConnectionEntity>> entityMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ConnectionsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ConnectionsEntity.class);
+ final Set<ConnectionEntity> nodeConnectionEntities = nodeResponseEntity.getConnections();
+
+ for (final ConnectionEntity nodeConnectionEntity : nodeConnectionEntities) {
+ final NodeIdentifier nodeId = nodeResponse.getNodeId();
+ Map<NodeIdentifier, ConnectionEntity> innerMap = entityMap.get(nodeId);
+ if (innerMap == null) {
+ innerMap = new HashMap<>();
+ entityMap.put(nodeConnectionEntity.getId(), innerMap);
+ }
+
+ innerMap.put(nodeResponse.getNodeId(), nodeConnectionEntity);
+ }
+ }
+
+ ConnectionsEntityMerger.mergeConnections(connectionEntities, entityMap);
+
+ // create a new client response
+ return new NodeResponse(clientResponse, responseEntity);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java
index 10fa281..879d3b5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java
@@ -17,29 +17,27 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.ControllerServiceEntityMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.controller.service.ControllerServiceState;
-import org.apache.nifi.web.api.dto.ControllerServiceDTO;
-import org.apache.nifi.web.api.dto.ControllerServiceReferencingComponentDTO;
import org.apache.nifi.web.api.entity.ControllerServiceEntity;
-import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentEntity;
import java.net.URI;
-import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
-public class ControllerServiceEndpointMerger extends AbstractSingleDTOEndpoint<ControllerServiceEntity, ControllerServiceDTO> {
- public static final String CONTROLLER_SERVICES_URI = "/nifi-api/controller-services/node";
- public static final Pattern CONTROLLER_SERVICE_URI_PATTERN = Pattern.compile("/nifi-api/controller-services/node/[a-f0-9\\-]{36}");
+public class ControllerServiceEndpointMerger extends AbstractSingleEntityEndpoint<ControllerServiceEntity> implements EndpointResponseMerger {
+ public static final String CONTROLLER_CONTROLLER_SERVICES_URI = "/nifi-api/controller/controller-services";
+ public static final Pattern PROCESS_GROUPS_CONTROLLER_SERVICES_URI = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/controller-services");
+ public static final Pattern CONTROLLER_SERVICE_URI_PATTERN = Pattern.compile("/nifi-api/controller-services/[a-f0-9\\-]{36}");
@Override
public boolean canHandle(URI uri, String method) {
if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && CONTROLLER_SERVICE_URI_PATTERN.matcher(uri.getPath()).matches()) {
return true;
- } else if ("POST".equalsIgnoreCase(method) && CONTROLLER_SERVICES_URI.equals(uri.getPath())) {
+ } else if ("POST".equalsIgnoreCase(method) && (CONTROLLER_CONTROLLER_SERVICES_URI.equals(uri.getPath()) || PROCESS_GROUPS_CONTROLLER_SERVICES_URI.matcher(uri.getPath()).matches())) {
return true;
}
@@ -52,95 +50,9 @@ public class ControllerServiceEndpointMerger extends AbstractSingleDTOEndpoint<C
}
@Override
- protected ControllerServiceDTO getDto(ControllerServiceEntity entity) {
- return entity.getComponent();
- }
-
- @Override
- protected void mergeResponses(ControllerServiceDTO clientDto, Map<NodeIdentifier, ControllerServiceDTO> dtoMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
- final Map<String, Set<NodeIdentifier>> validationErrorMap = new HashMap<>();
- final Set<ControllerServiceReferencingComponentEntity> referencingComponents = clientDto.getReferencingComponents();
- final Map<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> nodeReferencingComponentsMap = new HashMap<>();
-
- String state = null;
- for (final Map.Entry<NodeIdentifier, ControllerServiceDTO> nodeEntry : dtoMap.entrySet()) {
- final NodeIdentifier nodeId = nodeEntry.getKey();
- final ControllerServiceDTO nodeControllerService = nodeEntry.getValue();
-
- if (state == null) {
- if (ControllerServiceState.DISABLING.name().equals(nodeControllerService.getState())) {
- state = ControllerServiceState.DISABLING.name();
- } else if (ControllerServiceState.ENABLING.name().equals(nodeControllerService.getState())) {
- state = ControllerServiceState.ENABLING.name();
- }
- }
-
- nodeReferencingComponentsMap.put(nodeId, nodeControllerService.getReferencingComponents());
-
- // merge the validation errors
- mergeValidationErrors(validationErrorMap, nodeId, nodeControllerService.getValidationErrors());
- }
-
- // merge the referencing components
- mergeControllerServiceReferences(referencingComponents, nodeReferencingComponentsMap);
-
- // store the 'transition' state is applicable
- if (state != null) {
- clientDto.setState(state);
- }
+ protected void mergeResponses(ControllerServiceEntity clientEntity, Map<NodeIdentifier, ControllerServiceEntity> entityMap,
+ Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
- // set the merged the validation errors
- clientDto.setValidationErrors(normalizedMergedValidationErrors(validationErrorMap, dtoMap.size()));
+ ControllerServiceEntityMerger.mergeControllerServices(clientEntity, entityMap);
}
-
- public static void mergeControllerServiceReferences(Set<ControllerServiceReferencingComponentEntity> referencingComponents,
- Map<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> referencingComponentMap) {
-
- final Map<String, Integer> activeThreadCounts = new HashMap<>();
- final Map<String, String> states = new HashMap<>();
- for (final Map.Entry<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> nodeEntry : referencingComponentMap.entrySet()) {
- final Set<ControllerServiceReferencingComponentEntity> nodeReferencingComponents = nodeEntry.getValue();
-
- // go through all the nodes referencing components
- if (nodeReferencingComponents != null) {
- for (final ControllerServiceReferencingComponentEntity nodeReferencingComponentEntity : nodeReferencingComponents) {
- final ControllerServiceReferencingComponentDTO nodeReferencingComponent = nodeReferencingComponentEntity.getComponent();
-
- // handle active thread counts
- if (nodeReferencingComponent.getActiveThreadCount() != null && nodeReferencingComponent.getActiveThreadCount() > 0) {
- final Integer current = activeThreadCounts.get(nodeReferencingComponent.getId());
- if (current == null) {
- activeThreadCounts.put(nodeReferencingComponent.getId(), nodeReferencingComponent.getActiveThreadCount());
- } else {
- activeThreadCounts.put(nodeReferencingComponent.getId(), nodeReferencingComponent.getActiveThreadCount() + current);
- }
- }
-
- // handle controller service state
- final String state = states.get(nodeReferencingComponent.getId());
- if (state == null) {
- if (ControllerServiceState.DISABLING.name().equals(nodeReferencingComponent.getState())) {
- states.put(nodeReferencingComponent.getId(), ControllerServiceState.DISABLING.name());
- } else if (ControllerServiceState.ENABLING.name().equals(nodeReferencingComponent.getState())) {
- states.put(nodeReferencingComponent.getId(), ControllerServiceState.ENABLING.name());
- }
- }
- }
- }
- }
-
- // go through each referencing components
- for (final ControllerServiceReferencingComponentEntity referencingComponent : referencingComponents) {
- final Integer activeThreadCount = activeThreadCounts.get(referencingComponent.getId());
- if (activeThreadCount != null) {
- referencingComponent.getComponent().setActiveThreadCount(activeThreadCount);
- }
-
- final String state = states.get(referencingComponent.getId());
- if (state != null) {
- referencingComponent.getComponent().setState(state);
- }
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java
index c299054..129d36f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java
@@ -17,18 +17,19 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
-import java.net.URI;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Pattern;
-
import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.ControllerServiceEntityMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentEntity;
import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentsEntity;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
public class ControllerServiceReferenceEndpointMerger implements EndpointResponseMerger {
public static final Pattern CONTROLLER_SERVICE_REFERENCES_URI_PATTERN = Pattern.compile("/nifi-api/controller-services/node/[a-f0-9\\-]{36}/references");
@@ -59,7 +60,7 @@ public class ControllerServiceReferenceEndpointMerger implements EndpointRespons
resultsMap.put(nodeResponse.getNodeId(), nodeReferencingComponents);
}
- ControllerServiceEndpointMerger.mergeControllerServiceReferences(referencingComponents, resultsMap);
+ ControllerServiceEntityMerger.mergeControllerServiceReferences(referencingComponents, resultsMap);
return new NodeResponse(clientResponse, responseEntity);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java
index ba09d9d..854a027 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java
@@ -17,47 +17,57 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.ControllerServicesEntityMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.web.api.entity.ControllerServiceEntity;
import org.apache.nifi.web.api.entity.ControllerServicesEntity;
import java.net.URI;
+import java.util.HashMap;
import java.util.Map;
import java.util.Set;
-import java.util.stream.Collectors;
+import java.util.regex.Pattern;
-public class ControllerServicesEndpointMerger extends AbstractMultiEntityEndpoint<ControllerServicesEntity, ControllerServiceEntity> {
- public static final String CONTROLLER_SERVICES_URI = "/nifi-api/controller-services/node";
+public class ControllerServicesEndpointMerger implements EndpointResponseMerger {
+ public static final String CONTROLLER_SERVICES_URI = "/nifi-api/flow/controller/controller-services";
+ public static final Pattern PROCESS_GROUPS_CONTROLLER_SERVICES_URI = Pattern.compile("/nifi-api/flow/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/controller-services");
@Override
public boolean canHandle(URI uri, String method) {
- return "GET".equalsIgnoreCase(method) && CONTROLLER_SERVICES_URI.equals(uri.getPath());
+ return "GET".equalsIgnoreCase(method) && (CONTROLLER_SERVICES_URI.equals(uri.getPath()) || PROCESS_GROUPS_CONTROLLER_SERVICES_URI.matcher(uri.getPath()).matches());
}
@Override
- protected Class<ControllerServicesEntity> getEntityClass() {
- return ControllerServicesEntity.class;
- }
+ public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
- @Override
- protected Set<ControllerServiceEntity> getDtos(ControllerServicesEntity entity) {
- return entity.getControllerServices();
- }
+ final ControllerServicesEntity responseEntity = clientResponse.getClientResponse().getEntity(ControllerServicesEntity.class);
+ final Set<ControllerServiceEntity> controllerServiceEntities = responseEntity.getControllerServices();
- @Override
- protected String getComponentId(ControllerServiceEntity entity) {
- return entity.getComponent().getId();
- }
+ final Map<String, Map<NodeIdentifier, ControllerServiceEntity>> entityMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ControllerServicesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ControllerServicesEntity.class);
+ final Set<ControllerServiceEntity> nodeControllerServiceEntities = nodeResponseEntity.getControllerServices();
- @Override
- protected void mergeResponses(ControllerServiceEntity entity, Map<NodeIdentifier, ControllerServiceEntity> entityMap,
- Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) {
-
- new ControllerServiceEndpointMerger().mergeResponses(
- entity.getComponent(),
- entityMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getComponent())),
- successfulResponses,
- problematicResponses);
+ for (final ControllerServiceEntity nodeControllerServiceEntity : nodeControllerServiceEntities) {
+ final NodeIdentifier nodeId = nodeResponse.getNodeId();
+ Map<NodeIdentifier, ControllerServiceEntity> innerMap = entityMap.get(nodeId);
+ if (innerMap == null) {
+ innerMap = new HashMap<>();
+ entityMap.put(nodeControllerServiceEntity.getId(), innerMap);
+ }
+
+ innerMap.put(nodeResponse.getNodeId(), nodeControllerServiceEntity);
+ }
+ }
+
+ ControllerServicesEntityMerger.mergeControllerServices(controllerServiceEntities, entityMap);
+
+ // create a new client response
+ return new NodeResponse(clientResponse, responseEntity);
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java
index c033d19..0cbf213 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java
@@ -17,16 +17,15 @@
package org.apache.nifi.cluster.coordination.http.endpoints;
+import org.apache.nifi.cluster.manager.ConnectionsEntityMerger;
import org.apache.nifi.cluster.manager.NodeResponse;
-import org.apache.nifi.cluster.manager.StatusMerger;
+import org.apache.nifi.cluster.manager.PortsEntityMerger;
+import org.apache.nifi.cluster.manager.ProcessGroupsEntityMerger;
+import org.apache.nifi.cluster.manager.ProcessorsEntityMerger;
+import org.apache.nifi.cluster.manager.RemoteProcessGroupsEntityMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.web.api.dto.flow.FlowDTO;
import org.apache.nifi.web.api.dto.flow.ProcessGroupFlowDTO;
-import org.apache.nifi.web.api.dto.status.ConnectionStatusDTO;
-import org.apache.nifi.web.api.dto.status.PortStatusDTO;
-import org.apache.nifi.web.api.dto.status.ProcessGroupStatusDTO;
-import org.apache.nifi.web.api.dto.status.ProcessorStatusDTO;
-import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusDTO;
import org.apache.nifi.web.api.entity.ConnectionEntity;
import org.apache.nifi.web.api.entity.FunnelEntity;
import org.apache.nifi.web.api.entity.LabelEntity;
@@ -68,23 +67,31 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity
final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses) {
final FlowDTO flowDto = clientDto.getFlow();
-
- final Map<String, List<ConnectionEntity>> connections = new HashMap<>();
+ final Set<ConnectionEntity> clientConnections = flowDto.getConnections();
+ final Set<ProcessorEntity> clientProcessors = flowDto.getProcessors();
+ final Set<PortEntity> clientInputPorts = flowDto.getInputPorts();
+ final Set<PortEntity> clientOutputPorts = flowDto.getOutputPorts();
+ final Set<RemoteProcessGroupEntity> clientRemoteProcessGroups = flowDto.getRemoteProcessGroups();
+ final Set<ProcessGroupEntity> clientProcessGroups = flowDto.getProcessGroups();
+
+ final Map<String, Map<NodeIdentifier, ConnectionEntity>> connections = new HashMap<>();
final Map<String, List<FunnelEntity>> funnels = new HashMap<>();
- final Map<String, List<PortEntity>> inputPorts = new HashMap<>();
+ final Map<String, Map<NodeIdentifier, PortEntity>> inputPorts = new HashMap<>();
final Map<String, List<LabelEntity>> labels = new HashMap<>();
- final Map<String, List<PortEntity>> outputPorts = new HashMap<>();
- final Map<String, List<ProcessorEntity>> processors = new HashMap<>();
- final Map<String, List<RemoteProcessGroupEntity>> rpgs = new HashMap<>();
- final Map<String, List<ProcessGroupEntity>> processGroups = new HashMap<>();
-
- // Create mapping of ComponentID -> all components with that ID (one per node)
- for (final ProcessGroupFlowDTO nodeGroupFlowDto : dtoMap.values()) {
+ final Map<String, Map<NodeIdentifier, PortEntity>> outputPorts = new HashMap<>();
+ final Map<String, Map<NodeIdentifier, ProcessorEntity>> processors = new HashMap<>();
+ final Map<String, Map<NodeIdentifier, RemoteProcessGroupEntity>> rpgs = new HashMap<>();
+ final Map<String, Map<NodeIdentifier, ProcessGroupEntity>> processGroups = new HashMap<>();
+
+ // Create mapping of ComponentID -> [nodeId, entity on that node]
+ for (final Map.Entry<NodeIdentifier, ProcessGroupFlowDTO> nodeGroupFlowEntry : dtoMap.entrySet()) {
+ final NodeIdentifier nodeIdentifier = nodeGroupFlowEntry.getKey();
+ final ProcessGroupFlowDTO nodeGroupFlowDto = nodeGroupFlowEntry.getValue();
final FlowDTO nodeFlowDto = nodeGroupFlowDto.getFlow();
// Merge connection statuses
for (final ConnectionEntity entity : nodeFlowDto.getConnections()) {
- connections.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity);
+ connections.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity);
}
for (final FunnelEntity entity : nodeFlowDto.getFunnels()) {
@@ -92,11 +99,11 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity
}
for (final PortEntity entity : nodeFlowDto.getInputPorts()) {
- inputPorts.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity);
+ inputPorts.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity);
}
for (final PortEntity entity : nodeFlowDto.getOutputPorts()) {
- outputPorts.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity);
+ outputPorts.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity);
}
for (final LabelEntity entity : nodeFlowDto.getLabels()) {
@@ -104,15 +111,15 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity
}
for (final ProcessorEntity entity : nodeFlowDto.getProcessors()) {
- processors.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity);
+ processors.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity);
}
for (final RemoteProcessGroupEntity entity : nodeFlowDto.getRemoteProcessGroups()) {
- rpgs.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity);
+ rpgs.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity);
}
for (final ProcessGroupEntity entity : nodeFlowDto.getProcessGroups()) {
- processGroups.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity);
+ processGroups.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity);
}
}
@@ -121,11 +128,7 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity
//
// Merge connections
- final Set<ConnectionEntity> mergedConnections = new HashSet<>();
- for (final List<ConnectionEntity> connectionList : connections.values()) {
- mergedConnections.add(mergeConnections(connectionList));
- }
- flowDto.setConnections(mergedConnections);
+ ConnectionsEntityMerger.mergeConnections(clientConnections, connections);
// Merge funnel statuses
final Set<FunnelEntity> mergedFunnels = new HashSet<>();
@@ -135,18 +138,10 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity
flowDto.setFunnels(mergedFunnels);
// Merge input ports
- final Set<PortEntity> mergedInputPorts = new HashSet<>();
- for (final List<PortEntity> portList : inputPorts.values()) {
- mergedInputPorts.add(mergePorts(portList));
- }
- flowDto.setInputPorts(mergedInputPorts);
+ PortsEntityMerger.mergePorts(clientInputPorts, inputPorts);
// Merge output ports
- final Set<PortEntity> mergedOutputPorts = new HashSet<>();
- for (final List<PortEntity> portList : outputPorts.values()) {
- mergedOutputPorts.add(mergePorts(portList));
- }
- flowDto.setOutputPorts(mergedOutputPorts);
+ PortsEntityMerger.mergePorts(clientOutputPorts, outputPorts);
// Merge labels
final Set<LabelEntity> mergedLabels = new HashSet<>();
@@ -155,57 +150,14 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity
}
flowDto.setLabels(mergedLabels);
-
// Merge processors
- final Set<ProcessorEntity> mergedProcessors = new HashSet<>();
- for (final List<ProcessorEntity> processorList : processors.values()) {
- mergedProcessors.add(mergeProcessors(processorList));
- }
- flowDto.setProcessors(mergedProcessors);
-
+ ProcessorsEntityMerger.mergeProcessors(clientProcessors, processors);
// Merge Remote Process Groups
- final Set<RemoteProcessGroupEntity> mergedRpgs = new HashSet<>();
- for (final List<RemoteProcessGroupEntity> rpgList : rpgs.values()) {
- mergedRpgs.add(mergeRemoteProcessGroups(rpgList));
- }
- flowDto.setRemoteProcessGroups(mergedRpgs);
-
+ RemoteProcessGroupsEntityMerger.mergeRemoteProcessGroups(clientRemoteProcessGroups, rpgs);
// Merge Process Groups
- final Set<ProcessGroupEntity> mergedGroups = new HashSet<>();
- for (final List<ProcessGroupEntity> groupList : processGroups.values()) {
- mergedGroups.add(mergeProcessGroups(groupList));
- }
- flowDto.setProcessGroups(mergedGroups);
- }
-
- private ConnectionEntity mergeConnections(final List<ConnectionEntity> connections) {
- final ConnectionEntity merged = connections.get(0);
- final ConnectionStatusDTO statusDto = merged.getStatus();
- statusDto.setNodeSnapshots(null);
-
- for (final ConnectionEntity entity : connections) {
- if (entity != merged) {
- StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot());
- }
- }
-
- return merged;
- }
-
- private PortEntity mergePorts(final List<PortEntity> ports) {
- final PortEntity merged = ports.get(0);
- final PortStatusDTO statusDto = merged.getStatus();
- statusDto.setNodeSnapshots(null);
-
- for (final PortEntity entity : ports) {
- if (entity != merged) {
- StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot());
- }
- }
-
- return merged;
+ ProcessGroupsEntityMerger.mergeProcessGroups(clientProcessGroups, processGroups);
}
private FunnelEntity mergeFunnels(final List<FunnelEntity> funnels) {
@@ -215,53 +167,4 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity
private LabelEntity mergeLabels(final List<LabelEntity> labels) {
return labels.get(0);
}
-
- private ProcessorEntity mergeProcessors(final List<ProcessorEntity> processors) {
- final ProcessorEntity merged = processors.get(0);
- final ProcessorStatusDTO statusDto = merged.getStatus();
- statusDto.setNodeSnapshots(null);
-
- for (final ProcessorEntity entity : processors) {
- if (entity != merged) {
- StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot());
- }
- }
-
- return merged;
- }
-
-
- private RemoteProcessGroupEntity mergeRemoteProcessGroups(final List<RemoteProcessGroupEntity> rpgs) {
- final RemoteProcessGroupEntity merged = rpgs.get(0);
- final RemoteProcessGroupStatusDTO statusDto = merged.getStatus();
- statusDto.setNodeSnapshots(null);
-
- for (final RemoteProcessGroupEntity entity : rpgs) {
- if (entity != merged) {
- StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot());
- }
- }
-
- return merged;
- }
-
- private ProcessGroupEntity mergeProcessGroups(final List<ProcessGroupEntity> groups) {
- final ProcessGroupEntity merged = groups.get(0);
- final ProcessGroupStatusDTO statusDto = merged.getStatus();
- statusDto.setNodeSnapshots(null);
-
- for (final ProcessGroupEntity entity : groups) {
- if (entity != merged) {
- StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot());
- }
- }
-
- // We merge only the statuses of the Process Groups. The child components are not
- // necessary for a FlowProcessGroupDTO, so we just ensure that they are null
- if (merged.getComponent() != null) {
- merged.getComponent().setContents(null);
- }
-
- return merged;
- }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java
new file mode 100644
index 0000000..b206508
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PortsEntityMerger;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.InputPortsEntity;
+import org.apache.nifi.web.api.entity.PortEntity;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class InputPortsEndpointMerger implements EndpointResponseMerger {
+ public static final Pattern INPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/input-ports");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && INPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ @Override
+ public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final InputPortsEntity responseEntity = clientResponse.getClientResponse().getEntity(InputPortsEntity.class);
+ final Set<PortEntity> portEntities = responseEntity.getInputPorts();
+
+ final Map<String, Map<NodeIdentifier, PortEntity>> entityMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final InputPortsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(InputPortsEntity.class);
+ final Set<PortEntity> nodePortEntities = nodeResponseEntity.getInputPorts();
+
+ for (final PortEntity nodePortEntity : nodePortEntities) {
+ final NodeIdentifier nodeId = nodeResponse.getNodeId();
+ Map<NodeIdentifier, PortEntity> innerMap = entityMap.get(nodeId);
+ if (innerMap == null) {
+ innerMap = new HashMap<>();
+ entityMap.put(nodePortEntity.getId(), innerMap);
+ }
+
+ innerMap.put(nodeResponse.getNodeId(), nodePortEntity);
+ }
+ }
+
+ PortsEntityMerger.mergePorts(portEntities, entityMap);
+
+ // create a new client response
+ return new NodeResponse(clientResponse, responseEntity);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java
new file mode 100644
index 0000000..753c406
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PortsEntityMerger;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.OutputPortsEntity;
+import org.apache.nifi.web.api.entity.PortEntity;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class OutputPortsEndpointMerger implements EndpointResponseMerger {
+ public static final Pattern OUTPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/output-ports");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && OUTPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ @Override
+ public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final OutputPortsEntity responseEntity = clientResponse.getClientResponse().getEntity(OutputPortsEntity.class);
+ final Set<PortEntity> portEntities = responseEntity.getOutputPorts();
+
+ final Map<String, Map<NodeIdentifier, PortEntity>> entityMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final OutputPortsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(OutputPortsEntity.class);
+ final Set<PortEntity> nodePortEntities = nodeResponseEntity.getOutputPorts();
+
+ for (final PortEntity nodePortEntity : nodePortEntities) {
+ final NodeIdentifier nodeId = nodeResponse.getNodeId();
+ Map<NodeIdentifier, PortEntity> innerMap = entityMap.get(nodeId);
+ if (innerMap == null) {
+ innerMap = new HashMap<>();
+ entityMap.put(nodePortEntity.getId(), innerMap);
+ }
+
+ innerMap.put(nodeResponse.getNodeId(), nodePortEntity);
+ }
+ }
+
+ PortsEntityMerger.mergePorts(portEntities, entityMap);
+
+ // create a new client response
+ return new NodeResponse(clientResponse, responseEntity);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java
new file mode 100644
index 0000000..8ea6384
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PortEntityMerger;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.PortEntity;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class PortEndpointMerger extends AbstractSingleEntityEndpoint<PortEntity> implements EndpointResponseMerger {
+ public static final Pattern INPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/input-ports");
+ public static final Pattern INPUT_PORT_URI_PATTERN = Pattern.compile("/nifi-api/input-ports/[a-f0-9\\-]{36}");
+
+ public static final Pattern OUTPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/output-ports");
+ public static final Pattern OUTPUT_PORT_URI_PATTERN = Pattern.compile("/nifi-api/output-ports/[a-f0-9\\-]{36}");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return canHandleInputPort(uri, method) || canHandleOutputPort(uri, method);
+ }
+
+ private boolean canHandleInputPort(final URI uri, final String method) {
+ if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && (INPUT_PORT_URI_PATTERN.matcher(uri.getPath()).matches())) {
+ return true;
+ } else if ("POST".equalsIgnoreCase(method) && INPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches()) {
+ return true;
+ }
+
+ return false;
+ }
+
+ private boolean canHandleOutputPort(final URI uri, final String method) {
+ if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && (OUTPUT_PORT_URI_PATTERN.matcher(uri.getPath()).matches())) {
+ return true;
+ } else if ("POST".equalsIgnoreCase(method) && OUTPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches()) {
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ protected Class<PortEntity> getEntityClass() {
+ return PortEntity.class;
+ }
+
+
+ @Override
+ protected void mergeResponses(final PortEntity clientEntity, final Map<NodeIdentifier, PortEntity> entityMap,
+ final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses) {
+
+ PortEntityMerger.mergePorts(clientEntity, entityMap);
+ }
+}