NIFIREG-395 - Implemented the ability to import and export versioned …

nifi-registry/nifi-registry-core/nifi-registry-web-api/pom.xml

@@ -247,6 +247,15 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <groupId>org.apache.rat</groupId>
+                <artifactId>apache-rat-plugin</artifactId>
+                <configuration>
+                    <excludes combine.children="append">
+                        <exclude>src/test/resources/test-versioned-flow-snapshot.json</exclude>
+                    </excludes>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 

nifi-registry/nifi-registry-core/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java

@@ -24,11 +24,15 @@
 import io.swagger.annotations.Authorization;
 import io.swagger.annotations.Extension;
 import io.swagger.annotations.ExtensionProperty;
+import java.net.URI;
+import javax.ws.rs.HeaderParam;
+import javax.ws.rs.core.HttpHeaders;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.nifi.registry.bucket.BucketItem;
 import org.apache.nifi.registry.diff.VersionedFlowDifference;
 import org.apache.nifi.registry.event.EventFactory;
 import org.apache.nifi.registry.event.EventService;
+import org.apache.nifi.registry.web.service.ExportedVersionedFlowSnapshot;
 import org.apache.nifi.registry.flow.VersionedFlow;
 import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
 import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
@@ -291,6 +295,44 @@ public Response createFlowVersion(
         return Response.status(Response.Status.OK).entity(createdSnapshot).build();
     }
 
+    @POST
+    @Path("{flowId}/versions/import")
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    @ApiOperation(
+            value = "Import flow version",
+            notes = "Import the next version of a flow. The version number of the object being created will be the " +
+                    "next available version integer. Flow versions are immutable after they are created.",
+            response = VersionedFlowSnapshot.class,
+            extensions = {
+                    @Extension(name = "access-policy", properties = {
+                            @ExtensionProperty(name = "action", value = "write"),
+                            @ExtensionProperty(name = "resource", value = "/buckets/{bucketId}") })
+            }
+    )
+    @ApiResponses({
+            @ApiResponse(code = 201, message = HttpStatusMessages.MESSAGE_201),
+            @ApiResponse(code = 400, message = HttpStatusMessages.MESSAGE_400),
+            @ApiResponse(code = 401, message = HttpStatusMessages.MESSAGE_401),
+            @ApiResponse(code = 403, message = HttpStatusMessages.MESSAGE_403),
+            @ApiResponse(code = 404, message = HttpStatusMessages.MESSAGE_404),
+            @ApiResponse(code = 409, message = HttpStatusMessages.MESSAGE_409) })
+    public Response importVersionedFlow(
+            @PathParam("bucketId")
+            @ApiParam("The bucket identifier")
+            final String bucketId,
+            @PathParam("flowId")
+            @ApiParam(value = "The flow identifier")
+            final String flowId,
+            @ApiParam("file") final VersionedFlowSnapshot versionedFlowSnapshot,
+            @HeaderParam("Comments") final String comments) {
+
+        final VersionedFlowSnapshot createdSnapshot = serviceFacade.importVersionedFlowSnapshot(versionedFlowSnapshot, bucketId, flowId, comments);
+        publish(EventFactory.flowVersionCreated(createdSnapshot));
+        String locationUri = createdSnapshot.getSnapshotMetadata().getLink().getUri().getPath();
+        return generateCreatedResponse(URI.create(locationUri), createdSnapshot).build();
+    }
+
     @GET
     @Path("{flowId}/versions")
     @Consumes(MediaType.WILDCARD)
@@ -385,6 +427,47 @@ public Response getLatestFlowVersionMetadata(
         return Response.status(Response.Status.OK).entity(latest).build();
     }
 
