You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2017/09/21 13:50:37 UTC
[3/7] nifi-registry git commit: NIFIREG-18 Initial plumbling for H2
database - Setup Flyway with initial migration to define tables - Setup
entity classes with repositories - Setup unit testing for repositories -
Removed existing MetadataProvider concept
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowMetadata.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowMetadata.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowMetadata.java
deleted file mode 100644
index cfef423..0000000
--- a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowMetadata.java
+++ /dev/null
@@ -1,174 +0,0 @@
-/*
- * 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.metadata;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashSet;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * Standard immutable implementation of FlowMetadata.
- */
-public class StandardFlowMetadata implements FlowMetadata {
-
- private final String identifier;
- private final String name;
- private final String bucketIdentifier;
- private final long createdTimestamp;
- private final long modifiedTimestamp;
- private final String description;
- private final Set<FlowSnapshotMetadata> snapshotMetadata;
- private final Map<String,FlowSnapshotMetadata> snapshotMetadataByVersion;
-
- private StandardFlowMetadata(final Builder builder) {
- this.identifier = builder.identifier;
- this.name = builder.name;
- this.bucketIdentifier = builder.bucketIdentifier;
- this.createdTimestamp = builder.createdTimestamp;
- this.modifiedTimestamp = builder.modifiedTimestamp;
- this.description = builder.description;
- this.snapshotMetadata = Collections.unmodifiableSet(
- builder.snapshotMetadata == null
- ? Collections.emptySet() : new LinkedHashSet<>(builder.snapshotMetadata));
-
- final Map<String,FlowSnapshotMetadata> tempMetadataMap = new HashMap<>();
- this.snapshotMetadata.stream().forEach(s -> tempMetadataMap.put(String.valueOf(s.getVersion()), s));
- this.snapshotMetadataByVersion = Collections.unmodifiableMap(tempMetadataMap);
- }
-
- @Override
- public String getIdentifier() {
- return identifier;
- }
-
- @Override
- public String getName() {
- return name;
- }
-
- @Override
- public String getBucketIdentifier() {
- return bucketIdentifier;
- }
-
- @Override
- public long getCreatedTimestamp() {
- return createdTimestamp;
- }
-
- @Override
- public long getModifiedTimestamp() {
- return modifiedTimestamp;
- }
-
- @Override
- public String getDescription() {
- return description;
- }
-
- @Override
- public Set<FlowSnapshotMetadata> getSnapshotMetadata() {
- return snapshotMetadata;
- }
-
- @Override
- public FlowSnapshotMetadata getSnapshot(int version) {
- return snapshotMetadataByVersion.get(String.valueOf(version));
- }
-
- public static class Builder {
-
- private String identifier;
- private String name;
- private String bucketIdentifier;
- private long createdTimestamp;
- private long modifiedTimestamp;
- private String description;
- private Set<FlowSnapshotMetadata> snapshotMetadata = new LinkedHashSet<>();
-
- public Builder() {
-
- }
-
- public Builder(FlowMetadata flowMetadata) {
- identifier(flowMetadata.getIdentifier());
- name(flowMetadata.getName());
- bucketIdentifier(flowMetadata.getBucketIdentifier());
- created(flowMetadata.getCreatedTimestamp());
- modified(flowMetadata.getModifiedTimestamp());
- description(flowMetadata.getDescription());
- addSnapshots(flowMetadata.getSnapshotMetadata());
- }
-
- public Builder identifier(final String identifier) {
- this.identifier = identifier;
- return this;
- }
-
- public Builder name(final String name) {
- this.name = name;
- return this;
- }
-
- public Builder bucketIdentifier(final String bucketIdentifier) {
- this.bucketIdentifier = bucketIdentifier;
- return this;
- }
-
- public Builder created(final long createdTimestamp) {
- this.createdTimestamp = createdTimestamp;
- return this;
- }
-
- public Builder modified(final long modifiedTimestamp) {
- this.modifiedTimestamp = modifiedTimestamp;
- return this;
- }
-
- public Builder description(final String description) {
- this.description = description;
- return this;
- }
-
- public Builder addSnapshot(final FlowSnapshotMetadata snapshotMetadata) {
- if (snapshotMetadata != null) {
- this.snapshotMetadata.add(snapshotMetadata);
- }
- return this;
- }
-
- public Builder addSnapshots(final Collection<FlowSnapshotMetadata> snapshotMetadata) {
- if (snapshotMetadata != null) {
- this.snapshotMetadata.addAll(snapshotMetadata);
- }
- return this;
- }
-
- public Builder clearSnapshots() {
- this.snapshotMetadata.clear();
- return this;
- }
-
- public StandardFlowMetadata build() {
- return new StandardFlowMetadata(this);
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowSnapshotMetadata.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowSnapshotMetadata.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowSnapshotMetadata.java
deleted file mode 100644
index f8d09ce..0000000
--- a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/metadata/StandardFlowSnapshotMetadata.java
+++ /dev/null
@@ -1,136 +0,0 @@
-/*
- * 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.metadata;
-
-import java.util.Objects;
-
-/**
- * Standard immutable implementation of FlowSnapshotMetadata.
- */
-public class StandardFlowSnapshotMetadata implements FlowSnapshotMetadata {
-
- private final String bucketIdentifier;
- private final String flowIdentifier;
- private final String flowName;
- private final int version;
- private final long createdTimestamp;
- private final String comments;
-
- private StandardFlowSnapshotMetadata(final Builder builder) {
- this.bucketIdentifier = builder.bucketIdentifier;
- this.flowIdentifier = builder.flowIdentifier;
- this.flowName = builder.flowName;
- this.version = builder.version;
- this.createdTimestamp = builder.createdTimestamp;
- this.comments = builder.comments;
- }
-
- @Override
- public String getBucketIdentifier() {
- return bucketIdentifier;
- }
-
- @Override
- public String getFlowIdentifier() {
- return flowIdentifier;
- }
-
- @Override
- public String getFlowName() {
- return flowName;
- }
-
- @Override
- public int getVersion() {
- return version;
- }
-
- @Override
- public long getCreatedTimestamp() {
- return createdTimestamp;
- }
-
- @Override
- public String getComments() {
- return comments;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(this.bucketIdentifier, this.flowIdentifier, Integer.valueOf(this.version));
- }
-
- @Override
- public boolean equals(final Object obj) {
- if (obj == null) {
- return false;
- }
- if (getClass() != obj.getClass()) {
- return false;
- }
-
- final StandardFlowSnapshotMetadata other = (StandardFlowSnapshotMetadata) obj;
-
- return Objects.equals(this.bucketIdentifier, other.bucketIdentifier)
- && Objects.equals(this.flowIdentifier, other.flowIdentifier)
- && Objects.equals(this.version, other.version);
- }
-
- public static class Builder {
-
- private String bucketIdentifier;
- private String flowIdentifier;
- private String flowName;
- private int version;
- private long createdTimestamp;
- private String comments;
-
- public Builder bucketIdentifier(final String bucketIdentifier) {
- this.bucketIdentifier = bucketIdentifier;
- return this;
- }
-
- public Builder flowIdentifier(final String flowIdentifier) {
- this.flowIdentifier = flowIdentifier;
- return this;
- }
-
- public Builder flowName(final String flowName) {
- this.flowName = flowName;
- return this;
- }
-
- public Builder version(final int version) {
- this.version = version;
- return this;
- }
-
- public Builder created(long createdTimestamp) {
- this.createdTimestamp = createdTimestamp;
- return this;
- }
-
- public Builder comments(String comments) {
- this.comments = comments;
- return this;
- }
-
- public StandardFlowSnapshotMetadata build() {
- return new StandardFlowSnapshotMetadata(this);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-impl/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-impl/pom.xml b/nifi-registry-provider-impl/pom.xml
deleted file mode 100644
index 2405ba9..0000000
--- a/nifi-registry-provider-impl/pom.xml
+++ /dev/null
@@ -1,94 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- 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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <parent>
- <groupId>org.apache.nifi.registry</groupId>
- <artifactId>nifi-registry</artifactId>
- <version>0.0.1-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
- <artifactId>nifi-registry-provider-impl</artifactId>
- <packaging>jar</packaging>
-
- <build>
- <resources>
- <resource>
- <directory>src/main/resources</directory>
- </resource>
- <resource>
- <directory>src/main/xsd</directory>
- </resource>
- </resources>
- <plugins>
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>jaxb2-maven-plugin</artifactId>
- <executions>
- <execution>
- <id>current</id>
- <goals>
- <goal>xjc</goal>
- </goals>
- <configuration>
- <packageName>org.apache.nifi.registry.metadata.generated</packageName>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-checkstyle-plugin</artifactId>
- <configuration>
- <excludes>**/metadata/generated/*.java,</excludes>
- </configuration>
- </plugin>
- </plugins>
- </build>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.nifi.registry</groupId>
- <artifactId>nifi-registry-provider-api</artifactId>
- <version>0.0.1-SNAPSHOT</version>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi.registry</groupId>
- <artifactId>nifi-registry-utils</artifactId>
- <version>0.0.1-SNAPSHOT</version>
- </dependency>
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </dependency>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <version>4.12</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <version>2.7.22</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
-</project>
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/flow/FileSystemFlowPersistenceProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/flow/FileSystemFlowPersistenceProvider.java b/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/flow/FileSystemFlowPersistenceProvider.java
deleted file mode 100644
index db09a6d..0000000
--- a/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/flow/FileSystemFlowPersistenceProvider.java
+++ /dev/null
@@ -1,166 +0,0 @@
-/*
- * 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.flow;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.registry.provider.ProviderConfigurationContext;
-import org.apache.nifi.registry.provider.ProviderCreationException;
-import org.apache.nifi.registry.util.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Map;
-
-/**
- * A FlowPersistenceProvider that uses the local filesystem for storage.
- */
-public class FileSystemFlowPersistenceProvider implements FlowPersistenceProvider {
-
- static final Logger LOGGER = LoggerFactory.getLogger(FileSystemFlowPersistenceProvider.class);
-
- static final String FLOW_STORAGE_DIR_PROP = "Flow Storage Directory";
-
- static final String SNAPSHOT_EXTENSION = ".snapshot";
-
- private File flowStorageDir;
-
- @Override
- public void onConfigured(final ProviderConfigurationContext configurationContext) throws ProviderCreationException {
- final Map<String,String> props = configurationContext.getProperties();
- if (!props.containsKey(FLOW_STORAGE_DIR_PROP)) {
- throw new ProviderCreationException("The property " + FLOW_STORAGE_DIR_PROP + " must be provided");
- }
-
- final String flowStorageDirValue = props.get(FLOW_STORAGE_DIR_PROP);
- if (StringUtils.isBlank(flowStorageDirValue)) {
- throw new ProviderCreationException("The property " + FLOW_STORAGE_DIR_PROP + " cannot be null or blank");
- }
-
- try {
- flowStorageDir = new File(flowStorageDirValue);
- FileUtils.ensureDirectoryExistAndCanReadAndWrite(flowStorageDir);
- LOGGER.info("Configured FileSystemFlowPersistenceProvider with Flow Storage Directory {}", new Object[] {flowStorageDir.getAbsolutePath()});
- } catch (IOException e) {
- throw new ProviderCreationException(e);
- }
- }
-
- @Override
- public synchronized void saveSnapshot(final FlowSnapshotContext context, final byte[] content) throws FlowPersistenceException {
- final File bucketDir = new File(flowStorageDir, context.getBucketId());
- try {
- FileUtils.ensureDirectoryExistAndCanReadAndWrite(bucketDir);
- } catch (IOException e) {
- throw new FlowPersistenceException("Error accessing bucket directory at " + bucketDir.getAbsolutePath(), e);
- }
-
- final File flowDir = new File(bucketDir, context.getFlowId());
- try {
- FileUtils.ensureDirectoryExistAndCanReadAndWrite(flowDir);
- } catch (IOException e) {
- throw new FlowPersistenceException("Error accessing flow directory at " + flowDir.getAbsolutePath(), e);
- }
-
- final String versionString = String.valueOf(context.getVersion());
- final File versionDir = new File(flowDir, versionString);
- try {
- FileUtils.ensureDirectoryExistAndCanReadAndWrite(versionDir);
- } catch (IOException e) {
- throw new FlowPersistenceException("Error accessing version directory at " + versionDir.getAbsolutePath(), e);
- }
-
- final File versionFile = new File(versionDir, versionString + SNAPSHOT_EXTENSION);
- if (versionFile.exists()) {
- throw new FlowPersistenceException("Unable to save, a snapshot already exists with version " + versionString);
- }
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Saving snapshot with filename {}", new Object[] {versionFile.getAbsolutePath()});
- }
-
- try (final OutputStream out = new FileOutputStream(versionFile)) {
- out.write(content);
- out.flush();
- } catch (Exception e) {
- throw new FlowPersistenceException("Unable to write snapshot to disk due to " + e.getMessage(), e);
- }
- }
-
- @Override
- public synchronized byte[] getSnapshot(final String bucketId, final String flowId, final int version) throws FlowPersistenceException {
- final File snapshotFile = getSnapshotFile(bucketId, flowId, version);
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Retrieving snapshot with filename {}", new Object[] {snapshotFile.getAbsolutePath()});
- }
-
- if (!snapshotFile.exists()) {
- return null;
- }
-
- try (final InputStream in = new FileInputStream(snapshotFile)){
- return IOUtils.toByteArray(in);
- } catch (IOException e) {
- throw new FlowPersistenceException("Error reading snapshot file: " + snapshotFile.getAbsolutePath(), e);
- }
- }
-
- @Override
- public synchronized void deleteSnapshots(final String bucketId, final String flowId) throws FlowPersistenceException {
- final File flowDir = new File(flowStorageDir, bucketId + "/" + flowId);
- if (!flowDir.exists()) {
- LOGGER.debug("Snapshot directory does not exist at {}", new Object[] {flowDir.getAbsolutePath()});
- return;
- }
-
- try {
- org.apache.commons.io.FileUtils.cleanDirectory(flowDir);
- } catch (IOException e) {
- throw new FlowPersistenceException("Error deleting snapshots at " + flowDir.getAbsolutePath(), e);
- }
- }
-
- @Override
- public synchronized void deleteSnapshot(final String bucketId, final String flowId, final int version) throws FlowPersistenceException {
- final File snapshotFile = getSnapshotFile(bucketId, flowId, version);
- if (!snapshotFile.exists()) {
- LOGGER.debug("Snapshot file does not exist at {}", new Object[] {snapshotFile.getAbsolutePath()});
- return;
- }
-
- final boolean deleted = snapshotFile.delete();
- if (!deleted) {
- throw new FlowPersistenceException("Unable to delete snapshot at " + snapshotFile.getAbsolutePath());
- }
-
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("Deleted snapshot at {}", new Object[] {snapshotFile.getAbsolutePath()});
- }
- }
-
- protected File getSnapshotFile(final String bucketId, final String flowId, final int version) {
- final String snapshotFilename = bucketId + "/" + flowId + "/" + version + "/" + version + SNAPSHOT_EXTENSION;
- return new File(flowStorageDir, snapshotFilename);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/FileSystemMetadataProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/FileSystemMetadataProvider.java b/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/FileSystemMetadataProvider.java
deleted file mode 100644
index dd3e1ec..0000000
--- a/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/FileSystemMetadataProvider.java
+++ /dev/null
@@ -1,469 +0,0 @@
-/*
- * 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.metadata;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.registry.metadata.generated.Buckets;
-import org.apache.nifi.registry.metadata.generated.Flow;
-import org.apache.nifi.registry.metadata.generated.Flows;
-import org.apache.nifi.registry.metadata.generated.Metadata;
-import org.apache.nifi.registry.provider.ProviderConfigurationContext;
-import org.apache.nifi.registry.provider.ProviderCreationException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.xml.sax.SAXException;
-
-import javax.xml.XMLConstants;
-import javax.xml.bind.JAXBContext;
-import javax.xml.bind.JAXBElement;
-import javax.xml.bind.JAXBException;
-import javax.xml.bind.Marshaller;
-import javax.xml.bind.Unmarshaller;
-import javax.xml.transform.stream.StreamSource;
-import javax.xml.validation.Schema;
-import javax.xml.validation.SchemaFactory;
-import java.io.File;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * A MetadataProvider that persists metadata to the local filesystem.
- */
-public class FileSystemMetadataProvider implements MetadataProvider {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(FileSystemMetadataProvider.class);
-
- private static final String METADATA_XSD = "/metadata.xsd";
- private static final String JAXB_GENERATED_PATH = "org.apache.nifi.registry.metadata.generated";
- private static final JAXBContext JAXB_CONTEXT = initializeJaxbContext();
-
- /**
- * Load the JAXBContext.
- */
- private static JAXBContext initializeJaxbContext() {
- try {
- return JAXBContext.newInstance(JAXB_GENERATED_PATH, FileSystemMetadataProvider.class.getClassLoader());
- } catch (JAXBException e) {
- throw new RuntimeException("Unable to create JAXBContext.", e);
- }
- }
-
- static final String METADATA_FILE_PROP = "Metadata File";
-
- private File metadataFile;
- private Schema metadataSchema;
- private final AtomicReference<MetadataHolder> metadataHolder = new AtomicReference<>(null);
-
- public FileSystemMetadataProvider() throws ProviderCreationException {
- try {
- final SchemaFactory schemaFactory = SchemaFactory.newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI);
- metadataSchema = schemaFactory.newSchema(FileSystemMetadataProvider.class.getResource(METADATA_XSD));
- } catch (SAXException e) {
- throw new ProviderCreationException("Unable to create MetadataProvider due to: " + e.getMessage(), e);
- }
- }
-
- @Override
- public void onConfigured(final ProviderConfigurationContext configurationContext) throws ProviderCreationException {
- final Map<String,String> config = configurationContext.getProperties();
- if (!config.containsKey(METADATA_FILE_PROP)) {
- throw new ProviderCreationException("The property " + METADATA_FILE_PROP + " must be provided");
- }
-
- final String metadataFileValue = config.get(METADATA_FILE_PROP);
- if (StringUtils.isBlank(metadataFileValue)) {
- throw new ProviderCreationException("The property " + METADATA_FILE_PROP + " cannot be null or blank");
- }
-
- try {
- metadataFile = new File(metadataFileValue);
- if (metadataFile.exists()) {
- LOGGER.info("Loading metadata file from {}", new Object[] {metadataFile.getAbsolutePath()});
- final Metadata metadata = unmarshallMetadata();
- metadataHolder.set(new MetadataHolder(metadata));
- } else {
- LOGGER.info("Creating new metadata file at {}", new Object[] {metadataFile.getAbsolutePath()});
-
- final Metadata metadata = new Metadata();
- metadata.setBuckets(new Buckets());
- metadata.setFlows(new Flows());
-
- saveMetadata(metadata);
- metadataHolder.set(new MetadataHolder(metadata));
- }
- } catch (Exception e) {
- throw new ProviderCreationException("Unable to configure MetadataProvider due to: " + e.getMessage(), e);
- }
- }
-
- private Metadata unmarshallMetadata() throws JAXBException {
- final Unmarshaller unmarshaller = JAXB_CONTEXT.createUnmarshaller();
- unmarshaller.setSchema(metadataSchema);
-
- final JAXBElement<Metadata> element = unmarshaller.unmarshal(new StreamSource(metadataFile), Metadata.class);
- return element.getValue();
- }
-
- private void saveMetadata(final Metadata metadata) throws JAXBException {
- final Marshaller marshaller = JAXB_CONTEXT.createMarshaller();
- marshaller.setSchema(metadataSchema);
- marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
- marshaller.marshal(metadata, metadataFile);
- }
-
- private synchronized void saveAndRefresh(final Metadata metadata) {
- try {
- saveMetadata(metadata);
- metadataHolder.set(new MetadataHolder(metadata));
- } catch (JAXBException e) {
- throw new MetadataProviderException("Unable to save metadata", e);
- }
- }
-
- @Override
- public synchronized BucketMetadata createBucket(final BucketMetadata bucket) {
- if (bucket == null) {
- throw new IllegalArgumentException("Bucket cannot be null");
- }
-
- final org.apache.nifi.registry.metadata.generated.Bucket jaxbBucket = new org.apache.nifi.registry.metadata.generated.Bucket();
- jaxbBucket.setIdentifier(bucket.getIdentifier());
- jaxbBucket.setName(bucket.getName());
- jaxbBucket.setDescription(bucket.getDescription());
- jaxbBucket.setCreatedTimestamp(bucket.getCreatedTimestamp());
-
- final MetadataHolder holder = metadataHolder.get();
-
- final Metadata metadata = holder.getMetadata();
- metadata.getBuckets().getBucket().add(jaxbBucket);
-
- saveAndRefresh(metadata);
- return metadataHolder.get().getBucketsById().get(bucket.getIdentifier());
- }
-
- @Override
- public BucketMetadata getBucketById(final String bucketIdentifier) {
- if (bucketIdentifier == null) {
- throw new IllegalArgumentException("Bucket Identifier cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- return holder.getBucketsById().get(bucketIdentifier);
- }
-
- @Override
- public BucketMetadata getBucketByName(String name) {
- if (name == null) {
- throw new IllegalArgumentException("Bucket Name cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- return holder.getBucketsByName().get(name.toLowerCase());
- }
-
- @Override
- public Set<BucketMetadata> getBuckets() {
- final MetadataHolder holder = metadataHolder.get();
- final Map<String,BucketMetadata> bucketsBydId = holder.getBucketsById();
- return new HashSet<>(bucketsBydId.values());
- }
-
- @Override
- public synchronized BucketMetadata updateBucket(final BucketMetadata bucket) {
- if (bucket == null) {
- throw new IllegalArgumentException("Bucket cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- final Buckets buckets = holder.getMetadata().getBuckets();
-
- final org.apache.nifi.registry.metadata.generated.Bucket jaxbBucket = buckets.getBucket().stream()
- .filter(b -> bucket.getIdentifier().equals(b.getIdentifier()))
- .findFirst()
- .orElse(null);
-
- if (jaxbBucket == null) {
- return null;
- }
-
- jaxbBucket.setName(bucket.getName());
- jaxbBucket.setDescription(bucket.getDescription());
-
- saveAndRefresh(holder.getMetadata());
- return metadataHolder.get().getBucketsById().get(bucket.getIdentifier());
- }
-
- @Override
- public synchronized void deleteBucket(final String bucketIdentifier) {
- if (bucketIdentifier == null) {
- throw new IllegalArgumentException("Bucket Identifier cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- final Flows flows = holder.getMetadata().getFlows();
- final Buckets buckets = holder.getMetadata().getBuckets();
-
- // first remove any flow that reference the bucket
- boolean deletedFlow = false;
- final Iterator<Flow> flowIterator = flows.getFlow().iterator();
- while (flowIterator.hasNext()) {
- final Flow flow = flowIterator.next();
- if (flow.getBucketIdentifier().equals(bucketIdentifier)) {
- flowIterator.remove();
- deletedFlow = true;
- }
- }
-
- // now delete the actual bucket
- boolean deleteBucket = false;
- final Iterator<org.apache.nifi.registry.metadata.generated.Bucket> bucketIterator = buckets.getBucket().iterator();
- while (bucketIterator.hasNext()) {
- final org.apache.nifi.registry.metadata.generated.Bucket bucket = bucketIterator.next();
- if (bucket.getIdentifier().equals(bucketIdentifier)) {
- bucketIterator.remove();
- deleteBucket = true;
- break;
- }
- }
-
- if (deletedFlow || deleteBucket) {
- saveAndRefresh(holder.getMetadata());
- }
- }
-
- @Override
- public synchronized FlowMetadata createFlow(final String bucketIdentifier, final FlowMetadata versionedFlow) {
- if (bucketIdentifier == null) {
- throw new IllegalArgumentException("Bucket Identifier cannot be blank");
- }
-
- if (versionedFlow == null) {
- throw new IllegalArgumentException("Versioned Flow cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
-
- final BucketMetadata bucket = holder.getBucketsById().get(bucketIdentifier);
- if (bucket == null) {
- throw new IllegalStateException("Unable to create Versioned Flow because Bucket does not exist with id " + bucketIdentifier);
- }
-
- final Flow jaxbFlow = new Flow();
- jaxbFlow.setIdentifier(versionedFlow.getIdentifier());
- jaxbFlow.setName(versionedFlow.getName());
- jaxbFlow.setDescription(versionedFlow.getDescription());
- jaxbFlow.setCreatedTimestamp(versionedFlow.getCreatedTimestamp());
- jaxbFlow.setModifiedTimestamp(versionedFlow.getModifiedTimestamp());
- jaxbFlow.setBucketIdentifier(bucketIdentifier);
-
- final Metadata metadata = holder.getMetadata();
- metadata.getFlows().getFlow().add(jaxbFlow);
-
- saveAndRefresh(metadata);
- return metadataHolder.get().getFlowsById().get(versionedFlow.getIdentifier());
- }
-
- @Override
- public FlowMetadata getFlowById(final String flowIdentifier) {
- if (flowIdentifier == null) {
- throw new IllegalArgumentException("Flow Identifier cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- return holder.getFlowsById().get(flowIdentifier);
- }
-
- @Override
- public FlowMetadata getFlowByName(final String name) {
- if (name == null) {
- throw new IllegalArgumentException("Flow Name cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- return holder.getFlowsByName().get(name.toLowerCase());
- }
-
-
- @Override
- public Set<FlowMetadata> getFlows() {
- final MetadataHolder holder = metadataHolder.get();
- final Map<String,FlowMetadata> flowsById = holder.getFlowsById();
- return new HashSet<>(flowsById.values());
- }
-
- @Override
- public Set<FlowMetadata> getFlows(String bucketId) {
- final MetadataHolder holder = metadataHolder.get();
-
- final Map<String,Set<FlowMetadata>> flowsByBucket = holder.getFlowsByBucket();
- if (flowsByBucket.containsKey(bucketId)) {
- return new HashSet<>(flowsByBucket.get(bucketId));
- } else {
- return Collections.emptySet();
- }
- }
-
- @Override
- public synchronized FlowMetadata updateFlow(final FlowMetadata versionedFlow) {
- if (versionedFlow == null) {
- throw new IllegalArgumentException("Versioned Flow cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- final Flows flows = holder.getMetadata().getFlows();
-
- final Flow jaxbFlow = flows.getFlow().stream()
- .filter(f -> versionedFlow.getIdentifier().equals(f.getIdentifier()))
- .findFirst()
- .orElse(null);
-
- if (jaxbFlow == null) {
- return null;
- }
-
- // TODO should we allow changing the bucket id here, if so it needs to be passed in
- jaxbFlow.setName(versionedFlow.getName());
- jaxbFlow.setDescription(versionedFlow.getDescription());
- jaxbFlow.setModifiedTimestamp(System.currentTimeMillis());
-
- saveAndRefresh(holder.getMetadata());
- return metadataHolder.get().getFlowsById().get(versionedFlow.getIdentifier());
- }
-
- @Override
- public synchronized void deleteFlow(final String flowIdentifier) {
- if (flowIdentifier == null) {
- throw new IllegalArgumentException("Flow Identifier cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- final Flows flows = holder.getMetadata().getFlows();
-
- boolean deleted = false;
- final Iterator<Flow> flowIter = flows.getFlow().iterator();
-
- while (flowIter.hasNext()) {
- final Flow jaxbFlow = flowIter.next();
- if (jaxbFlow.getIdentifier().equals(flowIdentifier)) {
- flowIter.remove();
- deleted = true;
- break;
- }
- }
-
- if (deleted) {
- saveAndRefresh(holder.getMetadata());
- }
- }
-
- @Override
- public synchronized FlowSnapshotMetadata createFlowSnapshot(final FlowSnapshotMetadata flowSnapshot) {
- if (flowSnapshot == null) {
- throw new IllegalArgumentException("Versioned Flow Snapshot cannot be null");
- }
-
- final String flowIdentifier = flowSnapshot.getFlowIdentifier();
- final int snapshotVersion = flowSnapshot.getVersion();
-
- final MetadataHolder holder = metadataHolder.get();
- final Flows flows = holder.getMetadata().getFlows();
-
- final Flow jaxbFlow = flows.getFlow().stream()
- .filter(f -> flowIdentifier.equals(f.getIdentifier()))
- .findFirst()
- .orElse(null);
-
- if (jaxbFlow == null) {
- throw new IllegalStateException("Unable to create snapshot because Versioned Flow does not exist for id " + flowIdentifier);
- }
-
- final Flow.Snapshot jaxbSnapshot = new Flow.Snapshot();
- jaxbSnapshot.setVersion(flowSnapshot.getVersion());
- jaxbSnapshot.setComments(flowSnapshot.getComments());
- jaxbSnapshot.setCreatedTimestamp(flowSnapshot.getCreatedTimestamp());
-
- jaxbFlow.getSnapshot().add(jaxbSnapshot);
- saveAndRefresh(holder.getMetadata());
-
- final FlowMetadata versionedFlow = metadataHolder.get().getFlowsById().get(flowIdentifier);
- return versionedFlow.getSnapshot(snapshotVersion);
- }
-
- @Override
- public FlowSnapshotMetadata getFlowSnapshot(final String flowIdentifier, final Integer version) {
- if (flowIdentifier == null) {
- throw new IllegalArgumentException("Flow Identifier cannot be null");
- }
-
- if (version == null) {
- throw new IllegalArgumentException("Version cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
-
- final FlowMetadata versionedFlow = holder.getFlowsById().get(flowIdentifier);
- if (versionedFlow == null) {
- return null;
- }
-
- return versionedFlow.getSnapshot(version);
- }
-
- @Override
- public synchronized void deleteFlowSnapshot(final String flowIdentifier, final Integer version) {
- if (flowIdentifier == null) {
- throw new IllegalArgumentException("Flow Identifier cannot be null");
- }
-
- if (version == null) {
- throw new IllegalArgumentException("Version cannot be null");
- }
-
- final MetadataHolder holder = metadataHolder.get();
- final Flows flows = holder.getMetadata().getFlows();
-
- final Flow jaxbFlow = flows.getFlow().stream()
- .filter(f -> flowIdentifier.equals(f.getIdentifier()))
- .findFirst()
- .orElse(null);
-
- if (jaxbFlow == null) {
- return;
- }
-
- boolean deletedSnapshot = false;
- final Iterator<Flow.Snapshot> snapshotIterator = jaxbFlow.getSnapshot().iterator();
-
- while (snapshotIterator.hasNext()) {
- final Flow.Snapshot snapshot = snapshotIterator.next();
- if (snapshot.getVersion().equals(version)) {
- snapshotIterator.remove();
- deletedSnapshot = true;
- break;
- }
- }
-
- if (deletedSnapshot) {
- saveAndRefresh(holder.getMetadata());
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/MetadataHolder.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/MetadataHolder.java b/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/MetadataHolder.java
deleted file mode 100644
index e49a62b..0000000
--- a/nifi-registry-provider-impl/src/main/java/org/apache/nifi/registry/metadata/MetadataHolder.java
+++ /dev/null
@@ -1,176 +0,0 @@
-/*
- * 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.metadata;
-
-import org.apache.nifi.registry.metadata.generated.Bucket;
-import org.apache.nifi.registry.metadata.generated.Buckets;
-import org.apache.nifi.registry.metadata.generated.Flow;
-import org.apache.nifi.registry.metadata.generated.Flows;
-import org.apache.nifi.registry.metadata.generated.Metadata;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * Provides atomic access to the Metadata object.
- */
-public class MetadataHolder {
-
- private final Metadata metadata;
- private final Map<String,Set<FlowMetadata>> flowsByBucket;
- private final Map<String,FlowMetadata> flowsById;
- private final Map<String,FlowMetadata> flowsByName;
- private final Map<String,BucketMetadata> bucketsById;
- private final Map<String,BucketMetadata> bucketsByName;
-
- public MetadataHolder(final Metadata metadata) {
- this.metadata = metadata;
- this.flowsByBucket = Collections.unmodifiableMap(createFlowsByBucket(metadata));
- this.flowsByName = Collections.unmodifiableMap(createFlowsByName(flowsByBucket));
- this.flowsById = Collections.unmodifiableMap(createFlowsById(flowsByBucket));
- this.bucketsById = Collections.unmodifiableMap(createBucketsBydId(metadata, flowsByBucket));
- this.bucketsByName = Collections.unmodifiableMap(createBucketsByName(bucketsById));
- }
-
- private Map<String,BucketMetadata> createBucketsBydId(final Metadata metadata, final Map<String,Set<FlowMetadata>> flowsByBucket) {
- final Map<String,BucketMetadata> bucketsById = new HashMap<>();
-
- final Buckets buckets = metadata.getBuckets();
- if (buckets != null) {
- buckets.getBucket().stream().forEach(b -> {
- final Set<FlowMetadata> bucketFlows = flowsByBucket.get(b.getIdentifier());
- final BucketMetadata bucketMetadata = createBucketMetadata(b, bucketFlows);
- bucketsById.put(b.getIdentifier(), bucketMetadata);
- });
- }
-
- return bucketsById;
- }
-
- private Map<String,BucketMetadata> createBucketsByName(Map<String,BucketMetadata> bucketsById) {
- final Map<String,BucketMetadata> bucketsByName = new HashMap<>();
- bucketsById.values().stream().forEach(b -> bucketsByName.put(b.getName().toLowerCase(), b));
- return bucketsByName;
- }
-
- private BucketMetadata createBucketMetadata(final Bucket jaxbBucket, final Set<FlowMetadata> bucketFlows) {
- return new StandardBucketMetadata.Builder()
- .identifier(jaxbBucket.getIdentifier())
- .name(jaxbBucket.getName())
- .description(jaxbBucket.getDescription())
- .created(jaxbBucket.getCreatedTimestamp())
- .addFlows(bucketFlows)
- .build();
- }
-
- private Map<String,Set<FlowMetadata>> createFlowsByBucket(final Metadata metadata) {
- final Map<String,Set<FlowMetadata>> flowsByBucket = new HashMap<>();
-
- final Flows flows = metadata.getFlows();
- if (flows != null) {
- flows.getFlow().stream().forEach(f -> {
- Set<FlowMetadata> bucketFLows = flowsByBucket.get(f.getBucketIdentifier());
- if (bucketFLows == null) {
- bucketFLows = new HashSet<>();
- flowsByBucket.put(f.getBucketIdentifier(), bucketFLows);
- }
- bucketFLows.add(createFlowMetadata(f));
- });
- }
-
- return flowsByBucket;
- }
-
- private FlowMetadata createFlowMetadata(final Flow jaxbFlow) {
- final StandardFlowMetadata.Builder builder = new StandardFlowMetadata.Builder()
- .identifier(jaxbFlow.getIdentifier())
- .name(jaxbFlow.getName())
- .bucketIdentifier(jaxbFlow.getBucketIdentifier())
- .description(jaxbFlow.getDescription())
- .created(jaxbFlow.getCreatedTimestamp())
- .modified(jaxbFlow.getModifiedTimestamp());
-
- if (jaxbFlow.getSnapshot() != null) {
- jaxbFlow.getSnapshot().stream().forEach(s -> builder.addSnapshot(createSnapshotMetadata(jaxbFlow, s)));
- }
-
- return builder.build();
- }
-
- private FlowSnapshotMetadata createSnapshotMetadata(final Flow jaxbFlow, final Flow.Snapshot jaxbSnapshot) {
- return new StandardFlowSnapshotMetadata.Builder()
- .bucketIdentifier(jaxbFlow.getBucketIdentifier())
- .flowIdentifier(jaxbFlow.getIdentifier())
- .flowName(jaxbFlow.getName())
- .version(jaxbSnapshot.getVersion())
- .comments(jaxbSnapshot.getComments())
- .created(jaxbSnapshot.getCreatedTimestamp())
- .build();
- }
-
- private Map<String,FlowMetadata> createFlowsById(final Map<String,Set<FlowMetadata>> flowsByBucket) {
- final Map<String,FlowMetadata> flowsById = new HashMap<>();
-
- for (final Map.Entry<String,Set<FlowMetadata>> entry : flowsByBucket.entrySet()) {
- for (final FlowMetadata flowMetadata : entry.getValue()) {
- flowsById.put(flowMetadata.getIdentifier(), flowMetadata);
- }
- }
-
- return flowsById;
- }
-
- private Map<String,FlowMetadata> createFlowsByName(final Map<String,Set<FlowMetadata>> flowsByBucket) {
- final Map<String,FlowMetadata> flowsByName = new HashMap<>();
-
- for (final Map.Entry<String,Set<FlowMetadata>> entry : flowsByBucket.entrySet()) {
- for (final FlowMetadata flow : entry.getValue()) {
- flowsByName.put(flow.getName().toLowerCase(), flow);
- }
- }
-
- return flowsByName;
- }
-
- public Metadata getMetadata() {
- return metadata;
- }
-
- public Map<String,BucketMetadata> getBucketsById() {
- return bucketsById;
- }
-
- public Map<String,BucketMetadata> getBucketsByName() {
- return bucketsByName;
- }
-
- public Map<String,FlowMetadata> getFlowsById() {
- return flowsById;
- }
-
- public Map<String,FlowMetadata> getFlowsByName() {
- return flowsByName;
- }
-
- public Map<String,Set<FlowMetadata>> getFlowsByBucket() {
- return flowsByBucket;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-impl/src/main/xsd/metadata.xsd
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-impl/src/main/xsd/metadata.xsd b/nifi-registry-provider-impl/src/main/xsd/metadata.xsd
deleted file mode 100644
index 7f71745..0000000
--- a/nifi-registry-provider-impl/src/main/xsd/metadata.xsd
+++ /dev/null
@@ -1,139 +0,0 @@
-<?xml version="1.0"?>
-<!--
- 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.
--->
-<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
-
- <!-- Buckets -->
-
- <xs:complexType name="Bucket">
- <xs:sequence>
- <xs:element name="description" type="xs:string" minOccurs="0" maxOccurs="1"/>
- </xs:sequence>
- <xs:attribute name="identifier">
- <xs:simpleType>
- <xs:restriction base="xs:string">
- <xs:minLength value="1"/>
- <xs:pattern value=".*[^\s].*"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- <xs:attribute name="name">
- <xs:simpleType>
- <xs:restriction base="xs:string">
- <xs:minLength value="1"/>
- <xs:pattern value=".*[^\s].*"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- <xs:attribute name="createdTimestamp">
- <xs:simpleType>
- <xs:restriction base="xs:long">
- <xs:minInclusive value="1"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- </xs:complexType>
-
- <xs:complexType name="Buckets">
- <xs:sequence>
- <xs:element name="bucket" type="Bucket" minOccurs="0" maxOccurs="unbounded"/>
- </xs:sequence>
- </xs:complexType>
-
- <!-- Flows -->
-
- <xs:complexType name="Flow">
- <xs:sequence>
- <xs:element name="description" type="xs:string" minOccurs="0" maxOccurs="1"/>
- <xs:element name="snapshot" minOccurs="0" maxOccurs="unbounded" >
- <xs:complexType>
- <xs:sequence>
- <xs:element name="comments" type="xs:string" minOccurs="0" maxOccurs="1"/>
- </xs:sequence>
- <xs:attribute name="version">
- <xs:simpleType>
- <xs:restriction base="xs:int">
- <xs:minInclusive value="1"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- <xs:attribute name="createdTimestamp">
- <xs:simpleType>
- <xs:restriction base="xs:long">
- <xs:minInclusive value="1"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- </xs:complexType>
- </xs:element>
- </xs:sequence>
- <xs:attribute name="identifier">
- <xs:simpleType>
- <xs:restriction base="xs:string">
- <xs:minLength value="1"/>
- <xs:pattern value=".*[^\s].*"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- <xs:attribute name="name">
- <xs:simpleType>
- <xs:restriction base="xs:string">
- <xs:minLength value="1"/>
- <xs:pattern value=".*[^\s].*"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- <xs:attribute name="createdTimestamp">
- <xs:simpleType>
- <xs:restriction base="xs:long">
- <xs:minInclusive value="1"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- <xs:attribute name="modifiedTimestamp">
- <xs:simpleType>
- <xs:restriction base="xs:long">
- <xs:minInclusive value="1"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- <xs:attribute name="bucketIdentifier">
- <xs:simpleType>
- <xs:restriction base="xs:string">
- <xs:minLength value="1"/>
- <xs:pattern value=".*[^\s].*"/>
- </xs:restriction>
- </xs:simpleType>
- </xs:attribute>
- </xs:complexType>
-
- <xs:complexType name="Flows">
- <xs:sequence>
- <xs:element name="flow" type="Flow" minOccurs="0" maxOccurs="unbounded" />
- </xs:sequence>
- </xs:complexType>
-
- <!-- Metadata -->
-
- <xs:element name="metadata">
- <xs:complexType>
- <xs:sequence>
- <xs:element name="buckets" type="Buckets" minOccurs="0" maxOccurs="1"/>
- <xs:element name="flows" type="Flows" minOccurs="0" maxOccurs="1" />
- </xs:sequence>
- </xs:complexType>
- </xs:element>
-
-</xs:schema>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/d478c20e/nifi-registry-provider-impl/src/test/java/org/apache/nifi/registry/flow/TestFileSystemFlowPersistenceProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-impl/src/test/java/org/apache/nifi/registry/flow/TestFileSystemFlowPersistenceProvider.java b/nifi-registry-provider-impl/src/test/java/org/apache/nifi/registry/flow/TestFileSystemFlowPersistenceProvider.java
deleted file mode 100644
index aa0bf58..0000000
--- a/nifi-registry-provider-impl/src/test/java/org/apache/nifi/registry/flow/TestFileSystemFlowPersistenceProvider.java
+++ /dev/null
@@ -1,202 +0,0 @@
-/*
- * 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.flow;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.nifi.registry.provider.ProviderConfigurationContext;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mockito;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.mockito.Mockito.when;
-
-public class TestFileSystemFlowPersistenceProvider {
-
- static final String FLOW_STORAGE_DIR = "target/flow_storage";
-
- static final ProviderConfigurationContext CONFIGURATION_CONTEXT = new ProviderConfigurationContext() {
- @Override
- public Map<String, String> getProperties() {
- final Map<String,String> props = new HashMap<>();
- props.put(FileSystemFlowPersistenceProvider.FLOW_STORAGE_DIR_PROP, FLOW_STORAGE_DIR);
- return props;
- }
- };
-
- private File flowStorageDir;
- private FileSystemFlowPersistenceProvider fileSystemFlowProvider;
-
- @Before
- public void setup() throws IOException {
- flowStorageDir = new File(FLOW_STORAGE_DIR);
- if (flowStorageDir.exists()) {
- org.apache.commons.io.FileUtils.cleanDirectory(flowStorageDir);
- flowStorageDir.delete();
- }
-
- Assert.assertFalse(flowStorageDir.exists());
-
- fileSystemFlowProvider = new FileSystemFlowPersistenceProvider();
- fileSystemFlowProvider.onConfigured(CONFIGURATION_CONTEXT);
- Assert.assertTrue(flowStorageDir.exists());
- }
-
- @Test
- public void testSaveSuccessfully() throws IOException {
- createAndSaveSnapshot(fileSystemFlowProvider,"bucket1", "flow1", 1, "flow1v1");
- verifySnapshot(flowStorageDir, "bucket1", "flow1", 1, "flow1v1");
-
- createAndSaveSnapshot(fileSystemFlowProvider,"bucket1", "flow1", 2, "flow1v2");
- verifySnapshot(flowStorageDir, "bucket1", "flow1", 2, "flow1v2");
-
- createAndSaveSnapshot(fileSystemFlowProvider,"bucket1", "flow2", 1, "flow2v1");
- verifySnapshot(flowStorageDir, "bucket1", "flow2", 1, "flow2v1");
-
- createAndSaveSnapshot(fileSystemFlowProvider,"bucket2", "flow3", 1, "flow3v1");
- verifySnapshot(flowStorageDir, "bucket2", "flow3", 1, "flow3v1");
- }
-
- @Test
- public void testSaveWithExistingVersion() throws IOException {
- final FlowSnapshotContext context = Mockito.mock(FlowSnapshotContext.class);
- when(context.getBucketId()).thenReturn("bucket1");
- when(context.getFlowId()).thenReturn("flow1");
- when(context.getVersion()).thenReturn(1);
-
- final byte[] content = "flow1v1".getBytes(StandardCharsets.UTF_8);
- fileSystemFlowProvider.saveSnapshot(context, content);
-
- // save new content for an existing version
- final byte[] content2 = "XXX".getBytes(StandardCharsets.UTF_8);
- try {
- fileSystemFlowProvider.saveSnapshot(context, content2);
- Assert.fail("Should have thrown exception");
- } catch (Exception e) {
-
- }
-
- // verify the new content wasn't written
- final File flowSnapshotFile = new File(flowStorageDir, "bucket1/flow1/1/1" + FileSystemFlowPersistenceProvider.SNAPSHOT_EXTENSION);
- try (InputStream in = new FileInputStream(flowSnapshotFile)) {
- Assert.assertEquals("flow1v1", IOUtils.toString(in, StandardCharsets.UTF_8));
- }
- }
-
- @Test
- public void testSaveAndGet() throws IOException {
- createAndSaveSnapshot(fileSystemFlowProvider,"bucket1", "flow1", 1, "flow1v1");
- createAndSaveSnapshot(fileSystemFlowProvider,"bucket1", "flow1", 2, "flow1v2");
-
- final byte[] flow1v1 = fileSystemFlowProvider.getSnapshot("bucket1", "flow1", 1);
- Assert.assertEquals("flow1v1", new String(flow1v1, StandardCharsets.UTF_8));
-
- final byte[] flow1v2 = fileSystemFlowProvider.getSnapshot("bucket1", "flow1", 2);
- Assert.assertEquals("flow1v2", new String(flow1v2, StandardCharsets.UTF_8));
- }
-
- @Test
- public void testGetWhenDoesNotExist() {
- final byte[] flow1v1 = fileSystemFlowProvider.getSnapshot("bucket1", "flow1", 1);
- Assert.assertNull(flow1v1);
- }
-
- @Test
- public void testDeleteSnapshots() throws IOException {
- final String bucketId = "bucket1";
- final String flowId = "flow1";
-
- createAndSaveSnapshot(fileSystemFlowProvider, bucketId, flowId, 1, "flow1v1");
- createAndSaveSnapshot(fileSystemFlowProvider, bucketId, flowId, 2, "flow1v2");
-
- Assert.assertNotNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 1));
- Assert.assertNotNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 2));
-
- fileSystemFlowProvider.deleteSnapshots(bucketId, flowId);
-
- Assert.assertNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 1));
- Assert.assertNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 2));
-
- // delete a flow that doesn't exist
- fileSystemFlowProvider.deleteSnapshots(bucketId, "some-other-flow");
-
- // delete a bucket that doesn't exist
- fileSystemFlowProvider.deleteSnapshots("some-other-bucket", flowId);
- }
-
- @Test
- public void testDeleteSnapshot() throws IOException {
- final String bucketId = "bucket1";
- final String flowId = "flow1";
-
- createAndSaveSnapshot(fileSystemFlowProvider, bucketId, flowId, 1, "flow1v1");
- createAndSaveSnapshot(fileSystemFlowProvider, bucketId, flowId, 2, "flow1v2");
-
- Assert.assertNotNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 1));
- Assert.assertNotNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 2));
-
- fileSystemFlowProvider.deleteSnapshot(bucketId, flowId, 1);
-
- Assert.assertNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 1));
- Assert.assertNotNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 2));
-
- fileSystemFlowProvider.deleteSnapshot(bucketId, flowId, 2);
-
- Assert.assertNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 1));
- Assert.assertNull(fileSystemFlowProvider.getSnapshot(bucketId, flowId, 2));
-
- // delete a version that doesn't exist
- fileSystemFlowProvider.deleteSnapshot(bucketId, flowId, 3);
-
- // delete a flow that doesn't exist
- fileSystemFlowProvider.deleteSnapshot(bucketId, "some-other-flow", 1);
-
- // delete a bucket that doesn't exist
- fileSystemFlowProvider.deleteSnapshot("some-other-bucket", flowId, 1);
- }
-
- private void createAndSaveSnapshot(final FlowPersistenceProvider flowPersistenceProvider, final String bucketId, final String flowId, final int version,
- final String contentString) throws IOException {
- final FlowSnapshotContext context = Mockito.mock(FlowSnapshotContext.class);
- when(context.getBucketId()).thenReturn(bucketId);
- when(context.getFlowId()).thenReturn(flowId);
- when(context.getVersion()).thenReturn(version);
-
- final byte[] content = contentString.getBytes(StandardCharsets.UTF_8);
- flowPersistenceProvider.saveSnapshot(context, content);
- }
-
- private void verifySnapshot(final File flowStorageDir, final String bucketId, final String flowId, final int version,
- final String contentString) throws IOException {
- // verify the correct snapshot file was created
- final File flowSnapshotFile = new File(flowStorageDir,
- bucketId + "/" + flowId + "/" + version + "/" + version + FileSystemFlowPersistenceProvider.SNAPSHOT_EXTENSION);
- Assert.assertTrue(flowSnapshotFile.exists());
-
- try (InputStream in = new FileInputStream(flowSnapshotFile)) {
- Assert.assertEquals(contentString, IOUtils.toString(in, StandardCharsets.UTF_8));
- }
- }
-}