You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by pv...@apache.org on 2018/05/15 17:18:31 UTC
[1/3] nifi-registry git commit: NIFIREG-160 - Initial hook provider
Repository: nifi-registry
Updated Branches:
refs/heads/master e8c6497f3 -> 51b6c78b6
NIFIREG-160 - Initial hook provider
Project: http://git-wip-us.apache.org/repos/asf/nifi-registry/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-registry/commit/872a5b5d
Tree: http://git-wip-us.apache.org/repos/asf/nifi-registry/tree/872a5b5d
Diff: http://git-wip-us.apache.org/repos/asf/nifi-registry/diff/872a5b5d
Branch: refs/heads/master
Commit: 872a5b5dc83c0dda0af07e03dd9bae812c182539
Parents: e8c6497
Author: Pierre Villard <pi...@gmail.com>
Authored: Fri Apr 6 16:58:33 2018 +0200
Committer: Pierre Villard <pi...@gmail.com>
Committed: Tue May 15 19:18:02 2018 +0200
----------------------------------------------------------------------
.../registry/extension/ExtensionManager.java | 2 +
.../nifi/registry/provider/ProviderFactory.java | 8 +
.../provider/StandardProviderFactory.java | 53 +++++++
.../flow/StandardFlowSnapshotContext.java | 9 ++
.../provider/hook/ScriptFlowHookProvider.java | 152 +++++++++++++++++++
.../nifi/registry/service/RegistryService.java | 89 +++++++++++
...g.apache.nifi.registry.hook.FlowHookProvider | 15 ++
.../src/main/xsd/providers.xsd | 1 +
.../flow/TestStandardFlowSnapshotContext.java | 4 +-
.../hook/TestScriptFlowHookProvider.java | 45 ++++++
.../registry/service/TestRegistryService.java | 5 +-
.../provider/hook/bad-script-provider.xml | 30 ++++
.../nifi/registry/flow/FlowSnapshotContext.java | 2 +-
.../nifi/registry/hook/FlowHookEvent.java | 31 ++++
.../nifi/registry/hook/FlowHookException.java | 31 ++++
.../nifi/registry/hook/FlowHookProvider.java | 81 ++++++++++
.../src/main/resources/conf/providers.xml | 9 ++
17 files changed, 564 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
index 4c8b5ac..7f2b9e1 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
@@ -22,6 +22,7 @@ import org.apache.nifi.registry.security.authorization.AccessPolicyProvider;
import org.apache.nifi.registry.security.authorization.Authorizer;
import org.apache.nifi.registry.security.authorization.UserGroupProvider;
import org.apache.nifi.registry.flow.FlowPersistenceProvider;
+import org.apache.nifi.registry.hook.FlowHookProvider;
import org.apache.nifi.registry.properties.NiFiRegistryProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,6 +57,7 @@ public class ExtensionManager {
classes.add(AccessPolicyProvider.class);
classes.add(Authorizer.class);
classes.add(IdentityProvider.class);
+ classes.add(FlowHookProvider.class);
EXTENSION_CLASSES = Collections.unmodifiableList(classes);
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
index d5b40a2..bb62eb2 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
@@ -16,7 +16,10 @@
*/
package org.apache.nifi.registry.provider;
+import java.util.List;
+
import org.apache.nifi.registry.flow.FlowPersistenceProvider;
+import org.apache.nifi.registry.hook.FlowHookProvider;
/**
* A factory for obtaining the configured providers.
@@ -35,4 +38,9 @@ public interface ProviderFactory {
*/
FlowPersistenceProvider getFlowPersistenceProvider();
+ /**
+ * @return the configured FlowHookProviders
+ */
+ List<FlowHookProvider> getFlowHookProviders();
+
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
index 5f2da59..34f307c 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
@@ -18,6 +18,7 @@ package org.apache.nifi.registry.provider;
import org.apache.nifi.registry.extension.ExtensionManager;
import org.apache.nifi.registry.flow.FlowPersistenceProvider;
+import org.apache.nifi.registry.hook.FlowHookProvider;
import org.apache.nifi.registry.properties.NiFiRegistryProperties;
import org.apache.nifi.registry.provider.generated.Property;
import org.apache.nifi.registry.provider.generated.Providers;
@@ -39,6 +40,7 @@ import javax.xml.validation.Schema;
import javax.xml.validation.SchemaFactory;
import java.io.File;
import java.lang.reflect.Constructor;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -72,6 +74,7 @@ public class StandardProviderFactory implements ProviderFactory {
private final AtomicReference<Providers> providersHolder = new AtomicReference<>(null);
private FlowPersistenceProvider flowPersistenceProvider;
+ private List<FlowHookProvider> flowHookProviders;
@Autowired
public StandardProviderFactory(final NiFiRegistryProperties properties, final ExtensionManager extensionManager) {
@@ -151,6 +154,56 @@ public class StandardProviderFactory implements ProviderFactory {
return flowPersistenceProvider;
}
+ @Bean
+ @Override
+ public List<FlowHookProvider> getFlowHookProviders() {
+ if (flowHookProviders == null) {
+ flowHookProviders = new ArrayList<FlowHookProvider>();
+
+ if (providersHolder.get() == null) {
+ throw new ProviderFactoryException("ProviderFactory must be initialized before obtaining a Provider");
+ }
+
+ final Providers providers = providersHolder.get();
+ final List<org.apache.nifi.registry.provider.generated.Provider> jaxbFlowHookProvider = providers.getFlowHookProvider();
+
+ if(jaxbFlowHookProvider == null || jaxbFlowHookProvider.isEmpty()) {
+ // no hook provided
+ return flowHookProviders;
+ }
+
+ for (org.apache.nifi.registry.provider.generated.Provider flowHookProvider : jaxbFlowHookProvider) {
+
+ final String flowHookProviderClassName = flowHookProvider.getClazz();
+ FlowHookProvider hook;
+
+ try {
+ final ClassLoader classLoader = extensionManager.getExtensionClassLoader(flowHookProviderClassName);
+ if (classLoader == null) {
+ throw new IllegalStateException("Extension not found in any of the configured class loaders: " + flowHookProviderClassName);
+ }
+
+ final Class<?> rawFlowHookProviderClass = Class.forName(flowHookProviderClassName, true, classLoader);
+ final Class<? extends FlowHookProvider> flowHookProviderClass = rawFlowHookProviderClass.asSubclass(FlowHookProvider.class);
+
+ final Constructor constructor = flowHookProviderClass.getConstructor();
+ hook = (FlowHookProvider) constructor.newInstance();
+
+ LOGGER.info("Instantiated FlowHookProvider with class name {}", new Object[] {flowHookProviderClassName});
+ } catch (Exception e) {
+ throw new ProviderFactoryException("Error creating FlowHookProvider with class name: " + flowHookProviderClassName, e);
+ }
+
+ final ProviderConfigurationContext configurationContext = createConfigurationContext(flowHookProvider.getProperty());
+ hook.onConfigured(configurationContext);
+ flowHookProviders.add(hook);
+ LOGGER.info("Configured FlowHookProvider with class name {}", new Object[] {flowHookProviderClassName});
+ }
+ }
+
+ return flowHookProviders;
+ }
+
private ProviderConfigurationContext createConfigurationContext(final List<Property> configProperties) {
final Map<String,String> properties = new HashMap<>();
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
index 513f37f..1f6f840 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
@@ -33,6 +33,7 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
private final String flowName;
private final int version;
private final String comments;
+ private final String author;
private final long snapshotTimestamp;
private final String author;
@@ -43,6 +44,7 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
this.flowName = builder.flowName;
this.version = builder.version;
this.comments = builder.comments;
+ this.author = builder.author;
this.snapshotTimestamp = builder.snapshotTimestamp;
this.author = builder.author;
@@ -105,6 +107,7 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
private String flowName;
private int version;
private String comments;
+ private String author;
private long snapshotTimestamp;
private String author;
@@ -119,6 +122,7 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
flowName(versionedFlow.getName());
version(snapshotMetadata.getVersion());
comments(snapshotMetadata.getComments());
+ author(snapshotMetadata.getAuthor());
snapshotTimestamp(snapshotMetadata.getTimestamp());
author(snapshotMetadata.getAuthor());
}
@@ -153,6 +157,11 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
return this;
}
+ public Builder author(final String author) {
+ this.author = author;
+ return this;
+ }
+
public Builder snapshotTimestamp(final long snapshotTimestamp) {
this.snapshotTimestamp = snapshotTimestamp;
return this;
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java
new file mode 100644
index 0000000..d57df16
--- /dev/null
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java
@@ -0,0 +1,152 @@
+/*
+ * 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.provider.hook;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.registry.flow.FlowSnapshotContext;
+import org.apache.nifi.registry.hook.FlowHookEvent;
+import org.apache.nifi.registry.hook.FlowHookException;
+import org.apache.nifi.registry.hook.FlowHookProvider;
+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;
+
+/**
+ * A FlowHookProvider that is used to execute a script before a flow snapshot version is committed.
+ */
+public class ScriptFlowHookProvider implements FlowHookProvider {
+
+ static final Logger LOGGER = LoggerFactory.getLogger(ScriptFlowHookProvider.class);
+ static final String SCRIPT_PATH_PROP = "Script Path";
+ static final String SCRIPT_WORKDIR_PROP = "Working Directory";
+ private File scriptFile;
+ private File workDirFile;
+
+ @Override
+ public void postCreateBucket(String bucketId) throws FlowHookException {
+ this.executeScript(FlowHookEvent.CREATE_BUCKET, bucketId, null, null, null, null);
+ }
+
+ @Override
+ public void postCreateFlow(String bucketId, String flowId) throws FlowHookException {
+ this.executeScript(FlowHookEvent.CREATE_FLOW, bucketId, flowId, null, null, null);
+ }
+
+ @Override
+ public void postDeleteBucket(String bucketId) throws FlowHookException {
+ this.executeScript(FlowHookEvent.DELETE_BUCKET, bucketId, null, null, null, null);
+ }
+
+ @Override
+ public void postDeleteFlow(String bucketId, String flowId) throws FlowHookException {
+ this.executeScript(FlowHookEvent.DELETE_FLOW, bucketId, flowId, null, null, null);
+ }
+
+ @Override
+ public void postDeleteFlowVersion(String bucketId, String flowId, int version) throws FlowHookException {
+ this.executeScript(FlowHookEvent.DELETE_FLOW, bucketId, flowId, Integer.toString(version), null, null);
+ }
+
+ @Override
+ public void postUpdateBucket(String bucketId) throws FlowHookException {
+ this.executeScript(FlowHookEvent.UPDATE_BUCKET, bucketId, null, null, null, null);
+ }
+
+ @Override
+ public void postUpdateFlow(String bucketId, String flowId) throws FlowHookException {
+ this.executeScript(FlowHookEvent.UPDATE_FLOW, bucketId, flowId, null, null, null);
+ }
+
+ @Override
+ public void postCreateFlowVersion(final FlowSnapshotContext flowSnapshotContext) throws FlowHookException {
+ this.executeScript(FlowHookEvent.CREATE_VERSION, flowSnapshotContext.getBucketId(), flowSnapshotContext.getFlowId(),
+ Integer.toString(flowSnapshotContext.getVersion()), flowSnapshotContext.getComments(), flowSnapshotContext.getAuthor());
+ }
+
+ private void executeScript(final FlowHookEvent eventType, String bucketId, String flowId, String version, String comment, String author) {
+ List<String> command = new ArrayList<String>();
+ command.add(scriptFile.getAbsolutePath());
+ command.add(eventType.name());
+ command.add(bucketId);
+
+ if(flowId != null) {
+ command.add(flowId);
+ }
+
+ if(version != null) {
+ command.add(version);
+ }
+
+ if(comment != null) {
+ command.add(comment);
+ }
+
+ if(author != null) {
+ command.add(author);
+ }
+
+ final String commandString = StringUtils.join(command, " ");
+ final ProcessBuilder builder = new ProcessBuilder(command);
+ builder.directory(workDirFile);
+ LOGGER.debug("Execution of " + commandString);
+
+ try {
+ builder.start();
+ } catch (IOException e) {
+ LOGGER.error("Execution of {0} failed with: {1}", new Object[] { commandString, e.getLocalizedMessage() }, e);
+ }
+ }
+
+ @Override
+ public void onConfigured(ProviderConfigurationContext configurationContext) throws ProviderCreationException {
+ final Map<String,String> props = configurationContext.getProperties();
+ if (!props.containsKey(SCRIPT_PATH_PROP)) {
+ throw new ProviderCreationException("The property " + SCRIPT_PATH_PROP + " must be provided");
+ }
+
+ final String scripPath = props.get(SCRIPT_PATH_PROP);
+ if (StringUtils.isBlank(scripPath)) {
+ throw new ProviderCreationException("The property " + SCRIPT_PATH_PROP + " cannot be null or blank");
+ }
+
+ if(props.containsKey(SCRIPT_WORKDIR_PROP) && !StringUtils.isBlank(props.get(SCRIPT_WORKDIR_PROP))) {
+ final String workdir = props.get(SCRIPT_WORKDIR_PROP);
+ try {
+ workDirFile = new File(workdir);
+ FileUtils.ensureDirectoryExistAndCanRead(workDirFile);
+ } catch (IOException e) {
+ throw new ProviderCreationException("The working directory " + workdir + " cannot be read.");
+ }
+ }
+
+ scriptFile = new File(scripPath);
+ if(scriptFile.isFile() && scriptFile.canExecute()) {
+ LOGGER.info("Configured ScriptFlowHookProvider with script {}", new Object[] {scriptFile.getAbsolutePath()});
+ } else {
+ throw new ProviderCreationException("The script file " + scriptFile.getAbsolutePath() + " cannot be executed.");
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
index 23f1d14..95c62c3 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
@@ -42,6 +42,7 @@ import org.apache.nifi.registry.flow.diff.FlowComparison;
import org.apache.nifi.registry.flow.diff.FlowDifference;
import org.apache.nifi.registry.flow.diff.StandardComparableDataFlow;
import org.apache.nifi.registry.flow.diff.StandardFlowComparator;
+import org.apache.nifi.registry.hook.FlowHookProvider;
import org.apache.nifi.registry.provider.flow.StandardFlowSnapshotContext;
import org.apache.nifi.registry.serialization.Serializer;
import org.slf4j.Logger;
@@ -84,6 +85,7 @@ public class RegistryService {
private final MetadataService metadataService;
private final FlowPersistenceProvider flowPersistenceProvider;
+ private final List<FlowHookProvider> flowHookProviders;
private final Serializer<VersionedProcessGroup> processGroupSerializer;
private final Validator validator;
@@ -94,10 +96,12 @@ public class RegistryService {
@Autowired
public RegistryService(final MetadataService metadataService,
final FlowPersistenceProvider flowPersistenceProvider,
+ final List<FlowHookProvider> flowHookProviders,
final Serializer<VersionedProcessGroup> processGroupSerializer,
final Validator validator) {
this.metadataService = metadataService;
this.flowPersistenceProvider = flowPersistenceProvider;
+ this.flowHookProviders = flowHookProviders;
this.processGroupSerializer = processGroupSerializer;
this.validator = validator;
Validate.notNull(this.metadataService);
@@ -134,6 +138,17 @@ public class RegistryService {
}
final BucketEntity createdBucket = metadataService.createBucket(DataModelMapper.map(bucket));
+
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postCreateBucket(createdBucket.getId());
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
return DataModelMapper.map(createdBucket);
} finally {
writeLock.unlock();
@@ -225,6 +240,17 @@ public class RegistryService {
// perform the actual update
final BucketEntity updatedBucket = metadataService.updateBucket(existingBucketById);
+
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postUpdateBucket(existingBucketById.getId());
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
return DataModelMapper.map(updatedBucket);
} finally {
writeLock.unlock();
@@ -253,6 +279,16 @@ public class RegistryService {
// now delete the bucket from the metadata provider, which deletes all flows referencing it
metadataService.deleteBucket(existingBucket);
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postDeleteBucket(existingBucket.getId());
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
return DataModelMapper.map(existingBucket);
} finally {
writeLock.unlock();
@@ -356,6 +392,17 @@ public class RegistryService {
// persist the flow and return the created entity
final FlowEntity createdFlow = metadataService.createFlow(flowEntity);
+
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postCreateFlow(existingBucket.getId(), createdFlow.getId());
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
return DataModelMapper.map(existingBucket, createdFlow);
} finally {
writeLock.unlock();
@@ -497,6 +544,17 @@ public class RegistryService {
// perform the actual update
final FlowEntity updatedFlow = metadataService.updateFlow(existingFlow);
+
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postUpdateFlow(existingFlow.getBucketId(), existingFlow.getId());
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
return DataModelMapper.map(existingBucket, updatedFlow);
} finally {
writeLock.unlock();
@@ -537,6 +595,16 @@ public class RegistryService {
// now delete the flow from the metadata provider
metadataService.deleteFlow(existingFlow);
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postDeleteFlow(existingFlow.getBucketId(), existingFlow.getId());
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
return DataModelMapper.map(existingBucket, existingFlow);
} finally {
writeLock.unlock();
@@ -630,6 +698,16 @@ public class RegistryService {
}
final VersionedFlow updatedVersionedFlow = DataModelMapper.map(existingBucket, updatedFlow);
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postCreateFlowVersion(context);
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
flowSnapshot.setBucket(bucket);
flowSnapshot.setFlow(updatedVersionedFlow);
return flowSnapshot;
@@ -879,6 +957,17 @@ public class RegistryService {
// delete the snapshot itself
metadataService.deleteFlowSnapshot(snapshotEntity);
+
+ // call the post-event hook
+ for(FlowHookProvider flowHookProvider : flowHookProviders) {
+ try {
+ flowHookProvider.postDeleteFlowVersion(bucketIdentifier, flowIdentifier, version);
+ } catch (Exception e) {
+ // we don't want to throw anything here, hook are provided on best effort
+ LOGGER.error("Error while calling post-event hook", e);
+ }
+ }
+
return DataModelMapper.map(existingBucket, snapshotEntity);
} finally {
writeLock.unlock();
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider b/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider
new file mode 100644
index 0000000..150103a
--- /dev/null
+++ b/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider
@@ -0,0 +1,15 @@
+# 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.
+org.apache.nifi.registry.provider.hook.ScriptFlowHookProvider
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/main/xsd/providers.xsd
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/xsd/providers.xsd b/nifi-registry-framework/src/main/xsd/providers.xsd
index 1202f9e..16086af 100644
--- a/nifi-registry-framework/src/main/xsd/providers.xsd
+++ b/nifi-registry-framework/src/main/xsd/providers.xsd
@@ -43,6 +43,7 @@
<xs:complexType>
<xs:sequence>
<xs:element name="flowPersistenceProvider" type="Provider" minOccurs="1" maxOccurs="1" />
+ <xs:element name="flowHookProvider" type="Provider" minOccurs="0" maxOccurs="unbounded" />
</xs:sequence>
</xs:complexType>
</xs:element>
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/TestStandardFlowSnapshotContext.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/TestStandardFlowSnapshotContext.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/TestStandardFlowSnapshotContext.java
index 3e0a106..bff2ef9 100644
--- a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/TestStandardFlowSnapshotContext.java
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/TestStandardFlowSnapshotContext.java
@@ -17,7 +17,6 @@
package org.apache.nifi.registry.provider.flow;
import org.apache.nifi.registry.flow.FlowSnapshotContext;
-import org.apache.nifi.registry.provider.flow.StandardFlowSnapshotContext;
import org.junit.Assert;
import org.junit.Test;
@@ -31,6 +30,7 @@ public class TestStandardFlowSnapshotContext {
final String flowName = "Some Flow";
final int version = 2;
final String comments = "Some Comments";
+ final String author = "anonymous";
final long timestamp = System.currentTimeMillis();
final FlowSnapshotContext context = new StandardFlowSnapshotContext.Builder()
@@ -40,6 +40,7 @@ public class TestStandardFlowSnapshotContext {
.flowName(flowName)
.version(version)
.comments(comments)
+ .author(author)
.snapshotTimestamp(timestamp)
.build();
@@ -49,6 +50,7 @@ public class TestStandardFlowSnapshotContext {
Assert.assertEquals(flowName, context.getFlowName());
Assert.assertEquals(version, context.getVersion());
Assert.assertEquals(comments, context.getComments());
+ Assert.assertEquals(author, context.getAuthor());
Assert.assertEquals(timestamp, context.getSnapshotTimestamp());
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java
new file mode 100644
index 0000000..5bc888c
--- /dev/null
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java
@@ -0,0 +1,45 @@
+/*
+ * 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.provider.hook;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import org.apache.nifi.registry.extension.ExtensionManager;
+import org.apache.nifi.registry.properties.NiFiRegistryProperties;
+import org.apache.nifi.registry.provider.ProviderCreationException;
+import org.apache.nifi.registry.provider.ProviderFactory;
+import org.apache.nifi.registry.provider.StandardProviderFactory;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class TestScriptFlowHookProvider {
+
+ @Test(expected = ProviderCreationException.class)
+ public void testBadScriptProvider() {
+ final NiFiRegistryProperties props = new NiFiRegistryProperties();
+ props.setProperty(NiFiRegistryProperties.PROVIDERS_CONFIGURATION_FILE, "src/test/resources/provider/hook/bad-script-provider.xml");
+
+ final ExtensionManager extensionManager = Mockito.mock(ExtensionManager.class);
+ when(extensionManager.getExtensionClassLoader(any(String.class))).thenReturn(this.getClass().getClassLoader());
+
+ final ProviderFactory providerFactory = new StandardProviderFactory(props, extensionManager);
+ providerFactory.initialize();
+ providerFactory.getFlowHookProviders();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
index 95e2d1a..1f21248 100644
--- a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
@@ -30,6 +30,7 @@ import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
import org.apache.nifi.registry.flow.VersionedProcessGroup;
import org.apache.nifi.registry.flow.VersionedProcessor;
+import org.apache.nifi.registry.hook.FlowHookProvider;
import org.apache.nifi.registry.serialization.Serializer;
import org.apache.nifi.registry.serialization.VersionedProcessGroupSerializer;
import org.junit.Assert;
@@ -72,6 +73,7 @@ public class TestRegistryService {
private MetadataService metadataService;
private FlowPersistenceProvider flowPersistenceProvider;
+ private FlowHookProvider flowHookProvider;
private Serializer<VersionedProcessGroup> snapshotSerializer;
private Validator validator;
@@ -81,12 +83,13 @@ public class TestRegistryService {
public void setup() {
metadataService = mock(MetadataService.class);
flowPersistenceProvider = mock(FlowPersistenceProvider.class);
+ // flowHookProvider = mock(FlowHookProvider.class);
snapshotSerializer = mock(VersionedProcessGroupSerializer.class);
final ValidatorFactory validatorFactory = Validation.buildDefaultValidatorFactory();
validator = validatorFactory.getValidator();
- registryService = new RegistryService(metadataService, flowPersistenceProvider, snapshotSerializer, validator);
+ registryService = new RegistryService(metadataService, flowPersistenceProvider, new ArrayList<FlowHookProvider>(), snapshotSerializer, validator);
}
// ---------------------- Test Bucket methods ---------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml b/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml
new file mode 100644
index 0000000..ca6cc6c
--- /dev/null
+++ b/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml
@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<!--
+ 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.
+-->
+<providers>
+
+ <flowPersistenceProvider>
+ <class>org.apache.nifi.registry.provider.MockFlowPersistenceProvider</class>
+ <property name="Flow Property 1">flow foo</property>
+ <property name="Flow Property 2">flow bar</property>
+ </flowPersistenceProvider>
+
+ <flowHookProvider>
+ <class>org.apache.nifi.registry.provider.hook.ScriptFlowHookProvider</class>
+ <property name="Script Path"></property>
+ <property name="Working Directory"></property>
+ </flowHookProvider>
+
+</providers>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/flow/FlowSnapshotContext.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/flow/FlowSnapshotContext.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/flow/FlowSnapshotContext.java
index c4bdd46..9c2a818 100644
--- a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/flow/FlowSnapshotContext.java
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/flow/FlowSnapshotContext.java
@@ -57,7 +57,7 @@ public interface FlowSnapshotContext {
long getSnapshotTimestamp();
/**
- * @return the name of the user who created the snapshot
+ * @return the author of the snapshot
*/
String getAuthor();
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java
new file mode 100644
index 0000000..c0aebc7
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java
@@ -0,0 +1,31 @@
+/*
+ * 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.hook;
+
+public enum FlowHookEvent {
+ CREATE_BUCKET,
+ CREATE_FLOW,
+ CREATE_VERSION,
+
+ UPDATE_BUCKET,
+ UPDATE_FLOW,
+
+ DELETE_BUCKET,
+ DELETE_FLOW,
+ DELETE_VERSION;
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java
new file mode 100644
index 0000000..85033ca
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.hook;
+
+/**
+ * An Exception for errors encountered when a FlowHookProvider executes an action before/after a commit.
+ */
+public class FlowHookException extends RuntimeException {
+
+ public FlowHookException(String message) {
+ super(message);
+ }
+
+ public FlowHookException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java
new file mode 100644
index 0000000..599cbed
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java
@@ -0,0 +1,81 @@
+/*
+ * 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.hook;
+
+import org.apache.nifi.registry.flow.FlowSnapshotContext;
+import org.apache.nifi.registry.provider.Provider;
+
+/**
+ * A service that defines post event action hook
+ *
+ * NOTE: Although this interface is intended to be an extension point, it is not yet considered stable and thus may
+ * change across releases until the registry matures.
+ */
+public interface FlowHookProvider extends Provider {
+
+ /**
+ * @param bucketId
+ * @throws FlowHookException
+ */
+ default void postCreateBucket(String bucketId) throws FlowHookException { }
+
+ /**
+ * @param bucketId
+ * @param flowId
+ * @throws FlowHookException
+ */
+ default void postCreateFlow(String bucketId, String flowId) throws FlowHookException { }
+
+ /**
+ * @param flowSnapshotContext
+ * @throws FlowHookException
+ */
+ default void postCreateFlowVersion(FlowSnapshotContext flowSnapshotContext) throws FlowHookException { }
+
+ /**
+ * @param bucketId
+ * @throws FlowHookException
+ */
+ default void postDeleteBucket(String bucketId) throws FlowHookException { }
+
+ /**
+ * @param bucketId
+ * @param flowId
+ * @throws FlowHookException
+ */
+ default void postDeleteFlow(String bucketId, String flowId) throws FlowHookException { }
+
+ /**
+ * @param flowSnapshotContext
+ * @throws FlowHookException
+ */
+ default void postDeleteFlowVersion(String bucketIdentifier, String flowIdentifier, int version) throws FlowHookException { }
+
+ /**
+ * @param bucketId
+ * @throws FlowHookException
+ */
+ default void postUpdateBucket(String bucketId) throws FlowHookException { }
+
+ /**
+ * @param bucketId
+ * @param flowId
+ * @throws FlowHookException
+ */
+ default void postUpdateFlow(String bucketId, String flowId) throws FlowHookException { }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/872a5b5d/nifi-registry-resources/src/main/resources/conf/providers.xml
----------------------------------------------------------------------
diff --git a/nifi-registry-resources/src/main/resources/conf/providers.xml b/nifi-registry-resources/src/main/resources/conf/providers.xml
index 720bee2..3d7d392 100644
--- a/nifi-registry-resources/src/main/resources/conf/providers.xml
+++ b/nifi-registry-resources/src/main/resources/conf/providers.xml
@@ -29,4 +29,13 @@
<property name="Remote Access Password"></property>
</flowPersistenceProvider>
-->
+
+ <!--
+ <flowHookProvider>
+ <class>org.apache.nifi.registry.provider.hook.ScriptFlowHookProvider</class>
+ <property name="Script Path"></property>
+ <property name="Working Directory"></property>
+ </flowHookProvider>
+ -->
+
</providers>
\ No newline at end of file
[2/3] nifi-registry git commit: NIFIREG-160 - Making event hooks
asynchronous
Posted by pv...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookException.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookException.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookException.java
new file mode 100644
index 0000000..2d91735
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.hook;
+
+/**
+ * An Exception for errors encountered when a EventHookProvider executes an action before/after a commit.
+ */
+public class EventHookException extends RuntimeException {
+
+ public EventHookException(String message) {
+ super(message);
+ }
+
+ public EventHookException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookProvider.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookProvider.java
new file mode 100644
index 0000000..e134e8f
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventHookProvider.java
@@ -0,0 +1,39 @@
+/*
+ * 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.hook;
+
+import org.apache.nifi.registry.provider.Provider;
+
+/**
+ * An extension point that will be passed events produced by actions take in the registry.
+ *
+ * The list of event types can be found in {@link org.apache.nifi.registry.hook.EventType}.
+ *
+ * NOTE: Although this interface is intended to be an extension point, it is not yet considered stable and thus may
+ * change across releases until the registry matures.
+ */
+public interface EventHookProvider extends Provider {
+
+ /**
+ * Handles the given event.
+ *
+ * @param event the event to handle
+ * @throws EventHookException if an error occurs handling the event
+ */
+ void handle(Event event) throws EventHookException;
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventType.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventType.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventType.java
new file mode 100644
index 0000000..ce6930a
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventType.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.registry.hook;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Enumeration of possible EventTypes with the expected fields for each event.
+ *
+ * Producers of events must produce events with the fields in the same order specified here.
+ */
+public enum EventType {
+
+ CREATE_BUCKET(
+ EventFieldName.BUCKET_ID,
+ EventFieldName.USER),
+ CREATE_FLOW(
+ EventFieldName.BUCKET_ID,
+ EventFieldName.FLOW_ID,
+ EventFieldName.USER),
+ CREATE_FLOW_VERSION(
+ EventFieldName.BUCKET_ID,
+ EventFieldName.FLOW_ID,
+ EventFieldName.VERSION,
+ EventFieldName.USER,
+ EventFieldName.COMMENT),
+
+ UPDATE_BUCKET(
+ EventFieldName.BUCKET_ID,
+ EventFieldName.USER),
+ UPDATE_FLOW(
+ EventFieldName.BUCKET_ID,
+ EventFieldName.FLOW_ID,
+ EventFieldName.USER),
+
+ DELETE_BUCKET(
+ EventFieldName.BUCKET_ID,
+ EventFieldName.USER),
+ DELETE_FLOW(
+ EventFieldName.BUCKET_ID,
+ EventFieldName.FLOW_ID,
+ EventFieldName.USER);
+
+
+ private List<EventFieldName> fieldNames;
+
+ EventType(EventFieldName... fieldNames) {
+ this.fieldNames = Collections.unmodifiableList(Arrays.asList(fieldNames));
+ }
+
+ public List<EventFieldName> getFieldNames() {
+ return this.fieldNames;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java
deleted file mode 100644
index c0aebc7..0000000
--- a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookEvent.java
+++ /dev/null
@@ -1,31 +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.hook;
-
-public enum FlowHookEvent {
- CREATE_BUCKET,
- CREATE_FLOW,
- CREATE_VERSION,
-
- UPDATE_BUCKET,
- UPDATE_FLOW,
-
- DELETE_BUCKET,
- DELETE_FLOW,
- DELETE_VERSION;
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java
deleted file mode 100644
index 85033ca..0000000
--- a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookException.java
+++ /dev/null
@@ -1,31 +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.hook;
-
-/**
- * An Exception for errors encountered when a FlowHookProvider executes an action before/after a commit.
- */
-public class FlowHookException extends RuntimeException {
-
- public FlowHookException(String message) {
- super(message);
- }
-
- public FlowHookException(String message, Throwable cause) {
- super(message, cause);
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java
deleted file mode 100644
index 599cbed..0000000
--- a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/FlowHookProvider.java
+++ /dev/null
@@ -1,81 +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.hook;
-
-import org.apache.nifi.registry.flow.FlowSnapshotContext;
-import org.apache.nifi.registry.provider.Provider;
-
-/**
- * A service that defines post event action hook
- *
- * NOTE: Although this interface is intended to be an extension point, it is not yet considered stable and thus may
- * change across releases until the registry matures.
- */
-public interface FlowHookProvider extends Provider {
-
- /**
- * @param bucketId
- * @throws FlowHookException
- */
- default void postCreateBucket(String bucketId) throws FlowHookException { }
-
- /**
- * @param bucketId
- * @param flowId
- * @throws FlowHookException
- */
- default void postCreateFlow(String bucketId, String flowId) throws FlowHookException { }
-
- /**
- * @param flowSnapshotContext
- * @throws FlowHookException
- */
- default void postCreateFlowVersion(FlowSnapshotContext flowSnapshotContext) throws FlowHookException { }
-
- /**
- * @param bucketId
- * @throws FlowHookException
- */
- default void postDeleteBucket(String bucketId) throws FlowHookException { }
-
- /**
- * @param bucketId
- * @param flowId
- * @throws FlowHookException
- */
- default void postDeleteFlow(String bucketId, String flowId) throws FlowHookException { }
-
- /**
- * @param flowSnapshotContext
- * @throws FlowHookException
- */
- default void postDeleteFlowVersion(String bucketIdentifier, String flowIdentifier, int version) throws FlowHookException { }
-
- /**
- * @param bucketId
- * @throws FlowHookException
- */
- default void postUpdateBucket(String bucketId) throws FlowHookException { }
-
- /**
- * @param bucketId
- * @param flowId
- * @throws FlowHookException
- */
- default void postUpdateFlow(String bucketId, String flowId) throws FlowHookException { }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-resources/src/main/resources/conf/logback.xml
----------------------------------------------------------------------
diff --git a/nifi-registry-resources/src/main/resources/conf/logback.xml b/nifi-registry-resources/src/main/resources/conf/logback.xml
index 72e863c..7d65bda 100644
--- a/nifi-registry-resources/src/main/resources/conf/logback.xml
+++ b/nifi-registry-resources/src/main/resources/conf/logback.xml
@@ -58,6 +58,24 @@
</encoder>
</appender>
+ <appender name="EVENTS_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <file>${org.apache.nifi.registry.bootstrap.config.log.dir}/nifi-registry-event.log</file>
+ <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+ <!--
+ For daily rollover, use 'user_%d.log'.
+ For hourly rollover, use 'user_%d{yyyy-MM-dd_HH}.log'.
+ To GZIP rolled files, replace '.log' with '.log.gz'.
+ To ZIP rolled files, replace '.log' with '.log.zip'.
+ -->
+ <fileNamePattern>${org.apache.nifi.registry.bootstrap.config.log.dir}/nifi-registry-event_%d.log</fileNamePattern>
+ <!-- keep 5 log files worth of history -->
+ <maxHistory>5</maxHistory>
+ </rollingPolicy>
+ <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+ <pattern>%date ## %msg%n</pattern>
+ </encoder>
+ </appender>
+
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
@@ -94,6 +112,11 @@
<appender-ref ref="BOOTSTRAP_FILE" />
</logger>
+ <!-- This will log all events to a separate file when the LoggingEventHookProvider is enabled in providers.xml -->
+ <logger name="org.apache.nifi.registry.provider.hook.LoggingEventHookProvider" level="INFO" additivity="false">
+ <appender-ref ref="EVENTS_FILE" />
+ </logger>
+
<root level="INFO">
<appender-ref ref="APP_FILE"/>
</root>
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-resources/src/main/resources/conf/providers.xml
----------------------------------------------------------------------
diff --git a/nifi-registry-resources/src/main/resources/conf/providers.xml b/nifi-registry-resources/src/main/resources/conf/providers.xml
index 3d7d392..e59e277 100644
--- a/nifi-registry-resources/src/main/resources/conf/providers.xml
+++ b/nifi-registry-resources/src/main/resources/conf/providers.xml
@@ -31,11 +31,18 @@
-->
<!--
- <flowHookProvider>
- <class>org.apache.nifi.registry.provider.hook.ScriptFlowHookProvider</class>
+ <eventHookProvider>
+ <class>org.apache.nifi.registry.provider.hook.ScriptEventHookProvider</class>
<property name="Script Path"></property>
<property name="Working Directory"></property>
- </flowHookProvider>
+ </eventHookProvider>
+ -->
+
+ <!-- This will log all events to a separate file specified by the EVENT_APPENDER in logback.xml -->
+ <!--
+ <eventHookProvider>
+ <class>org.apache.nifi.registry.provider.hook.LoggingEventHookProvider</class>
+ </eventHookProvider>
-->
</providers>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessPolicyResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessPolicyResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessPolicyResource.java
index ac84b2f..713b369 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessPolicyResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessPolicyResource.java
@@ -27,6 +27,7 @@ import io.swagger.annotations.ExtensionProperty;
import org.apache.nifi.registry.authorization.AccessPolicy;
import org.apache.nifi.registry.authorization.AccessPolicySummary;
import org.apache.nifi.registry.authorization.Resource;
+import org.apache.nifi.registry.event.EventService;
import org.apache.nifi.registry.exception.ResourceNotFoundException;
import org.apache.nifi.registry.security.authorization.Authorizer;
import org.apache.nifi.registry.security.authorization.AuthorizerCapabilityDetection;
@@ -73,8 +74,9 @@ public class AccessPolicyResource extends AuthorizableApplicationResource {
@Autowired
public AccessPolicyResource(
Authorizer authorizer,
- AuthorizationService authorizationService) {
- super(authorizationService);
+ AuthorizationService authorizationService,
+ EventService eventService) {
+ super(authorizationService, eventService);
this.authorizer = authorizer;
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessResource.java
index 0804295..d310d0c 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AccessResource.java
@@ -24,6 +24,7 @@ import io.swagger.annotations.ApiResponses;
import io.swagger.annotations.Authorization;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.registry.authorization.CurrentUser;
+import org.apache.nifi.registry.event.EventService;
import org.apache.nifi.registry.exception.AdministrationException;
import org.apache.nifi.registry.properties.NiFiRegistryProperties;
import org.apache.nifi.registry.security.authentication.AuthenticationRequest;
@@ -86,7 +87,9 @@ public class AccessResource extends ApplicationResource {
JwtService jwtService,
X509IdentityProvider x509IdentityProvider,
@Nullable KerberosSpnegoIdentityProvider kerberosSpnegoIdentityProvider,
- @Nullable IdentityProvider identityProvider) {
+ @Nullable IdentityProvider identityProvider,
+ EventService eventService) {
+ super(eventService);
this.properties = properties;
this.jwtService = jwtService;
this.x509IdentityProvider = x509IdentityProvider;
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ApplicationResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ApplicationResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ApplicationResource.java
index ed5c281..34369d0 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ApplicationResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ApplicationResource.java
@@ -17,6 +17,9 @@
package org.apache.nifi.registry.web.api;
import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.Validate;
+import org.apache.nifi.registry.event.EventService;
+import org.apache.nifi.registry.hook.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +55,17 @@ public class ApplicationResource {
@Context
private UriInfo uriInfo;
+ private final EventService eventService;
+
+ public ApplicationResource(final EventService eventService) {
+ this.eventService = eventService;
+ Validate.notNull(this.eventService);
+ }
+
+ protected void publish(final Event event) {
+ eventService.publish(event);
+ }
+
protected String generateResourceUri(final String... path) {
final UriBuilder uriBuilder = uriInfo.getBaseUriBuilder();
uriBuilder.segment(path);
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AuthorizableApplicationResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AuthorizableApplicationResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AuthorizableApplicationResource.java
index 8f983cf..83240c7 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AuthorizableApplicationResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/AuthorizableApplicationResource.java
@@ -18,6 +18,7 @@ package org.apache.nifi.registry.web.api;
import org.apache.nifi.registry.authorization.Resource;
import org.apache.nifi.registry.bucket.BucketItem;
+import org.apache.nifi.registry.event.EventService;
import org.apache.nifi.registry.security.authorization.AuthorizableLookup;
import org.apache.nifi.registry.security.authorization.RequestAction;
import org.apache.nifi.registry.security.authorization.resource.Authorizable;
@@ -38,7 +39,9 @@ public class AuthorizableApplicationResource extends ApplicationResource {
protected final AuthorizableLookup authorizableLookup;
protected AuthorizableApplicationResource(
- AuthorizationService authorizationService) {
+ AuthorizationService authorizationService,
+ EventService eventService) {
+ super(eventService);
this.authorizationService = authorizationService;
this.authorizableLookup = authorizationService.getAuthorizableLookup();
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java
index 64f1830..942a3d4 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java
@@ -27,6 +27,8 @@ import io.swagger.annotations.ExtensionProperty;
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.flow.VersionedFlow;
import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
@@ -76,8 +78,9 @@ public class BucketFlowResource extends AuthorizableApplicationResource {
final RegistryService registryService,
final LinkService linkService,
final PermissionsService permissionsService,
- final AuthorizationService authorizationService) {
- super(authorizationService);
+ final AuthorizationService authorizationService,
+ final EventService eventService) {
+ super(authorizationService, eventService);
this.registryService = registryService;
this.linkService = linkService;
this.permissionsService =permissionsService;
@@ -111,7 +114,10 @@ public class BucketFlowResource extends AuthorizableApplicationResource {
authorizeBucketAccess(RequestAction.WRITE, bucketId);
verifyPathParamsMatchBody(bucketId, flow);
+
final VersionedFlow createdFlow = registryService.createFlow(bucketId, flow);
+ publish(EventFactory.flowCreated(createdFlow));
+
permissionsService.populateItemPermissions(createdFlow);
linkService.populateFlowLinks(createdFlow);
return Response.status(Response.Status.OK).entity(createdFlow).build();
@@ -222,6 +228,7 @@ public class BucketFlowResource extends AuthorizableApplicationResource {
setBucketItemMetadataIfMissing(bucketId, flowId, flow);
final VersionedFlow updatedFlow = registryService.updateFlow(flow);
+ publish(EventFactory.flowUpdated(updatedFlow));
permissionsService.populateItemPermissions(updatedFlow);
linkService.populateFlowLinks(updatedFlow);
@@ -256,6 +263,7 @@ public class BucketFlowResource extends AuthorizableApplicationResource {
authorizeBucketAccess(RequestAction.DELETE, bucketId);
final VersionedFlow deletedFlow = registryService.deleteFlow(bucketId, flowId);
+ publish(EventFactory.flowDeleted(deletedFlow));
return Response.status(Response.Status.OK).entity(deletedFlow).build();
}
@@ -300,6 +308,8 @@ public class BucketFlowResource extends AuthorizableApplicationResource {
snapshot.getSnapshotMetadata().setAuthor(userIdentity);
final VersionedFlowSnapshot createdSnapshot = registryService.createFlowSnapshot(snapshot);
+ publish(EventFactory.flowVersionCreated(createdSnapshot));
+
if (createdSnapshot.getSnapshotMetadata() != null) {
linkService.populateSnapshotLinks(createdSnapshot.getSnapshotMetadata());
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketResource.java
index 5f75e80..e905973 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketResource.java
@@ -27,6 +27,8 @@ import io.swagger.annotations.ExtensionProperty;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.registry.bucket.Bucket;
import org.apache.nifi.registry.bucket.BucketItem;
+import org.apache.nifi.registry.event.EventFactory;
+import org.apache.nifi.registry.event.EventService;
import org.apache.nifi.registry.field.Fields;
import org.apache.nifi.registry.security.authorization.RequestAction;
import org.apache.nifi.registry.security.authorization.exception.AccessDeniedException;
@@ -83,8 +85,9 @@ public class BucketResource extends AuthorizableApplicationResource {
final RegistryService registryService,
final LinkService linkService,
final PermissionsService permissionsService,
- final AuthorizationService authorizationService) {
- super(authorizationService);
+ final AuthorizationService authorizationService,
+ final EventService eventService) {
+ super(authorizationService, eventService);
this.registryService = registryService;
this.linkService = linkService;
this.permissionsService = permissionsService;
@@ -110,7 +113,10 @@ public class BucketResource extends AuthorizableApplicationResource {
@ApiParam(value = "The bucket to create", required = true)
final Bucket bucket) {
authorizeAccess(RequestAction.WRITE);
+
final Bucket createdBucket = registryService.createBucket(bucket);
+ publish(EventFactory.bucketCreated(createdBucket));
+
permissionsService.populateBucketPermissions(createdBucket);
linkService.populateBucketLinks(createdBucket);
return Response.status(Response.Status.OK).entity(createdBucket).build();
@@ -224,6 +230,8 @@ public class BucketResource extends AuthorizableApplicationResource {
authorizeBucketAccess(RequestAction.WRITE, bucketId);
final Bucket updatedBucket = registryService.updateBucket(bucket);
+ publish(EventFactory.bucketUpdated(updatedBucket));
+
permissionsService.populateBucketPermissions(updatedBucket);
linkService.populateBucketLinks(updatedBucket);
return Response.status(Response.Status.OK).entity(updatedBucket).build();
@@ -256,7 +264,10 @@ public class BucketResource extends AuthorizableApplicationResource {
throw new BadRequestException("Bucket id cannot be blank");
}
authorizeBucketAccess(RequestAction.DELETE, bucketId);
+
final Bucket deletedBucket = registryService.deleteBucket(bucketId);
+ publish(EventFactory.bucketDeleted(deletedBucket));
+
return Response.status(Response.Status.OK).entity(deletedBucket).build();
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/FlowResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/FlowResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/FlowResource.java
index 2c04d60..0228334 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/FlowResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/FlowResource.java
@@ -25,6 +25,7 @@ import io.swagger.annotations.Authorization;
import io.swagger.annotations.Extension;
import io.swagger.annotations.ExtensionProperty;
import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.registry.event.EventService;
import org.apache.nifi.registry.field.Fields;
import org.apache.nifi.registry.flow.VersionedFlow;
import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
@@ -65,8 +66,9 @@ public class FlowResource extends AuthorizableApplicationResource {
public FlowResource(final RegistryService registryService,
final LinkService linkService,
final PermissionsService permissionsService,
- final AuthorizationService authorizationService) {
- super(authorizationService);
+ final AuthorizationService authorizationService,
+ final EventService eventService) {
+ super(authorizationService, eventService);
this.registryService = registryService;
this.linkService = linkService;
this.permissionsService = permissionsService;
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ItemResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ItemResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ItemResource.java
index e7ae6be..02b63d2 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ItemResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/ItemResource.java
@@ -25,6 +25,7 @@ import io.swagger.annotations.Authorization;
import io.swagger.annotations.Extension;
import io.swagger.annotations.ExtensionProperty;
import org.apache.nifi.registry.bucket.BucketItem;
+import org.apache.nifi.registry.event.EventService;
import org.apache.nifi.registry.field.Fields;
import org.apache.nifi.registry.security.authorization.RequestAction;
import org.apache.nifi.registry.service.AuthorizationService;
@@ -73,8 +74,9 @@ public class ItemResource extends AuthorizableApplicationResource {
final RegistryService registryService,
final LinkService linkService,
final PermissionsService permissionsService,
- final AuthorizationService authorizationService) {
- super(authorizationService);
+ final AuthorizationService authorizationService,
+ final EventService eventService) {
+ super(authorizationService, eventService);
this.registryService = registryService;
this.linkService = linkService;
this.permissionsService = permissionsService;
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/TenantResource.java
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/TenantResource.java b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/TenantResource.java
index d00038d..7215ccb 100644
--- a/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/TenantResource.java
+++ b/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/TenantResource.java
@@ -27,6 +27,7 @@ import io.swagger.annotations.ExtensionProperty;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.registry.authorization.User;
import org.apache.nifi.registry.authorization.UserGroup;
+import org.apache.nifi.registry.event.EventService;
import org.apache.nifi.registry.exception.ResourceNotFoundException;
import org.apache.nifi.registry.security.authorization.Authorizer;
import org.apache.nifi.registry.security.authorization.AuthorizerCapabilityDetection;
@@ -70,8 +71,8 @@ public class TenantResource extends AuthorizableApplicationResource {
private Authorizer authorizer;
@Autowired
- public TenantResource(AuthorizationService authorizationService) {
- super(authorizationService);
+ public TenantResource(AuthorizationService authorizationService, EventService eventService) {
+ super(authorizationService, eventService);
authorizer = authorizationService.getAuthorizer();
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-web-api/src/main/resources/META-INF/NOTICE
----------------------------------------------------------------------
diff --git a/nifi-registry-web-api/src/main/resources/META-INF/NOTICE b/nifi-registry-web-api/src/main/resources/META-INF/NOTICE
index e715d71..737dad8 100644
--- a/nifi-registry-web-api/src/main/resources/META-INF/NOTICE
+++ b/nifi-registry-web-api/src/main/resources/META-INF/NOTICE
@@ -66,7 +66,7 @@ The following binary components are provided under the Apache Software License v
(ASLv2) Spring Security
The following NOTICE information applies:
- Spring Framework 5.0.0.RELEASE
+ Spring Framework 5.0.5.RELEASE
Copyright (c) 2002-2017 Pivotal, Inc.
This product includes software developed by Spring Security
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 94bb5e0..86deaa4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -110,8 +110,8 @@
<jax.rs.api.version>2.1</jax.rs.api.version>
<jersey.version>2.26</jersey.version>
<jackson.version>2.9.2</jackson.version>
- <spring.boot.version>2.0.0.M7</spring.boot.version>
- <spring.security.version>5.0.0.RELEASE</spring.security.version>
+ <spring.boot.version>2.0.2.RELEASE</spring.boot.version>
+ <spring.security.version>5.0.5.RELEASE</spring.security.version>
<flyway.version>4.2.0</flyway.version>
</properties>
[3/3] nifi-registry git commit: NIFIREG-160 - Making event hooks
asynchronous
Posted by pv...@apache.org.
NIFIREG-160 - Making event hooks asynchronous
Signed-off-by: Pierre Villard <pi...@gmail.com>
This closes #117.
Project: http://git-wip-us.apache.org/repos/asf/nifi-registry/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-registry/commit/51b6c78b
Tree: http://git-wip-us.apache.org/repos/asf/nifi-registry/tree/51b6c78b
Diff: http://git-wip-us.apache.org/repos/asf/nifi-registry/diff/51b6c78b
Branch: refs/heads/master
Commit: 51b6c78b6faed2e9540c7234faf98f6098970afa
Parents: 872a5b5
Author: Bryan Bende <bb...@apache.org>
Authored: Tue May 8 13:38:33 2018 -0400
Committer: Pierre Villard <pi...@gmail.com>
Committed: Tue May 15 19:18:13 2018 +0200
----------------------------------------------------------------------
nifi-registry-assembly/NOTICE | 2 +-
.../nifi/registry/event/EventFactory.java | 94 +++++++++++
.../nifi/registry/event/EventService.java | 112 +++++++++++++
.../nifi/registry/event/StandardEvent.java | 124 ++++++++++++++
.../nifi/registry/event/StandardEventField.java | 49 ++++++
.../registry/extension/ExtensionManager.java | 4 +-
.../nifi/registry/provider/ProviderFactory.java | 4 +-
.../provider/StandardProviderFactory.java | 46 +++---
.../flow/StandardFlowSnapshotContext.java | 9 --
.../provider/hook/LoggingEventHookProvider.java | 57 +++++++
.../provider/hook/ScriptEventHookProvider.java | 98 +++++++++++
.../provider/hook/ScriptFlowHookProvider.java | 152 -----------------
.../nifi/registry/service/RegistryService.java | 89 ----------
....apache.nifi.registry.hook.EventHookProvider | 16 ++
...g.apache.nifi.registry.hook.FlowHookProvider | 15 --
.../src/main/xsd/providers.xsd | 2 +-
.../nifi/registry/event/TestEventFactory.java | 161 +++++++++++++++++++
.../nifi/registry/event/TestEventService.java | 97 +++++++++++
.../nifi/registry/event/TestStandardEvent.java | 47 ++++++
.../hook/TestScriptEventHookProvider.java | 45 ++++++
.../hook/TestScriptFlowHookProvider.java | 45 ------
.../registry/service/TestRegistryService.java | 5 +-
.../provider/hook/bad-script-provider.xml | 6 +-
.../org/apache/nifi/registry/hook/Event.java | 50 ++++++
.../apache/nifi/registry/hook/EventField.java | 33 ++++
.../nifi/registry/hook/EventFieldName.java | 30 ++++
.../nifi/registry/hook/EventHookException.java | 31 ++++
.../nifi/registry/hook/EventHookProvider.java | 39 +++++
.../apache/nifi/registry/hook/EventType.java | 72 +++++++++
.../nifi/registry/hook/FlowHookEvent.java | 31 ----
.../nifi/registry/hook/FlowHookException.java | 31 ----
.../nifi/registry/hook/FlowHookProvider.java | 81 ----------
.../src/main/resources/conf/logback.xml | 23 +++
.../src/main/resources/conf/providers.xml | 13 +-
.../registry/web/api/AccessPolicyResource.java | 6 +-
.../nifi/registry/web/api/AccessResource.java | 5 +-
.../registry/web/api/ApplicationResource.java | 14 ++
.../api/AuthorizableApplicationResource.java | 5 +-
.../registry/web/api/BucketFlowResource.java | 14 +-
.../nifi/registry/web/api/BucketResource.java | 15 +-
.../nifi/registry/web/api/FlowResource.java | 6 +-
.../nifi/registry/web/api/ItemResource.java | 6 +-
.../nifi/registry/web/api/TenantResource.java | 5 +-
.../src/main/resources/META-INF/NOTICE | 2 +-
pom.xml | 4 +-
45 files changed, 1286 insertions(+), 509 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-assembly/NOTICE
----------------------------------------------------------------------
diff --git a/nifi-registry-assembly/NOTICE b/nifi-registry-assembly/NOTICE
index d5fe07d..a2214e3 100644
--- a/nifi-registry-assembly/NOTICE
+++ b/nifi-registry-assembly/NOTICE
@@ -170,7 +170,7 @@ The following binary components are provided under the Apache Software License v
(ASLv2) Spring Security
The following NOTICE information applies:
- Spring Framework 5.0.0.RELEASE
+ Spring Framework 5.0.5.RELEASE
Copyright (c) 2002-2017 Pivotal, Inc.
This product includes software developed by Spring Security
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventFactory.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventFactory.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventFactory.java
new file mode 100644
index 0000000..ba9e775
--- /dev/null
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventFactory.java
@@ -0,0 +1,94 @@
+/*
+ * 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.event;
+
+import org.apache.nifi.registry.bucket.Bucket;
+import org.apache.nifi.registry.flow.VersionedFlow;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventFieldName;
+import org.apache.nifi.registry.hook.EventType;
+import org.apache.nifi.registry.security.authorization.user.NiFiUserUtils;
+
+/**
+ * Factory to create Events from domain objects.
+ */
+public class EventFactory {
+
+ public static Event bucketCreated(final Bucket bucket) {
+ return new StandardEvent.Builder()
+ .eventType(EventType.CREATE_BUCKET)
+ .addField(EventFieldName.BUCKET_ID, bucket.getIdentifier())
+ .addField(EventFieldName.USER, NiFiUserUtils.getNiFiUserIdentity())
+ .build();
+ }
+
+ public static Event bucketUpdated(final Bucket bucket) {
+ return new StandardEvent.Builder()
+ .eventType(EventType.UPDATE_BUCKET)
+ .addField(EventFieldName.BUCKET_ID, bucket.getIdentifier())
+ .addField(EventFieldName.USER, NiFiUserUtils.getNiFiUserIdentity())
+ .build();
+ }
+
+ public static Event bucketDeleted(final Bucket bucket) {
+ return new StandardEvent.Builder()
+ .eventType(EventType.DELETE_BUCKET)
+ .addField(EventFieldName.BUCKET_ID, bucket.getIdentifier())
+ .addField(EventFieldName.USER, NiFiUserUtils.getNiFiUserIdentity())
+ .build();
+ }
+
+ public static Event flowCreated(final VersionedFlow versionedFlow) {
+ return new StandardEvent.Builder()
+ .eventType(EventType.CREATE_FLOW)
+ .addField(EventFieldName.BUCKET_ID, versionedFlow.getBucketIdentifier())
+ .addField(EventFieldName.FLOW_ID, versionedFlow.getIdentifier())
+ .addField(EventFieldName.USER, NiFiUserUtils.getNiFiUserIdentity())
+ .build();
+ }
+
+ public static Event flowUpdated(final VersionedFlow versionedFlow) {
+ return new StandardEvent.Builder()
+ .eventType(EventType.UPDATE_FLOW)
+ .addField(EventFieldName.BUCKET_ID, versionedFlow.getBucketIdentifier())
+ .addField(EventFieldName.FLOW_ID, versionedFlow.getIdentifier())
+ .addField(EventFieldName.USER, NiFiUserUtils.getNiFiUserIdentity())
+ .build();
+ }
+
+ public static Event flowDeleted(final VersionedFlow versionedFlow) {
+ return new StandardEvent.Builder()
+ .eventType(EventType.DELETE_FLOW)
+ .addField(EventFieldName.BUCKET_ID, versionedFlow.getBucketIdentifier())
+ .addField(EventFieldName.FLOW_ID, versionedFlow.getIdentifier())
+ .addField(EventFieldName.USER, NiFiUserUtils.getNiFiUserIdentity())
+ .build();
+ }
+
+ public static Event flowVersionCreated(final VersionedFlowSnapshot versionedFlowSnapshot) {
+ return new StandardEvent.Builder()
+ .eventType(EventType.CREATE_FLOW_VERSION)
+ .addField(EventFieldName.BUCKET_ID, versionedFlowSnapshot.getSnapshotMetadata().getBucketIdentifier())
+ .addField(EventFieldName.FLOW_ID, versionedFlowSnapshot.getSnapshotMetadata().getFlowIdentifier())
+ .addField(EventFieldName.VERSION, String.valueOf(versionedFlowSnapshot.getSnapshotMetadata().getVersion()))
+ .addField(EventFieldName.USER, versionedFlowSnapshot.getSnapshotMetadata().getAuthor())
+ .addField(EventFieldName.COMMENT, versionedFlowSnapshot.getSnapshotMetadata().getComments())
+ .build();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventService.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventService.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventService.java
new file mode 100644
index 0000000..ef30f84
--- /dev/null
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/EventService.java
@@ -0,0 +1,112 @@
+/*
+ * 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.event;
+
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventHookProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Service used for publishing events and passing events to the hook providers.
+ */
+@Service
+public class EventService implements DisposableBean {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EventService.class);
+
+ // Should only be a few events in the queue at a time, but setting a capacity just so it isn't unbounded
+ static final int EVENT_QUEUE_SIZE = 10_000;
+
+ private final BlockingQueue<Event> eventQueue;
+ private final ExecutorService scheduledExecutorService;
+ private final List<EventHookProvider> eventHookProviders;
+
+ @Autowired
+ public EventService(final List<EventHookProvider> eventHookProviders) {
+ this.eventQueue = new LinkedBlockingQueue<>(EVENT_QUEUE_SIZE);
+ this.scheduledExecutorService = Executors.newSingleThreadExecutor();
+ this.eventHookProviders = new ArrayList<>(eventHookProviders);
+ }
+
+ @PostConstruct
+ public void postConstruct() {
+ LOGGER.info("Starting event consumer...");
+
+ this.scheduledExecutorService.execute(() -> {
+ while (!Thread.interrupted()) {
+ try {
+ final Event event = eventQueue.poll(1000, TimeUnit.MILLISECONDS);
+ if (event == null) {
+ continue;
+ }
+
+ // event was available so notify each provider, contain errors per-provider
+ for(final EventHookProvider provider : eventHookProviders) {
+ try {
+ provider.handle(event);
+ } catch (Exception e) {
+ LOGGER.error("Error handling event hook", e);
+ }
+ }
+ } catch (InterruptedException e) {
+ LOGGER.warn("Interrupted while polling event queue");
+ return;
+ }
+ }
+ });
+
+ LOGGER.info("Event consumer started!");
+ }
+
+ @Override
+ public void destroy() throws Exception {
+ LOGGER.info("Shutting down event consumer...");
+ this.scheduledExecutorService.shutdownNow();
+ LOGGER.info("Event consumer shutdown!");
+ }
+
+ public void publish(final Event event) {
+ if (event == null) {
+ return;
+ }
+
+ try {
+ event.validate();
+
+ final boolean queued = eventQueue.offer(event);
+ if (!queued) {
+ LOGGER.error("Unable to queue event because queue is full");
+ }
+ } catch (IllegalStateException e) {
+ LOGGER.error("Invalid event due to: " + e.getMessage(), e);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEvent.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEvent.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEvent.java
new file mode 100644
index 0000000..4ad459d
--- /dev/null
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEvent.java
@@ -0,0 +1,124 @@
+/*
+ * 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.event;
+
+import org.apache.commons.lang3.Validate;
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventField;
+import org.apache.nifi.registry.hook.EventFieldName;
+import org.apache.nifi.registry.hook.EventType;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Standard implementation of Event.
+ */
+public class StandardEvent implements Event {
+
+ private final EventType eventType;
+
+ private final List<EventField> eventFields;
+
+ private StandardEvent(final Builder builder) {
+ this.eventType = builder.eventType;
+ this.eventFields = Collections.unmodifiableList(builder.eventFields == null
+ ? Collections.emptyList() : new ArrayList<>(builder.eventFields));
+ Validate.notNull(this.eventType);
+ }
+
+ @Override
+ public EventType getEventType() {
+ return eventType;
+ }
+
+ @Override
+ public List<EventField> getFields() {
+ return eventFields;
+ }
+
+ @Override
+ public EventField getField(final EventFieldName fieldName) {
+ if (fieldName == null) {
+ return null;
+ }
+
+ return eventFields.stream().filter(e -> fieldName.equals(e.getName())).findFirst().orElse(null);
+ }
+
+ @Override
+ public void validate() throws IllegalStateException {
+ final int numProvidedFields = eventFields.size();
+ final int numRequiredFields = eventType.getFieldNames().size();
+
+ if (numProvidedFields != numRequiredFields) {
+ throw new IllegalStateException(numRequiredFields + " fields were required, but only " + numProvidedFields + " were provided");
+ }
+
+ for (int i=0; i < numRequiredFields; i++) {
+ final EventFieldName required = eventType.getFieldNames().get(i);
+ final EventFieldName provided = eventFields.get(i).getName();
+ if (!required.equals(provided)) {
+ throw new IllegalStateException("Expected " + required.name() + ", but found " + provided.name());
+ }
+ }
+ }
+
+ /**
+ * Builder for Events.
+ */
+ public static class Builder {
+
+ private EventType eventType;
+ private List<EventField> eventFields = new ArrayList<>();
+
+ public Builder eventType(final EventType eventType) {
+ this.eventType = eventType;
+ return this;
+ }
+
+ public Builder addField(final EventFieldName name, final String value) {
+ this.eventFields.add(new StandardEventField(name, value));
+ return this;
+ }
+
+ public Builder addField(final EventField arg) {
+ if (arg != null) {
+ this.eventFields.add(arg);
+ }
+ return this;
+ }
+
+ public Builder addFields(final Collection<EventField> fields) {
+ if (fields != null) {
+ this.eventFields.addAll(fields);
+ }
+ return this;
+ }
+
+ public Builder clearFields() {
+ this.eventFields.clear();
+ return this;
+ }
+
+ public Event build() {
+ return new StandardEvent(this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEventField.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEventField.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEventField.java
new file mode 100644
index 0000000..7ff94c5
--- /dev/null
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/event/StandardEventField.java
@@ -0,0 +1,49 @@
+/*
+ * 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.event;
+
+import org.apache.commons.lang3.Validate;
+import org.apache.nifi.registry.hook.EventField;
+import org.apache.nifi.registry.hook.EventFieldName;
+
+/**
+ * Standard implementation of EventField.
+ */
+public class StandardEventField implements EventField {
+
+ private final EventFieldName name;
+
+ private final String value;
+
+ public StandardEventField(final EventFieldName name, final String value) {
+ this.name = name;
+ this.value = value;
+ Validate.notNull(this.name);
+ Validate.notBlank(this.value);
+ }
+
+ @Override
+ public EventFieldName getName() {
+ return name;
+ }
+
+ @Override
+ public String getValue() {
+ return value;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
index 7f2b9e1..ca3259d 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/extension/ExtensionManager.java
@@ -17,12 +17,12 @@
package org.apache.nifi.registry.extension;
import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.registry.hook.EventHookProvider;
import org.apache.nifi.registry.security.authentication.IdentityProvider;
import org.apache.nifi.registry.security.authorization.AccessPolicyProvider;
import org.apache.nifi.registry.security.authorization.Authorizer;
import org.apache.nifi.registry.security.authorization.UserGroupProvider;
import org.apache.nifi.registry.flow.FlowPersistenceProvider;
-import org.apache.nifi.registry.hook.FlowHookProvider;
import org.apache.nifi.registry.properties.NiFiRegistryProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,7 +57,7 @@ public class ExtensionManager {
classes.add(AccessPolicyProvider.class);
classes.add(Authorizer.class);
classes.add(IdentityProvider.class);
- classes.add(FlowHookProvider.class);
+ classes.add(EventHookProvider.class);
EXTENSION_CLASSES = Collections.unmodifiableList(classes);
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
index bb62eb2..a3f3276 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/ProviderFactory.java
@@ -19,7 +19,7 @@ package org.apache.nifi.registry.provider;
import java.util.List;
import org.apache.nifi.registry.flow.FlowPersistenceProvider;
-import org.apache.nifi.registry.hook.FlowHookProvider;
+import org.apache.nifi.registry.hook.EventHookProvider;
/**
* A factory for obtaining the configured providers.
@@ -41,6 +41,6 @@ public interface ProviderFactory {
/**
* @return the configured FlowHookProviders
*/
- List<FlowHookProvider> getFlowHookProviders();
+ List<EventHookProvider> getEventHookProviders();
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
index 34f307c..65ba914 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/StandardProviderFactory.java
@@ -18,7 +18,7 @@ package org.apache.nifi.registry.provider;
import org.apache.nifi.registry.extension.ExtensionManager;
import org.apache.nifi.registry.flow.FlowPersistenceProvider;
-import org.apache.nifi.registry.hook.FlowHookProvider;
+import org.apache.nifi.registry.hook.EventHookProvider;
import org.apache.nifi.registry.properties.NiFiRegistryProperties;
import org.apache.nifi.registry.provider.generated.Property;
import org.apache.nifi.registry.provider.generated.Providers;
@@ -74,7 +74,7 @@ public class StandardProviderFactory implements ProviderFactory {
private final AtomicReference<Providers> providersHolder = new AtomicReference<>(null);
private FlowPersistenceProvider flowPersistenceProvider;
- private List<FlowHookProvider> flowHookProviders;
+ private List<EventHookProvider> eventHookProviders;
@Autowired
public StandardProviderFactory(final NiFiRegistryProperties properties, final ExtensionManager extensionManager) {
@@ -156,52 +156,52 @@ public class StandardProviderFactory implements ProviderFactory {
@Bean
@Override
- public List<FlowHookProvider> getFlowHookProviders() {
- if (flowHookProviders == null) {
- flowHookProviders = new ArrayList<FlowHookProvider>();
+ public List<EventHookProvider> getEventHookProviders() {
+ if (eventHookProviders == null) {
+ eventHookProviders = new ArrayList<>();
if (providersHolder.get() == null) {
throw new ProviderFactoryException("ProviderFactory must be initialized before obtaining a Provider");
}
final Providers providers = providersHolder.get();
- final List<org.apache.nifi.registry.provider.generated.Provider> jaxbFlowHookProvider = providers.getFlowHookProvider();
+ final List<org.apache.nifi.registry.provider.generated.Provider> jaxbHookProvider = providers.getEventHookProvider();
- if(jaxbFlowHookProvider == null || jaxbFlowHookProvider.isEmpty()) {
+ if(jaxbHookProvider == null || jaxbHookProvider.isEmpty()) {
// no hook provided
- return flowHookProviders;
+ return eventHookProviders;
}
- for (org.apache.nifi.registry.provider.generated.Provider flowHookProvider : jaxbFlowHookProvider) {
+ for (org.apache.nifi.registry.provider.generated.Provider hookProvider : jaxbHookProvider) {
- final String flowHookProviderClassName = flowHookProvider.getClazz();
- FlowHookProvider hook;
+ final String hookProviderClassName = hookProvider.getClazz();
+ EventHookProvider hook;
try {
- final ClassLoader classLoader = extensionManager.getExtensionClassLoader(flowHookProviderClassName);
+ final ClassLoader classLoader = extensionManager.getExtensionClassLoader(hookProviderClassName);
if (classLoader == null) {
- throw new IllegalStateException("Extension not found in any of the configured class loaders: " + flowHookProviderClassName);
+ throw new IllegalStateException("Extension not found in any of the configured class loaders: " + hookProviderClassName);
}
- final Class<?> rawFlowHookProviderClass = Class.forName(flowHookProviderClassName, true, classLoader);
- final Class<? extends FlowHookProvider> flowHookProviderClass = rawFlowHookProviderClass.asSubclass(FlowHookProvider.class);
+ final Class<?> rawHookProviderClass = Class.forName(hookProviderClassName, true, classLoader);
+ final Class<? extends EventHookProvider> hookProviderClass = rawHookProviderClass.asSubclass(EventHookProvider.class);
- final Constructor constructor = flowHookProviderClass.getConstructor();
- hook = (FlowHookProvider) constructor.newInstance();
+ final Constructor constructor = hookProviderClass.getConstructor();
+ hook = (EventHookProvider) constructor.newInstance();
- LOGGER.info("Instantiated FlowHookProvider with class name {}", new Object[] {flowHookProviderClassName});
+ LOGGER.info("Instantiated EventHookProvider with class name {}", new Object[] {hookProviderClassName});
} catch (Exception e) {
- throw new ProviderFactoryException("Error creating FlowHookProvider with class name: " + flowHookProviderClassName, e);
+ throw new ProviderFactoryException("Error creating EventHookProvider with class name: " + hookProviderClassName, e);
}
- final ProviderConfigurationContext configurationContext = createConfigurationContext(flowHookProvider.getProperty());
+ final ProviderConfigurationContext configurationContext = createConfigurationContext(hookProvider.getProperty());
hook.onConfigured(configurationContext);
- flowHookProviders.add(hook);
- LOGGER.info("Configured FlowHookProvider with class name {}", new Object[] {flowHookProviderClassName});
+ eventHookProviders.add(hook);
+ LOGGER.info("Configured EventHookProvider with class name {}", new Object[] {hookProviderClassName});
}
}
- return flowHookProviders;
+ return eventHookProviders;
}
private ProviderConfigurationContext createConfigurationContext(final List<Property> configProperties) {
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
index 1f6f840..1728513 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/StandardFlowSnapshotContext.java
@@ -35,7 +35,6 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
private final String comments;
private final String author;
private final long snapshotTimestamp;
- private final String author;
private StandardFlowSnapshotContext(final Builder builder) {
this.bucketId = builder.bucketId;
@@ -46,7 +45,6 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
this.comments = builder.comments;
this.author = builder.author;
this.snapshotTimestamp = builder.snapshotTimestamp;
- this.author = builder.author;
Validate.notBlank(bucketId);
Validate.notBlank(bucketName);
@@ -109,7 +107,6 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
private String comments;
private String author;
private long snapshotTimestamp;
- private String author;
public Builder() {
@@ -124,7 +121,6 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
comments(snapshotMetadata.getComments());
author(snapshotMetadata.getAuthor());
snapshotTimestamp(snapshotMetadata.getTimestamp());
- author(snapshotMetadata.getAuthor());
}
public Builder bucketId(final String bucketId) {
@@ -167,11 +163,6 @@ public class StandardFlowSnapshotContext implements FlowSnapshotContext {
return this;
}
- public Builder author(final String author) {
- this.author = author;
- return this;
- }
-
public StandardFlowSnapshotContext build() {
return new StandardFlowSnapshotContext(this);
}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/LoggingEventHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/LoggingEventHookProvider.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/LoggingEventHookProvider.java
new file mode 100644
index 0000000..a942d5a
--- /dev/null
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/LoggingEventHookProvider.java
@@ -0,0 +1,57 @@
+/*
+ * 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.provider.hook;
+
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventField;
+import org.apache.nifi.registry.hook.EventHookException;
+import org.apache.nifi.registry.hook.EventHookProvider;
+import org.apache.nifi.registry.provider.ProviderConfigurationContext;
+import org.apache.nifi.registry.provider.ProviderCreationException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LoggingEventHookProvider implements EventHookProvider {
+
+ static final Logger LOGGER = LoggerFactory.getLogger(LoggingEventHookProvider.class);
+
+ @Override
+ public void onConfigured(final ProviderConfigurationContext configurationContext) throws ProviderCreationException {
+ // nothing to do
+ }
+
+ @Override
+ public void handle(final Event event) throws EventHookException {
+ final StringBuilder builder = new StringBuilder()
+ .append(event.getEventType())
+ .append(" [");
+
+ int count = 0;
+ for (final EventField argument : event.getFields()) {
+ if (count > 0) {
+ builder.append(", ");
+ }
+ builder.append(argument.getName()).append("=").append(argument.getValue());
+ count++;
+ }
+
+ builder.append("] ");
+
+ LOGGER.info(builder.toString());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptEventHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptEventHookProvider.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptEventHookProvider.java
new file mode 100644
index 0000000..4900a2b
--- /dev/null
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptEventHookProvider.java
@@ -0,0 +1,98 @@
+/*
+ * 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.provider.hook;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventField;
+import org.apache.nifi.registry.hook.EventHookProvider;
+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;
+
+/**
+ * A EventHookProvider that is used to execute a script to handle the event.
+ */
+public class ScriptEventHookProvider implements EventHookProvider {
+
+ static final Logger LOGGER = LoggerFactory.getLogger(ScriptEventHookProvider.class);
+ static final String SCRIPT_PATH_PROP = "Script Path";
+ static final String SCRIPT_WORKDIR_PROP = "Working Directory";
+ private File scriptFile;
+ private File workDirFile;
+
+ @Override
+ public void handle(final Event event) {
+ List<String> command = new ArrayList<String>();
+ command.add(scriptFile.getAbsolutePath());
+ command.add(event.getEventType().name());
+
+ for (EventField arg : event.getFields()) {
+ command.add(arg.getValue());
+ }
+
+ final String commandString = StringUtils.join(command, " ");
+ final ProcessBuilder builder = new ProcessBuilder(command);
+ builder.directory(workDirFile);
+ LOGGER.debug("Execution of " + commandString);
+
+ try {
+ builder.start();
+ } catch (IOException e) {
+ LOGGER.error("Execution of {0} failed with: {1}", new Object[] { commandString, e.getLocalizedMessage() }, e);
+ }
+ }
+
+ @Override
+ public void onConfigured(ProviderConfigurationContext configurationContext) throws ProviderCreationException {
+ final Map<String,String> props = configurationContext.getProperties();
+ if (!props.containsKey(SCRIPT_PATH_PROP)) {
+ throw new ProviderCreationException("The property " + SCRIPT_PATH_PROP + " must be provided");
+ }
+
+ final String scripPath = props.get(SCRIPT_PATH_PROP);
+ if (StringUtils.isBlank(scripPath)) {
+ throw new ProviderCreationException("The property " + SCRIPT_PATH_PROP + " cannot be null or blank");
+ }
+
+ if(props.containsKey(SCRIPT_WORKDIR_PROP) && !StringUtils.isBlank(props.get(SCRIPT_WORKDIR_PROP))) {
+ final String workdir = props.get(SCRIPT_WORKDIR_PROP);
+ try {
+ workDirFile = new File(workdir);
+ FileUtils.ensureDirectoryExistAndCanRead(workDirFile);
+ } catch (IOException e) {
+ throw new ProviderCreationException("The working directory " + workdir + " cannot be read.");
+ }
+ }
+
+ scriptFile = new File(scripPath);
+ if(scriptFile.isFile() && scriptFile.canExecute()) {
+ LOGGER.info("Configured ScriptEventHookProvider with script {}", new Object[] {scriptFile.getAbsolutePath()});
+ } else {
+ throw new ProviderCreationException("The script file " + scriptFile.getAbsolutePath() + " cannot be executed.");
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java
deleted file mode 100644
index d57df16..0000000
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/hook/ScriptFlowHookProvider.java
+++ /dev/null
@@ -1,152 +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.provider.hook;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.registry.flow.FlowSnapshotContext;
-import org.apache.nifi.registry.hook.FlowHookEvent;
-import org.apache.nifi.registry.hook.FlowHookException;
-import org.apache.nifi.registry.hook.FlowHookProvider;
-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;
-
-/**
- * A FlowHookProvider that is used to execute a script before a flow snapshot version is committed.
- */
-public class ScriptFlowHookProvider implements FlowHookProvider {
-
- static final Logger LOGGER = LoggerFactory.getLogger(ScriptFlowHookProvider.class);
- static final String SCRIPT_PATH_PROP = "Script Path";
- static final String SCRIPT_WORKDIR_PROP = "Working Directory";
- private File scriptFile;
- private File workDirFile;
-
- @Override
- public void postCreateBucket(String bucketId) throws FlowHookException {
- this.executeScript(FlowHookEvent.CREATE_BUCKET, bucketId, null, null, null, null);
- }
-
- @Override
- public void postCreateFlow(String bucketId, String flowId) throws FlowHookException {
- this.executeScript(FlowHookEvent.CREATE_FLOW, bucketId, flowId, null, null, null);
- }
-
- @Override
- public void postDeleteBucket(String bucketId) throws FlowHookException {
- this.executeScript(FlowHookEvent.DELETE_BUCKET, bucketId, null, null, null, null);
- }
-
- @Override
- public void postDeleteFlow(String bucketId, String flowId) throws FlowHookException {
- this.executeScript(FlowHookEvent.DELETE_FLOW, bucketId, flowId, null, null, null);
- }
-
- @Override
- public void postDeleteFlowVersion(String bucketId, String flowId, int version) throws FlowHookException {
- this.executeScript(FlowHookEvent.DELETE_FLOW, bucketId, flowId, Integer.toString(version), null, null);
- }
-
- @Override
- public void postUpdateBucket(String bucketId) throws FlowHookException {
- this.executeScript(FlowHookEvent.UPDATE_BUCKET, bucketId, null, null, null, null);
- }
-
- @Override
- public void postUpdateFlow(String bucketId, String flowId) throws FlowHookException {
- this.executeScript(FlowHookEvent.UPDATE_FLOW, bucketId, flowId, null, null, null);
- }
-
- @Override
- public void postCreateFlowVersion(final FlowSnapshotContext flowSnapshotContext) throws FlowHookException {
- this.executeScript(FlowHookEvent.CREATE_VERSION, flowSnapshotContext.getBucketId(), flowSnapshotContext.getFlowId(),
- Integer.toString(flowSnapshotContext.getVersion()), flowSnapshotContext.getComments(), flowSnapshotContext.getAuthor());
- }
-
- private void executeScript(final FlowHookEvent eventType, String bucketId, String flowId, String version, String comment, String author) {
- List<String> command = new ArrayList<String>();
- command.add(scriptFile.getAbsolutePath());
- command.add(eventType.name());
- command.add(bucketId);
-
- if(flowId != null) {
- command.add(flowId);
- }
-
- if(version != null) {
- command.add(version);
- }
-
- if(comment != null) {
- command.add(comment);
- }
-
- if(author != null) {
- command.add(author);
- }
-
- final String commandString = StringUtils.join(command, " ");
- final ProcessBuilder builder = new ProcessBuilder(command);
- builder.directory(workDirFile);
- LOGGER.debug("Execution of " + commandString);
-
- try {
- builder.start();
- } catch (IOException e) {
- LOGGER.error("Execution of {0} failed with: {1}", new Object[] { commandString, e.getLocalizedMessage() }, e);
- }
- }
-
- @Override
- public void onConfigured(ProviderConfigurationContext configurationContext) throws ProviderCreationException {
- final Map<String,String> props = configurationContext.getProperties();
- if (!props.containsKey(SCRIPT_PATH_PROP)) {
- throw new ProviderCreationException("The property " + SCRIPT_PATH_PROP + " must be provided");
- }
-
- final String scripPath = props.get(SCRIPT_PATH_PROP);
- if (StringUtils.isBlank(scripPath)) {
- throw new ProviderCreationException("The property " + SCRIPT_PATH_PROP + " cannot be null or blank");
- }
-
- if(props.containsKey(SCRIPT_WORKDIR_PROP) && !StringUtils.isBlank(props.get(SCRIPT_WORKDIR_PROP))) {
- final String workdir = props.get(SCRIPT_WORKDIR_PROP);
- try {
- workDirFile = new File(workdir);
- FileUtils.ensureDirectoryExistAndCanRead(workDirFile);
- } catch (IOException e) {
- throw new ProviderCreationException("The working directory " + workdir + " cannot be read.");
- }
- }
-
- scriptFile = new File(scripPath);
- if(scriptFile.isFile() && scriptFile.canExecute()) {
- LOGGER.info("Configured ScriptFlowHookProvider with script {}", new Object[] {scriptFile.getAbsolutePath()});
- } else {
- throw new ProviderCreationException("The script file " + scriptFile.getAbsolutePath() + " cannot be executed.");
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
index 95c62c3..23f1d14 100644
--- a/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
+++ b/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
@@ -42,7 +42,6 @@ import org.apache.nifi.registry.flow.diff.FlowComparison;
import org.apache.nifi.registry.flow.diff.FlowDifference;
import org.apache.nifi.registry.flow.diff.StandardComparableDataFlow;
import org.apache.nifi.registry.flow.diff.StandardFlowComparator;
-import org.apache.nifi.registry.hook.FlowHookProvider;
import org.apache.nifi.registry.provider.flow.StandardFlowSnapshotContext;
import org.apache.nifi.registry.serialization.Serializer;
import org.slf4j.Logger;
@@ -85,7 +84,6 @@ public class RegistryService {
private final MetadataService metadataService;
private final FlowPersistenceProvider flowPersistenceProvider;
- private final List<FlowHookProvider> flowHookProviders;
private final Serializer<VersionedProcessGroup> processGroupSerializer;
private final Validator validator;
@@ -96,12 +94,10 @@ public class RegistryService {
@Autowired
public RegistryService(final MetadataService metadataService,
final FlowPersistenceProvider flowPersistenceProvider,
- final List<FlowHookProvider> flowHookProviders,
final Serializer<VersionedProcessGroup> processGroupSerializer,
final Validator validator) {
this.metadataService = metadataService;
this.flowPersistenceProvider = flowPersistenceProvider;
- this.flowHookProviders = flowHookProviders;
this.processGroupSerializer = processGroupSerializer;
this.validator = validator;
Validate.notNull(this.metadataService);
@@ -138,17 +134,6 @@ public class RegistryService {
}
final BucketEntity createdBucket = metadataService.createBucket(DataModelMapper.map(bucket));
-
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postCreateBucket(createdBucket.getId());
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
return DataModelMapper.map(createdBucket);
} finally {
writeLock.unlock();
@@ -240,17 +225,6 @@ public class RegistryService {
// perform the actual update
final BucketEntity updatedBucket = metadataService.updateBucket(existingBucketById);
-
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postUpdateBucket(existingBucketById.getId());
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
return DataModelMapper.map(updatedBucket);
} finally {
writeLock.unlock();
@@ -279,16 +253,6 @@ public class RegistryService {
// now delete the bucket from the metadata provider, which deletes all flows referencing it
metadataService.deleteBucket(existingBucket);
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postDeleteBucket(existingBucket.getId());
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
return DataModelMapper.map(existingBucket);
} finally {
writeLock.unlock();
@@ -392,17 +356,6 @@ public class RegistryService {
// persist the flow and return the created entity
final FlowEntity createdFlow = metadataService.createFlow(flowEntity);
-
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postCreateFlow(existingBucket.getId(), createdFlow.getId());
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
return DataModelMapper.map(existingBucket, createdFlow);
} finally {
writeLock.unlock();
@@ -544,17 +497,6 @@ public class RegistryService {
// perform the actual update
final FlowEntity updatedFlow = metadataService.updateFlow(existingFlow);
-
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postUpdateFlow(existingFlow.getBucketId(), existingFlow.getId());
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
return DataModelMapper.map(existingBucket, updatedFlow);
} finally {
writeLock.unlock();
@@ -595,16 +537,6 @@ public class RegistryService {
// now delete the flow from the metadata provider
metadataService.deleteFlow(existingFlow);
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postDeleteFlow(existingFlow.getBucketId(), existingFlow.getId());
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
return DataModelMapper.map(existingBucket, existingFlow);
} finally {
writeLock.unlock();
@@ -698,16 +630,6 @@ public class RegistryService {
}
final VersionedFlow updatedVersionedFlow = DataModelMapper.map(existingBucket, updatedFlow);
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postCreateFlowVersion(context);
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
flowSnapshot.setBucket(bucket);
flowSnapshot.setFlow(updatedVersionedFlow);
return flowSnapshot;
@@ -957,17 +879,6 @@ public class RegistryService {
// delete the snapshot itself
metadataService.deleteFlowSnapshot(snapshotEntity);
-
- // call the post-event hook
- for(FlowHookProvider flowHookProvider : flowHookProviders) {
- try {
- flowHookProvider.postDeleteFlowVersion(bucketIdentifier, flowIdentifier, version);
- } catch (Exception e) {
- // we don't want to throw anything here, hook are provided on best effort
- LOGGER.error("Error while calling post-event hook", e);
- }
- }
-
return DataModelMapper.map(existingBucket, snapshotEntity);
} finally {
writeLock.unlock();
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.EventHookProvider
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.EventHookProvider b/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.EventHookProvider
new file mode 100644
index 0000000..2676a35
--- /dev/null
+++ b/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.EventHookProvider
@@ -0,0 +1,16 @@
+# 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.
+org.apache.nifi.registry.provider.hook.ScriptEventHookProvider
+org.apache.nifi.registry.provider.hook.LoggingEventHookProvider
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider b/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider
deleted file mode 100644
index 150103a..0000000
--- a/nifi-registry-framework/src/main/resources/META-INF/services/org.apache.nifi.registry.hook.FlowHookProvider
+++ /dev/null
@@ -1,15 +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.
-org.apache.nifi.registry.provider.hook.ScriptFlowHookProvider
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/main/xsd/providers.xsd
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/main/xsd/providers.xsd b/nifi-registry-framework/src/main/xsd/providers.xsd
index 16086af..ce82dcc 100644
--- a/nifi-registry-framework/src/main/xsd/providers.xsd
+++ b/nifi-registry-framework/src/main/xsd/providers.xsd
@@ -43,7 +43,7 @@
<xs:complexType>
<xs:sequence>
<xs:element name="flowPersistenceProvider" type="Provider" minOccurs="1" maxOccurs="1" />
- <xs:element name="flowHookProvider" type="Provider" minOccurs="0" maxOccurs="unbounded" />
+ <xs:element name="eventHookProvider" type="Provider" minOccurs="0" maxOccurs="unbounded" />
</xs:sequence>
</xs:complexType>
</xs:element>
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventFactory.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventFactory.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventFactory.java
new file mode 100644
index 0000000..95dd6ab
--- /dev/null
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventFactory.java
@@ -0,0 +1,161 @@
+/*
+ * 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.event;
+
+import org.apache.nifi.registry.bucket.Bucket;
+import org.apache.nifi.registry.flow.VersionedFlow;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
+import org.apache.nifi.registry.flow.VersionedProcessGroup;
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventFieldName;
+import org.apache.nifi.registry.hook.EventType;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestEventFactory {
+
+ private Bucket bucket;
+ private VersionedFlow versionedFlow;
+ private VersionedFlowSnapshot versionedFlowSnapshot;
+
+ @Before
+ public void setup() {
+ bucket = new Bucket();
+ bucket.setName("Bucket1");
+ bucket.setIdentifier(UUID.randomUUID().toString());
+ bucket.setCreatedTimestamp(System.currentTimeMillis());
+
+ versionedFlow = new VersionedFlow();
+ versionedFlow.setIdentifier(UUID.randomUUID().toString());
+ versionedFlow.setName("Flow 1");
+ versionedFlow.setBucketIdentifier(bucket.getIdentifier());
+ versionedFlow.setBucketName(bucket.getName());
+
+ VersionedFlowSnapshotMetadata metadata = new VersionedFlowSnapshotMetadata();
+ metadata.setAuthor("user1");
+ metadata.setComments("This is flow 1");
+ metadata.setVersion(1);
+ metadata.setBucketIdentifier(bucket.getIdentifier());
+ metadata.setFlowIdentifier(versionedFlow.getIdentifier());
+
+ versionedFlowSnapshot = new VersionedFlowSnapshot();
+ versionedFlowSnapshot.setSnapshotMetadata(metadata);
+ versionedFlowSnapshot.setFlowContents(new VersionedProcessGroup());
+ }
+
+ @Test
+ public void testBucketCreatedEvent() {
+ final Event event = EventFactory.bucketCreated(bucket);
+ event.validate();
+
+ assertEquals(EventType.CREATE_BUCKET, event.getEventType());
+ assertEquals(2, event.getFields().size());
+
+ assertEquals(bucket.getIdentifier(), event.getField(EventFieldName.BUCKET_ID).getValue());
+ assertEquals("unknown", event.getField(EventFieldName.USER).getValue());
+ }
+
+ @Test
+ public void testBucketUpdatedEvent() {
+ final Event event = EventFactory.bucketUpdated(bucket);
+ event.validate();
+
+ assertEquals(EventType.UPDATE_BUCKET, event.getEventType());
+ assertEquals(2, event.getFields().size());
+
+ assertEquals(bucket.getIdentifier(), event.getField(EventFieldName.BUCKET_ID).getValue());
+ assertEquals("unknown", event.getField(EventFieldName.USER).getValue());
+ }
+
+ @Test
+ public void testBucketDeletedEvent() {
+ final Event event = EventFactory.bucketDeleted(bucket);
+ event.validate();
+
+ assertEquals(EventType.DELETE_BUCKET, event.getEventType());
+ assertEquals(2, event.getFields().size());
+
+ assertEquals(bucket.getIdentifier(), event.getField(EventFieldName.BUCKET_ID).getValue());
+ assertEquals("unknown", event.getField(EventFieldName.USER).getValue());
+ }
+
+ @Test
+ public void testFlowCreated() {
+ final Event event = EventFactory.flowCreated(versionedFlow);
+ event.validate();
+
+ assertEquals(EventType.CREATE_FLOW, event.getEventType());
+ assertEquals(3, event.getFields().size());
+
+ assertEquals(bucket.getIdentifier(), event.getField(EventFieldName.BUCKET_ID).getValue());
+ assertEquals(versionedFlow.getIdentifier(), event.getField(EventFieldName.FLOW_ID).getValue());
+ assertEquals("unknown", event.getField(EventFieldName.USER).getValue());
+ }
+
+ @Test
+ public void testFlowUpdated() {
+ final Event event = EventFactory.flowUpdated(versionedFlow);
+ event.validate();
+
+ assertEquals(EventType.UPDATE_FLOW, event.getEventType());
+ assertEquals(3, event.getFields().size());
+
+ assertEquals(bucket.getIdentifier(), event.getField(EventFieldName.BUCKET_ID).getValue());
+ assertEquals(versionedFlow.getIdentifier(), event.getField(EventFieldName.FLOW_ID).getValue());
+ assertEquals("unknown", event.getField(EventFieldName.USER).getValue());
+ }
+
+ @Test
+ public void testFlowDeleted() {
+ final Event event = EventFactory.flowDeleted(versionedFlow);
+ event.validate();
+
+ assertEquals(EventType.DELETE_FLOW, event.getEventType());
+ assertEquals(3, event.getFields().size());
+
+ assertEquals(bucket.getIdentifier(), event.getField(EventFieldName.BUCKET_ID).getValue());
+ assertEquals(versionedFlow.getIdentifier(), event.getField(EventFieldName.FLOW_ID).getValue());
+ assertEquals("unknown", event.getField(EventFieldName.USER).getValue());
+ }
+
+ @Test
+ public void testFlowVersionedCreated() {
+ final Event event = EventFactory.flowVersionCreated(versionedFlowSnapshot);
+ event.validate();
+
+ assertEquals(EventType.CREATE_FLOW_VERSION, event.getEventType());
+ assertEquals(5, event.getFields().size());
+
+ assertEquals(bucket.getIdentifier(), event.getField(EventFieldName.BUCKET_ID).getValue());
+ assertEquals(versionedFlow.getIdentifier(), event.getField(EventFieldName.FLOW_ID).getValue());
+
+ assertEquals(String.valueOf(versionedFlowSnapshot.getSnapshotMetadata().getVersion()),
+ event.getField(EventFieldName.VERSION).getValue());
+
+ assertEquals(versionedFlowSnapshot.getSnapshotMetadata().getAuthor(),
+ event.getField(EventFieldName.USER).getValue());
+
+ assertEquals(versionedFlowSnapshot.getSnapshotMetadata().getComments(),
+ event.getField(EventFieldName.COMMENT).getValue());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventService.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventService.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventService.java
new file mode 100644
index 0000000..0270dd8
--- /dev/null
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestEventService.java
@@ -0,0 +1,97 @@
+/*
+ * 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.event;
+
+import org.apache.nifi.registry.bucket.Bucket;
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventHookException;
+import org.apache.nifi.registry.hook.EventHookProvider;
+import org.apache.nifi.registry.provider.ProviderConfigurationContext;
+import org.apache.nifi.registry.provider.ProviderCreationException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+public class TestEventService {
+
+ private CapturingEventHook eventHook;
+ private EventService eventService;
+
+ @Before
+ public void setup() {
+ eventHook = new CapturingEventHook();
+ eventService = new EventService(Collections.singletonList(eventHook));
+ eventService.postConstruct();
+ }
+
+ @After
+ public void teardown() throws Exception {
+ eventService.destroy();
+ }
+
+ @Test
+ public void testPublishConsume() throws InterruptedException {
+ final Bucket bucket = new Bucket();
+ bucket.setIdentifier(UUID.randomUUID().toString());
+
+ final Event bucketCreatedEvent = EventFactory.bucketCreated(bucket);
+ eventService.publish(bucketCreatedEvent);
+
+ final Event bucketDeletedEvent = EventFactory.bucketDeleted(bucket);
+ eventService.publish(bucketDeletedEvent);
+
+ Thread.sleep(1000);
+
+ final List<Event> events = eventHook.getEvents();
+ Assert.assertEquals(2, events.size());
+
+ final Event firstEvent = events.get(0);
+ Assert.assertEquals(bucketCreatedEvent.getEventType(), firstEvent.getEventType());
+
+ final Event secondEvent = events.get(1);
+ Assert.assertEquals(bucketDeletedEvent.getEventType(), secondEvent.getEventType());
+ }
+
+ /**
+ * Simple implementation of EventHookProvider that captures event for later verification.
+ */
+ private class CapturingEventHook implements EventHookProvider {
+
+ private List<Event> events = new ArrayList<>();
+
+ @Override
+ public void onConfigured(ProviderConfigurationContext configurationContext) throws ProviderCreationException {
+
+ }
+
+ @Override
+ public void handle(Event event) throws EventHookException {
+ events.add(event);
+ }
+
+ public List<Event> getEvents() {
+ return events;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestStandardEvent.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestStandardEvent.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestStandardEvent.java
new file mode 100644
index 0000000..8e9f73f
--- /dev/null
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/event/TestStandardEvent.java
@@ -0,0 +1,47 @@
+/*
+ * 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.event;
+
+import org.apache.nifi.registry.hook.Event;
+import org.apache.nifi.registry.hook.EventField;
+import org.apache.nifi.registry.hook.EventFieldName;
+import org.apache.nifi.registry.hook.EventType;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestStandardEvent {
+
+ @Test(expected = IllegalStateException.class)
+ public void testInvalidEvent() {
+ final Event event = new StandardEvent.Builder()
+ .eventType(EventType.CREATE_BUCKET)
+ .build();
+
+ event.validate();
+ }
+
+ @Test
+ public void testGetFieldWhenDoesNotExist() {
+ final Event event = new StandardEvent.Builder()
+ .eventType(EventType.CREATE_BUCKET)
+ .build();
+
+ final EventField field = event.getField(EventFieldName.BUCKET_ID);
+ Assert.assertNull(field);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptEventHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptEventHookProvider.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptEventHookProvider.java
new file mode 100644
index 0000000..ab24998
--- /dev/null
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptEventHookProvider.java
@@ -0,0 +1,45 @@
+/*
+ * 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.provider.hook;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import org.apache.nifi.registry.extension.ExtensionManager;
+import org.apache.nifi.registry.properties.NiFiRegistryProperties;
+import org.apache.nifi.registry.provider.ProviderCreationException;
+import org.apache.nifi.registry.provider.ProviderFactory;
+import org.apache.nifi.registry.provider.StandardProviderFactory;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class TestScriptEventHookProvider {
+
+ @Test(expected = ProviderCreationException.class)
+ public void testBadScriptProvider() {
+ final NiFiRegistryProperties props = new NiFiRegistryProperties();
+ props.setProperty(NiFiRegistryProperties.PROVIDERS_CONFIGURATION_FILE, "src/test/resources/provider/hook/bad-script-provider.xml");
+
+ final ExtensionManager extensionManager = Mockito.mock(ExtensionManager.class);
+ when(extensionManager.getExtensionClassLoader(any(String.class))).thenReturn(this.getClass().getClassLoader());
+
+ final ProviderFactory providerFactory = new StandardProviderFactory(props, extensionManager);
+ providerFactory.initialize();
+ providerFactory.getEventHookProviders();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java
deleted file mode 100644
index 5bc888c..0000000
--- a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/hook/TestScriptFlowHookProvider.java
+++ /dev/null
@@ -1,45 +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.provider.hook;
-
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.when;
-
-import org.apache.nifi.registry.extension.ExtensionManager;
-import org.apache.nifi.registry.properties.NiFiRegistryProperties;
-import org.apache.nifi.registry.provider.ProviderCreationException;
-import org.apache.nifi.registry.provider.ProviderFactory;
-import org.apache.nifi.registry.provider.StandardProviderFactory;
-import org.junit.Test;
-import org.mockito.Mockito;
-
-public class TestScriptFlowHookProvider {
-
- @Test(expected = ProviderCreationException.class)
- public void testBadScriptProvider() {
- final NiFiRegistryProperties props = new NiFiRegistryProperties();
- props.setProperty(NiFiRegistryProperties.PROVIDERS_CONFIGURATION_FILE, "src/test/resources/provider/hook/bad-script-provider.xml");
-
- final ExtensionManager extensionManager = Mockito.mock(ExtensionManager.class);
- when(extensionManager.getExtensionClassLoader(any(String.class))).thenReturn(this.getClass().getClassLoader());
-
- final ProviderFactory providerFactory = new StandardProviderFactory(props, extensionManager);
- providerFactory.initialize();
- providerFactory.getFlowHookProviders();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
index 1f21248..95e2d1a 100644
--- a/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
+++ b/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
@@ -30,7 +30,6 @@ import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
import org.apache.nifi.registry.flow.VersionedProcessGroup;
import org.apache.nifi.registry.flow.VersionedProcessor;
-import org.apache.nifi.registry.hook.FlowHookProvider;
import org.apache.nifi.registry.serialization.Serializer;
import org.apache.nifi.registry.serialization.VersionedProcessGroupSerializer;
import org.junit.Assert;
@@ -73,7 +72,6 @@ public class TestRegistryService {
private MetadataService metadataService;
private FlowPersistenceProvider flowPersistenceProvider;
- private FlowHookProvider flowHookProvider;
private Serializer<VersionedProcessGroup> snapshotSerializer;
private Validator validator;
@@ -83,13 +81,12 @@ public class TestRegistryService {
public void setup() {
metadataService = mock(MetadataService.class);
flowPersistenceProvider = mock(FlowPersistenceProvider.class);
- // flowHookProvider = mock(FlowHookProvider.class);
snapshotSerializer = mock(VersionedProcessGroupSerializer.class);
final ValidatorFactory validatorFactory = Validation.buildDefaultValidatorFactory();
validator = validatorFactory.getValidator();
- registryService = new RegistryService(metadataService, flowPersistenceProvider, new ArrayList<FlowHookProvider>(), snapshotSerializer, validator);
+ registryService = new RegistryService(metadataService, flowPersistenceProvider, snapshotSerializer, validator);
}
// ---------------------- Test Bucket methods ---------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml
----------------------------------------------------------------------
diff --git a/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml b/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml
index ca6cc6c..568e756 100644
--- a/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml
+++ b/nifi-registry-framework/src/test/resources/provider/hook/bad-script-provider.xml
@@ -21,10 +21,10 @@
<property name="Flow Property 2">flow bar</property>
</flowPersistenceProvider>
- <flowHookProvider>
- <class>org.apache.nifi.registry.provider.hook.ScriptFlowHookProvider</class>
+ <eventHookProvider>
+ <class>org.apache.nifi.registry.provider.hook.ScriptEventHookProvider</class>
<property name="Script Path"></property>
<property name="Working Directory"></property>
- </flowHookProvider>
+ </eventHookProvider>
</providers>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/Event.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/Event.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/Event.java
new file mode 100644
index 0000000..2948962
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/Event.java
@@ -0,0 +1,50 @@
+/*
+ * 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.hook;
+
+import java.util.List;
+
+/**
+ * An event that will be passed to EventHookProviders.
+ */
+public interface Event {
+
+ /**
+ * @return the type of the event
+ */
+ EventType getEventType();
+
+ /**
+ * @return the fields of the event in the order they were added to the event
+ */
+ List<EventField> getFields();
+
+ /**
+ * @param fieldName the name of the field to return
+ * @return the EventField with the given name, or null if it does not exist
+ */
+ EventField getField(EventFieldName fieldName);
+
+ /**
+ * Will be called before publishing the event to ensure the event contains the required
+ * fields for the given event type in the order specified by the type.
+ *
+ * @throws IllegalStateException if the event does not contain the required fields
+ */
+ void validate() throws IllegalStateException;
+
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventField.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventField.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventField.java
new file mode 100644
index 0000000..4859266
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventField.java
@@ -0,0 +1,33 @@
+/*
+ * 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.hook;
+
+/**
+ * A field for an event.
+ */
+public interface EventField {
+
+ /**
+ * @return the name of the field
+ */
+ EventFieldName getName();
+
+ /**
+ * @return the value of the field
+ */
+ String getValue();
+}
http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/51b6c78b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventFieldName.java
----------------------------------------------------------------------
diff --git a/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventFieldName.java b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventFieldName.java
new file mode 100644
index 0000000..35b0cfe
--- /dev/null
+++ b/nifi-registry-provider-api/src/main/java/org/apache/nifi/registry/hook/EventFieldName.java
@@ -0,0 +1,30 @@
+/*
+ * 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.hook;
+
+/**
+ * Enumeration of possible field names for an EventField.
+ */
+public enum EventFieldName {
+
+ BUCKET_ID,
+ FLOW_ID,
+ VERSION,
+ USER,
+ COMMENT;
+
+}