+    @GET
+    @Path("{flowId}/versions/{versionNumber: \\d+}/export")
+    @Consumes(MediaType.WILDCARD)
+    @Produces(MediaType.APPLICATION_JSON)
+    @ApiOperation(
+            value = "Exports specified bucket flow version content",
+            notes = "Exports the specified version of a flow, including the metadata and content of the flow.",
+            response = VersionedFlowSnapshot.class,
+            extensions = {
+                    @Extension(name = "access-policy", properties = {
+                            @ExtensionProperty(name = "action", value = "read"),
+                            @ExtensionProperty(name = "resource", value = "/buckets/{bucketId}")})
+            }
+    )
+    @ApiResponses({
+            @ApiResponse(code = 401, message = HttpStatusMessages.MESSAGE_401),
+            @ApiResponse(code = 403, message = HttpStatusMessages.MESSAGE_403),
+            @ApiResponse(code = 404, message = HttpStatusMessages.MESSAGE_404),
+            @ApiResponse(code = 409, message = HttpStatusMessages.MESSAGE_409)})
+    public Response exportVersionedFlow(
+            @PathParam("bucketId")
+            @ApiParam("The bucket identifier") final String bucketId,
+            @PathParam("flowId")
+            @ApiParam("The flow identifier") final String flowId,
+            @PathParam("versionNumber")
+            @ApiParam("The version number") final Integer versionNumber) {
+
+        final ExportedVersionedFlowSnapshot exportedVersionedFlowSnapshot = serviceFacade.exportFlowSnapshot(bucketId, flowId, versionNumber);
+
+        final VersionedFlowSnapshot versionedFlowSnapshot = exportedVersionedFlowSnapshot.getVersionedFlowSnapshot();
+
+        final String contentDisposition = String.format(
+                "attachment; filename=\"%s\"",
+                exportedVersionedFlowSnapshot.getFilename());
+
+        return generateOkResponse(versionedFlowSnapshot)
+                .header(HttpHeaders.CONTENT_DISPOSITION, contentDisposition)
+                .header("Filename", exportedVersionedFlowSnapshot.getFilename())
+                .build();
+    }
+
     @GET
     @Path("{flowId}/versions/{versionNumber: \\d+}")
     @Consumes(MediaType.WILDCARD)

nifi-registry/nifi-registry-core/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/HttpStatusMessages.java

@@ -18,6 +18,9 @@
 
 class HttpStatusMessages {
 
+    /* 2xx messages */
+    static final String MESSAGE_201 = "The resource has been successfully created.";
+
     /* 4xx messages */
     static final String MESSAGE_400 = "NiFi Registry was unable to complete the request because it was invalid. The request should not be retried without modification.";
     static final String MESSAGE_401 = "Client could not be authenticated.";

nifi-registry/nifi-registry-core/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/service/ExportedVersionedFlowSnapshot.java

@@ -0,0 +1,65 @@
+/*
+ * 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.registry.web.service;
+
+import io.swagger.annotations.ApiModel;
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+
+/**
+ * <p>
+ * Represents a snapshot of a versioned flow and a filename for exporting the flow. A versioned flow may change many times
+ * over the course of its life. This flow is saved to the registry with information such as its name, a description,
+ * and each version of the flow.
+ * </p>
+ *
+ * @see VersionedFlowSnapshot
+ */
+@ApiModel
+public class ExportedVersionedFlowSnapshot {
+
+    @Valid
+    @NotNull
+    private VersionedFlowSnapshot versionedFlowSnapshot;
+
+    @Valid
+    @NotNull
+    private String filename;
+
+    public ExportedVersionedFlowSnapshot(final VersionedFlowSnapshot versionedFlowSnapshot, final String filename) {
+        this.versionedFlowSnapshot = versionedFlowSnapshot;
+        this.filename = filename;
+    }
+
+    public void setVersionedFlowSnapshot(VersionedFlowSnapshot versionedFlowSnapshot) {
+        this.versionedFlowSnapshot = versionedFlowSnapshot;
+    }
+
+    public void setFilename(String filename) {
+        this.filename = filename;
+    }
+
+    public VersionedFlowSnapshot getVersionedFlowSnapshot() {
+        return versionedFlowSnapshot;
+    }
+
+    public String getFilename() {
+        return filename;
+    }
+}

nifi-registry/nifi-registry-core/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/service/ServiceFacade.java

@@ -102,6 +102,10 @@ public interface ServiceFacade {
 
     VersionedFlowSnapshot getLatestFlowSnapshot(String flowIdentifier);
 
+    VersionedFlowSnapshot importVersionedFlowSnapshot(VersionedFlowSnapshot versionedFlowSnapshot, String bucketIdentifier, String flowIdentifier, String comments);
+
+    ExportedVersionedFlowSnapshot exportFlowSnapshot(String bucketIdentifier, String flowIdentifier, Integer versionNumber);
+
     SortedSet<VersionedFlowSnapshotMetadata> getFlowSnapshots(String bucketIdentifier, String flowIdentifier);
 
     SortedSet<VersionedFlowSnapshotMetadata> getFlowSnapshots(String flowIdentifier);

nifi-registry/nifi-registry-core/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/service/StandardServiceFacade.java

@@ -109,6 +109,8 @@ public class StandardServiceFacade implements ServiceFacade {
     private final PermissionsService permissionsService;
     private final LinkService linkService;
 
+    private static final int LATEST_VERSION = -1;
+
     @Autowired
     public StandardServiceFacade(final RegistryService registryService,
                                  final ExtensionService extensionService,
@@ -360,6 +362,48 @@ public VersionedFlowSnapshot getLatestFlowSnapshot(final String flowIdentifier)
         return lastSnapshot;
     }
 
+    @Override
+    public VersionedFlowSnapshot importVersionedFlowSnapshot(VersionedFlowSnapshot versionedFlowSnapshot, String bucketIdentifier,
+                                                             String flowIdentifier, String comments) {
+        // set new snapshotMetadata
+        final VersionedFlowSnapshotMetadata metadata = new VersionedFlowSnapshotMetadata();
+        metadata.setBucketIdentifier(bucketIdentifier);
+        metadata.setFlowIdentifier(flowIdentifier);
+        metadata.setVersion(LATEST_VERSION);
+
+        // if there are new comments, then set it
+        // otherwise, keep the original comments
+        if (StringUtils.isNotBlank(comments)) {
+            metadata.setComments(comments);
+        } else if (versionedFlowSnapshot.getSnapshotMetadata() != null && StringUtils.isNotBlank(versionedFlowSnapshot.getSnapshotMetadata().getComments())) {
+            metadata.setComments(versionedFlowSnapshot.getSnapshotMetadata().getComments());
+        }
+
+        versionedFlowSnapshot.setSnapshotMetadata(metadata);
+
+        final String userIdentity = NiFiUserUtils.getNiFiUserIdentity();
+        versionedFlowSnapshot.getSnapshotMetadata().setAuthor(userIdentity);
+
+        return createFlowSnapshot(versionedFlowSnapshot);
+    }
+
+    @Override
+    public ExportedVersionedFlowSnapshot exportFlowSnapshot(String bucketIdentifier, String flowIdentifier, Integer versionNumber) {
+        final VersionedFlowSnapshot versionedFlowSnapshot = getFlowSnapshot(bucketIdentifier, flowIdentifier, versionNumber);
+
+        String flowName = versionedFlowSnapshot.getFlow().getName();
+        final String dashFlowName = flowName.replaceAll("\\s", "-");
+        final String filename = String.format("%s-version-%d.json", dashFlowName, versionedFlowSnapshot.getSnapshotMetadata().getVersion());
+
+        versionedFlowSnapshot.setFlow(null);
+        versionedFlowSnapshot.setBucket(null);
+        versionedFlowSnapshot.getSnapshotMetadata().setBucketIdentifier(null);
+        versionedFlowSnapshot.getSnapshotMetadata().setFlowIdentifier(null);
+        versionedFlowSnapshot.getSnapshotMetadata().setLink(null);
+
+        return new ExportedVersionedFlowSnapshot(versionedFlowSnapshot, filename);
+    }
+
     @Override
     public SortedSet<VersionedFlowSnapshotMetadata> getFlowSnapshots(final String bucketIdentifier, final String flowIdentifier) {
         authorizeBucketAccess(RequestAction.READ, bucketIdentifier);