You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2015/08/16 21:58:26 UTC
[10/53] [abbrv] [partial] nifi git commit: NIFI-850 removed nifi
parent, updated nifi pom, moved all nifi subdirs up one level, fixed readme.
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
new file mode 100644
index 0000000..d73a09b
--- /dev/null
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
@@ -0,0 +1,135 @@
+/*
+ * 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.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.nifi.attribute.expression.language.Query;
+import org.apache.nifi.attribute.expression.language.Query.Range;
+import org.apache.nifi.attribute.expression.language.StandardExpressionLanguageCompiler;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.controller.ControllerServiceLookup;
+import org.apache.nifi.expression.ExpressionLanguageCompiler;
+
+public class MockValidationContext implements ValidationContext, ControllerServiceLookup {
+
+ private final MockProcessContext context;
+ private final Map<String, Boolean> expressionLanguageSupported;
+
+ public MockValidationContext(final MockProcessContext processContext) {
+ this.context = processContext;
+
+ final Map<PropertyDescriptor, String> properties = processContext.getProperties();
+ expressionLanguageSupported = new HashMap<>(properties.size());
+ for (final PropertyDescriptor descriptor : properties.keySet()) {
+ expressionLanguageSupported.put(descriptor.getName(), descriptor.isExpressionLanguageSupported());
+ }
+ }
+
+ @Override
+ public ControllerService getControllerService(final String identifier) {
+ return context.getControllerService(identifier);
+ }
+
+ @Override
+ public PropertyValue newPropertyValue(final String rawValue) {
+ return new MockPropertyValue(rawValue, this);
+ }
+
+ @Override
+ public ExpressionLanguageCompiler newExpressionLanguageCompiler() {
+ return new StandardExpressionLanguageCompiler();
+ }
+
+ @Override
+ public ValidationContext getControllerServiceValidationContext(final ControllerService controllerService) {
+ final MockProcessContext serviceProcessContext = new MockProcessContext(controllerService, context);
+ return new MockValidationContext(serviceProcessContext);
+ }
+
+ @Override
+ public PropertyValue getProperty(final PropertyDescriptor property) {
+ return context.getProperty(property);
+ }
+
+ @Override
+ public Map<PropertyDescriptor, String> getProperties() {
+ return context.getProperties();
+ }
+
+ @Override
+ public String getAnnotationData() {
+ return context.getAnnotationData();
+ }
+
+ @Override
+ public Set<String> getControllerServiceIdentifiers(Class<? extends ControllerService> serviceType) {
+ return context.getControllerServiceIdentifiers(serviceType);
+ }
+
+ @Override
+ public ControllerServiceLookup getControllerServiceLookup() {
+ return this;
+ }
+
+ @Override
+ public boolean isControllerServiceEnabled(final String serviceIdentifier) {
+ return context.isControllerServiceEnabled(serviceIdentifier);
+ }
+
+ @Override
+ public boolean isControllerServiceEnabled(final ControllerService service) {
+ return context.isControllerServiceEnabled(service);
+ }
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ final ControllerServiceConfiguration configuration = context.getConfiguration(serviceIdentifier);
+ return configuration == null ? null : serviceIdentifier;
+ }
+
+ @Override
+ public boolean isValidationRequired(final ControllerService service) {
+ return true;
+ }
+
+ @Override
+ public boolean isControllerServiceEnabling(String serviceIdentifier) {
+ return context.isControllerServiceEnabling(serviceIdentifier);
+ }
+
+ public boolean isExpressionLanguagePresent(final String value) {
+ if (value == null) {
+ return false;
+ }
+
+ final List<Range> elRanges = Query.extractExpressionRanges(value);
+ return (elRanges != null && !elRanges.isEmpty());
+ }
+
+ @Override
+ public boolean isExpressionLanguageSupported(final String propertyName) {
+ final Boolean supported = expressionLanguageSupported.get(propertyName);
+ return Boolean.TRUE.equals(supported);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/main/java/org/apache/nifi/util/ReflectionUtils.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/ReflectionUtils.java b/nifi-mock/src/main/java/org/apache/nifi/util/ReflectionUtils.java
new file mode 100644
index 0000000..940eeea
--- /dev/null
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/ReflectionUtils.java
@@ -0,0 +1,150 @@
+/*
+ * 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.util;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ReflectionUtils {
+
+ private final static Logger LOG = LoggerFactory.getLogger(ReflectionUtils.class);
+
+ /**
+ * Invokes all methods on the given instance that have been annotated with
+ * the given Annotation. If the signature of the method that is defined in
+ * <code>instance</code> uses 1 or more parameters, those parameters must be
+ * specified by the <code>args</code> parameter. However, if more arguments
+ * are supplied by the <code>args</code> parameter than needed, the extra
+ * arguments will be ignored.
+ *
+ * @param annotation the annotation to look for
+ * @param instance to invoke a method of
+ * @param args to supply in a method call
+ * @throws InvocationTargetException ite
+ * @throws IllegalArgumentException iae
+ * @throws IllegalAccessException if not allowed to invoke that method
+ */
+ public static void invokeMethodsWithAnnotation(final Class<? extends Annotation> annotation, final Object instance, final Object... args)
+ throws IllegalAccessException, IllegalArgumentException, InvocationTargetException {
+ for (final Method method : instance.getClass().getMethods()) {
+ if (method.isAnnotationPresent(annotation)) {
+ final boolean isAccessible = method.isAccessible();
+ method.setAccessible(true);
+
+ try {
+ final Class<?>[] argumentTypes = method.getParameterTypes();
+ if (argumentTypes.length > args.length) {
+ throw new IllegalArgumentException(String.format("Unable to invoke method %1$s on %2$s because method expects %3$s parameters but only %4$s were given",
+ method.getName(), instance, argumentTypes.length, args.length));
+ }
+
+ for (int i = 0; i < argumentTypes.length; i++) {
+ final Class<?> argType = argumentTypes[i];
+ if (!argType.isAssignableFrom(args[i].getClass())) {
+ throw new IllegalArgumentException(String.format(
+ "Unable to invoke method %1$s on %2$s because method parameter %3$s is expected to be of type %4$s but argument passed was of type %5$s",
+ method.getName(), instance, i, argType, args[i].getClass()));
+ }
+ }
+
+ if (argumentTypes.length == args.length) {
+ method.invoke(instance, args);
+ } else {
+ final Object[] argsToPass = new Object[argumentTypes.length];
+ for (int i = 0; i < argsToPass.length; i++) {
+ argsToPass[i] = args[i];
+ }
+
+ method.invoke(instance, argsToPass);
+ }
+ } finally {
+ if (!isAccessible) {
+ method.setAccessible(false);
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Invokes all methods on the given instance that have been annotated with
+ * the given Annotation. If the signature of the method that is defined in
+ * <code>instance</code> uses 1 or more parameters, those parameters must be
+ * specified by the <code>args</code> parameter. However, if more arguments
+ * are supplied by the <code>args</code> parameter than needed, the extra
+ * arguments will be ignored.
+ *
+ * @param annotation the annotation to look for
+ * @param instance to invoke a method of
+ * @param args to supply in a method call
+ * @return <code>true</code> if all appropriate methods were invoked and
+ * returned without throwing an Exception, <code>false</code> if one of the
+ * methods threw an Exception or could not be invoked; if <code>false</code>
+ * is returned, an error will have been logged.
+ */
+ public static boolean quietlyInvokeMethodsWithAnnotation(final Class<? extends Annotation> annotation, final Object instance, final Object... args) {
+ for (final Method method : instance.getClass().getMethods()) {
+ if (method.isAnnotationPresent(annotation)) {
+ final boolean isAccessible = method.isAccessible();
+ method.setAccessible(true);
+
+ try {
+ final Class<?>[] argumentTypes = method.getParameterTypes();
+ if (argumentTypes.length > args.length) {
+ LOG.error("Unable to invoke method {} on {} because method expects {} parameters but only {} were given",
+ new Object[]{method.getName(), instance, argumentTypes.length, args.length});
+ return false;
+ }
+
+ for (int i = 0; i < argumentTypes.length; i++) {
+ final Class<?> argType = argumentTypes[i];
+ if (!argType.isAssignableFrom(args[i].getClass())) {
+ LOG.error("Unable to invoke method {} on {} because method parameter {} is expected to be of type {} but argument passed was of type {}",
+ new Object[]{method.getName(), instance, i, argType, args[i].getClass()});
+ return false;
+ }
+ }
+
+ try {
+ if (argumentTypes.length == args.length) {
+ method.invoke(instance, args);
+ } else {
+ final Object[] argsToPass = new Object[argumentTypes.length];
+ for (int i = 0; i < argsToPass.length; i++) {
+ argsToPass[i] = args[i];
+ }
+
+ method.invoke(instance, argsToPass);
+ }
+ } catch (final IllegalAccessException | IllegalArgumentException | InvocationTargetException t) {
+ LOG.error("Unable to invoke method {} on {} due to {}", new Object[]{method.getName(), instance, t});
+ LOG.error("", t);
+ return false;
+ }
+ } finally {
+ if (!isAccessible) {
+ method.setAccessible(false);
+ }
+ }
+ }
+ }
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java b/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java
new file mode 100644
index 0000000..994735b
--- /dev/null
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java
@@ -0,0 +1,91 @@
+/*
+ * 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.util;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceReporter;
+
+public class SharedSessionState {
+
+ private final MockFlowFileQueue flowFileQueue;
+ private final ProvenanceReporter provenanceReporter;
+ @SuppressWarnings("unused")
+ private final Processor processor;
+ private final AtomicLong flowFileIdGenerator;
+ private final ConcurrentMap<String, AtomicLong> counterMap = new ConcurrentHashMap<>();
+ private final Set<ProvenanceEventRecord> events = new LinkedHashSet<>();
+
+ public SharedSessionState(final Processor processor, final AtomicLong flowFileIdGenerator) {
+ flowFileQueue = new MockFlowFileQueue();
+ provenanceReporter = new MockProvenanceReporter(null, this, UUID.randomUUID().toString(), "N/A");
+ this.flowFileIdGenerator = flowFileIdGenerator;
+ this.processor = processor;
+ }
+
+ void addProvenanceEvents(final Collection<ProvenanceEventRecord> events) {
+ this.events.addAll(events);
+ }
+
+ void clearProvenanceEvents() {
+ this.events.clear();
+ }
+
+ public List<ProvenanceEventRecord> getProvenanceEvents() {
+ return new ArrayList<>(this.events);
+ }
+
+ public MockFlowFileQueue getFlowFileQueue() {
+ return flowFileQueue;
+ }
+
+ public ProvenanceReporter getProvenanceReporter() {
+ return provenanceReporter;
+ }
+
+ public long nextFlowFileId() {
+ return flowFileIdGenerator.getAndIncrement();
+ }
+
+ public void adjustCounter(final String name, final long delta) {
+ AtomicLong counter = counterMap.get(name);
+ if (counter == null) {
+ counter = new AtomicLong(0L);
+ final AtomicLong existingCounter = counterMap.putIfAbsent(name, counter);
+ if (existingCounter != null) {
+ counter = existingCounter;
+ }
+ }
+
+ counter.addAndGet(delta);
+ }
+
+ public Long getCounterValue(final String name) {
+ final AtomicLong counterValue = counterMap.get(name);
+ return counterValue == null ? null : counterValue.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/main/java/org/apache/nifi/util/SingleSessionFactory.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/SingleSessionFactory.java b/nifi-mock/src/main/java/org/apache/nifi/util/SingleSessionFactory.java
new file mode 100644
index 0000000..ef518dc
--- /dev/null
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/SingleSessionFactory.java
@@ -0,0 +1,35 @@
+/*
+ * 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.util;
+
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+
+public class SingleSessionFactory implements ProcessSessionFactory {
+
+ private final MockProcessSession session;
+
+ public SingleSessionFactory(final MockProcessSession session) {
+ this.session = session;
+ }
+
+ @Override
+ public ProcessSession createSession() {
+ return session;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
new file mode 100644
index 0000000..048e2b9
--- /dev/null
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
@@ -0,0 +1,728 @@
+/*
+ * 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.util;
+
+import static java.util.Objects.requireNonNull;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.lifecycle.OnAdded;
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnRemoved;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnShutdown;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processor.QueueSize;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceReporter;
+import org.apache.nifi.reporting.InitializationException;
+import org.junit.Assert;
+
+public class StandardProcessorTestRunner implements TestRunner {
+
+ private final Processor processor;
+ private final MockProcessContext context;
+ private final MockFlowFileQueue flowFileQueue;
+ private final MockSessionFactory sessionFactory;
+ private final SharedSessionState sharedState;
+ private final AtomicLong idGenerator;
+ private final boolean triggerSerially;
+
+ private int numThreads = 1;
+ private final AtomicInteger invocations = new AtomicInteger(0);
+
+ private static final Set<Class<? extends Annotation>> deprecatedTypeAnnotations = new HashSet<>();
+ private static final Set<Class<? extends Annotation>> deprecatedMethodAnnotations = new HashSet<>();
+
+ static {
+ // do this in a separate method, just so that we can add a @SuppressWarnings annotation
+ // because we want to indicate explicitly that we know that we are using deprecated
+ // classes here.
+ populateDeprecatedMethods();
+ }
+
+ StandardProcessorTestRunner(final Processor processor) {
+ this.processor = processor;
+ this.idGenerator = new AtomicLong(0L);
+ this.sharedState = new SharedSessionState(processor, idGenerator);
+ this.flowFileQueue = sharedState.getFlowFileQueue();
+ this.sessionFactory = new MockSessionFactory(sharedState, processor);
+ this.context = new MockProcessContext(processor);
+
+ detectDeprecatedAnnotations(processor);
+
+ final MockProcessorInitializationContext mockInitContext = new MockProcessorInitializationContext(processor, context);
+ processor.initialize(mockInitContext);
+
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnAdded.class, processor);
+ } catch (final Exception e) {
+ Assert.fail("Could not invoke methods annotated with @OnAdded annotation due to: " + e);
+ }
+
+ triggerSerially = null != processor.getClass().getAnnotation(TriggerSerially.class);
+ }
+
+ @SuppressWarnings("deprecation")
+ private static void populateDeprecatedMethods() {
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.CapabilityDescription.class);
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.EventDriven.class);
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.SideEffectFree.class);
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.SupportsBatching.class);
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.Tags.class);
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.TriggerWhenEmpty.class);
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.TriggerWhenAnyDestinationAvailable.class);
+ deprecatedTypeAnnotations.add(org.apache.nifi.processor.annotation.TriggerSerially.class);
+
+ deprecatedMethodAnnotations.add(org.apache.nifi.processor.annotation.OnRemoved.class);
+ deprecatedMethodAnnotations.add(org.apache.nifi.processor.annotation.OnAdded.class);
+ deprecatedMethodAnnotations.add(org.apache.nifi.processor.annotation.OnScheduled.class);
+ deprecatedMethodAnnotations.add(org.apache.nifi.processor.annotation.OnShutdown.class);
+ deprecatedMethodAnnotations.add(org.apache.nifi.processor.annotation.OnStopped.class);
+ deprecatedMethodAnnotations.add(org.apache.nifi.processor.annotation.OnUnscheduled.class);
+ }
+
+ private static void detectDeprecatedAnnotations(final Processor processor) {
+ for (final Class<? extends Annotation> annotationClass : deprecatedTypeAnnotations) {
+ if (processor.getClass().isAnnotationPresent(annotationClass)) {
+ Assert.fail("Processor is using deprecated Annotation " + annotationClass.getCanonicalName());
+ }
+ }
+
+ for (final Class<? extends Annotation> annotationClass : deprecatedMethodAnnotations) {
+ for (final Method method : processor.getClass().getMethods()) {
+ if (method.isAnnotationPresent(annotationClass)) {
+ Assert.fail("Processor is using deprecated Annotation " + annotationClass.getCanonicalName() + " for method " + method);
+ }
+ }
+ }
+
+ }
+
+ @Override
+ public void setValidateExpressionUsage(final boolean validate) {
+ context.setValidateExpressionUsage(validate);
+ }
+
+ @Override
+ public Processor getProcessor() {
+ return processor;
+ }
+
+ @Override
+ public MockProcessContext getProcessContext() {
+ return context;
+ }
+
+ @Override
+ public void run() {
+ run(1);
+ }
+
+ @Override
+ public void run(int iterations) {
+ run(iterations, true);
+ }
+
+ @Override
+ public void run(final int iterations, final boolean stopOnFinish) {
+ run(iterations, stopOnFinish, true);
+ }
+
+ @Override
+ public void run(final int iterations, final boolean stopOnFinish, final boolean initialize) {
+ run(iterations, stopOnFinish, initialize, 5000);
+ }
+
+ @Override
+ public void run(final int iterations, final boolean stopOnFinish, final boolean initialize, final long runWait) {
+ if (iterations < 1) {
+ throw new IllegalArgumentException();
+ }
+
+ context.assertValid();
+ context.enableExpressionValidation();
+ try {
+ if (initialize) {
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor, context);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ Assert.fail("Could not invoke methods annotated with @OnScheduled annotation due to: " + e);
+ }
+ }
+
+ final ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
+ @SuppressWarnings("unchecked")
+ final Future<Throwable>[] futures = new Future[iterations];
+ for (int i = 0; i < iterations; i++) {
+ final Future<Throwable> future = executorService.submit(new RunProcessor());
+ futures[i] = future;
+ }
+
+ executorService.shutdown();
+ try {
+ executorService.awaitTermination(runWait, TimeUnit.MILLISECONDS);
+ } catch (final InterruptedException e1) {
+ }
+
+ int finishedCount = 0;
+ boolean unscheduledRun = false;
+ for (final Future<Throwable> future : futures) {
+ try {
+ final Throwable thrown = future.get(); // wait for the result
+ if (thrown != null) {
+ throw new AssertionError(thrown);
+ }
+
+ if (++finishedCount == 1) {
+ unscheduledRun = true;
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnUnscheduled.class, processor, context);
+ } catch (final Exception e) {
+ Assert.fail("Could not invoke methods annotated with @OnUnscheduled annotation due to: " + e);
+ }
+ }
+ } catch (final Exception e) {
+ }
+ }
+
+ if (!unscheduledRun) {
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnUnscheduled.class, processor, context);
+ } catch (final Exception e) {
+ Assert.fail("Could not invoke methods annotated with @OnUnscheduled annotation due to: " + e);
+ }
+ }
+
+ if (stopOnFinish) {
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnStopped.class, processor);
+ } catch (final Exception e) {
+ Assert.fail("Could not invoke methods annotated with @OnStopped annotation due to: " + e);
+ }
+ }
+ } finally {
+ context.disableExpressionValidation();
+ }
+ }
+
+ @Override
+ public void shutdown() {
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnShutdown.class, processor);
+ } catch (final Exception e) {
+ Assert.fail("Could not invoke methods annotated with @OnShutdown annotation due to: " + e);
+ }
+ }
+
+ private class RunProcessor implements Callable<Throwable> {
+
+ @Override
+ public Throwable call() throws Exception {
+ invocations.incrementAndGet();
+ try {
+ processor.onTrigger(context, sessionFactory);
+ } catch (final Throwable t) {
+ return t;
+ }
+
+ return null;
+ }
+ }
+
+ @Override
+ public ProcessSessionFactory getProcessSessionFactory() {
+ return sessionFactory;
+ }
+
+ @Override
+ public void assertAllFlowFilesTransferred(final String relationship) {
+ for (final MockProcessSession session : sessionFactory.getCreatedSessions()) {
+ session.assertAllFlowFilesTransferred(relationship);
+ }
+ }
+
+ @Override
+ public void assertAllFlowFilesTransferred(final Relationship relationship) {
+ for (final MockProcessSession session : sessionFactory.getCreatedSessions()) {
+ session.assertAllFlowFilesTransferred(relationship);
+ }
+ }
+
+ @Override
+ public void assertAllFlowFilesTransferred(final String relationship, final int count) {
+ assertAllFlowFilesTransferred(relationship);
+ assertTransferCount(relationship, count);
+ }
+
+ @Override
+ public void assertAllFlowFilesTransferred(final Relationship relationship, final int count) {
+ assertAllFlowFilesTransferred(relationship);
+ assertTransferCount(relationship, count);
+ }
+
+ @Override
+ public void assertTransferCount(final Relationship relationship, final int count) {
+ Assert.assertEquals(count, getFlowFilesForRelationship(relationship).size());
+ }
+
+ @Override
+ public void assertTransferCount(final String relationship, final int count) {
+ Assert.assertEquals(count, getFlowFilesForRelationship(relationship).size());
+ }
+
+ @Override
+ public void assertValid() {
+ context.assertValid();
+ }
+
+ @Override
+ public void assertNotValid() {
+ Assert.assertFalse("Processor appears to be valid but expected it to be invalid", context.isValid());
+ }
+
+ @Override
+ public boolean isQueueEmpty() {
+ return flowFileQueue.isEmpty();
+ }
+
+ @Override
+ public void assertQueueEmpty() {
+ Assert.assertTrue(flowFileQueue.isEmpty());
+ }
+
+ @Override
+ public void assertQueueNotEmpty() {
+ Assert.assertFalse(flowFileQueue.isEmpty());
+ }
+
+ @Override
+ public void clearTransferState() {
+ for (final MockProcessSession session : sessionFactory.getCreatedSessions()) {
+ session.clearTransferState();
+ }
+ }
+
+ @Override
+ public void enqueue(final FlowFile... flowFiles) {
+ for (final FlowFile flowFile : flowFiles) {
+ flowFileQueue.offer((MockFlowFile) flowFile);
+ }
+ }
+
+ @Override
+ public void enqueue(final Path path) throws IOException {
+ enqueue(path, new HashMap<String, String>());
+ }
+
+ @Override
+ public void enqueue(final Path path, final Map<String, String> attributes) throws IOException {
+ final Map<String, String> modifiedAttributes = new HashMap<>(attributes);
+ if (!modifiedAttributes.containsKey(CoreAttributes.FILENAME.key())) {
+ modifiedAttributes.put(CoreAttributes.FILENAME.key(), path.toFile().getName());
+ }
+ try (final InputStream in = Files.newInputStream(path)) {
+ enqueue(in, modifiedAttributes);
+ }
+ }
+
+ @Override
+ public void enqueue(final byte[] data) {
+ enqueue(data, new HashMap<String, String>());
+ }
+
+ @Override
+ public void enqueue(final byte[] data, final Map<String, String> attributes) {
+ enqueue(new ByteArrayInputStream(data), attributes);
+ }
+
+ @Override
+ public void enqueue(final InputStream data) {
+ enqueue(data, new HashMap<String, String>());
+ }
+
+ @Override
+ public void enqueue(final InputStream data, final Map<String, String> attributes) {
+ final MockProcessSession session = new MockProcessSession(new SharedSessionState(processor, idGenerator), processor);
+ MockFlowFile flowFile = session.create();
+ flowFile = session.importFrom(data, flowFile);
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ enqueue(flowFile);
+ }
+
+ @Override
+ public byte[] getContentAsByteArray(final MockFlowFile flowFile) {
+ return flowFile.getData();
+ }
+
+ @Override
+ public List<MockFlowFile> getFlowFilesForRelationship(final String relationship) {
+ final Relationship rel = new Relationship.Builder().name(relationship).build();
+ return getFlowFilesForRelationship(rel);
+ }
+
+ @Override
+ public List<MockFlowFile> getFlowFilesForRelationship(final Relationship relationship) {
+ final List<MockFlowFile> flowFiles = new ArrayList<>();
+ for (final MockProcessSession session : sessionFactory.getCreatedSessions()) {
+ flowFiles.addAll(session.getFlowFilesForRelationship(relationship));
+ }
+
+ Collections.sort(flowFiles, new Comparator<MockFlowFile>() {
+ @Override
+ public int compare(final MockFlowFile o1, final MockFlowFile o2) {
+ return Long.compare(o1.getCreationTime(), o2.getCreationTime());
+ }
+ });
+
+ return flowFiles;
+ }
+
+ /**
+ * @deprecated The ProvenanceReporter should not be accessed through the test runner, as it does not expose the events that were emitted.
+ */
+ @Override
+ @Deprecated
+ public ProvenanceReporter getProvenanceReporter() {
+ return sharedState.getProvenanceReporter();
+ }
+
+ @Override
+ public QueueSize getQueueSize() {
+ return flowFileQueue.size();
+ }
+
+ @Override
+ public Long getCounterValue(final String name) {
+ return sharedState.getCounterValue(name);
+ }
+
+ @Override
+ public int getRemovedCount() {
+ int count = 0;
+ for (final MockProcessSession session : sessionFactory.getCreatedSessions()) {
+ count += session.getRemovedCount();
+ }
+
+ return count;
+ }
+
+ @Override
+ public void setAnnotationData(final String annotationData) {
+ context.setAnnotationData(annotationData);
+ }
+
+ @Override
+ public ValidationResult setProperty(final String propertyName, final String propertyValue) {
+ return context.setProperty(propertyName, propertyValue);
+ }
+
+ @Override
+ public ValidationResult setProperty(final PropertyDescriptor descriptor, final String value) {
+ return context.setProperty(descriptor, value);
+ }
+
+ @Override
+ public ValidationResult setProperty(final PropertyDescriptor descriptor, final AllowableValue value) {
+ return context.setProperty(descriptor, value.getValue());
+ }
+
+ @Override
+ public void setThreadCount(final int threadCount) {
+ if (threadCount > 1 && triggerSerially) {
+ Assert.fail("Cannot set thread-count higher than 1 because the processor is triggered serially");
+ }
+
+ this.numThreads = threadCount;
+ }
+
+ @Override
+ public int getThreadCount() {
+ return numThreads;
+ }
+
+ @Override
+ public void setRelationshipAvailable(final Relationship relationship) {
+ final Set<Relationship> unavailable = new HashSet<>(context.getUnavailableRelationships());
+ unavailable.remove(relationship);
+ context.setUnavailableRelationships(unavailable);
+ }
+
+ @Override
+ public void setRelationshipAvailable(final String relationshipName) {
+ setRelationshipAvailable(new Relationship.Builder().name(relationshipName).build());
+ }
+
+ @Override
+ public void setRelationshipUnavailable(final Relationship relationship) {
+ final Set<Relationship> unavailable = new HashSet<>(context.getUnavailableRelationships());
+ unavailable.add(relationship);
+ context.setUnavailableRelationships(unavailable);
+ }
+
+ @Override
+ public void setRelationshipUnavailable(final String relationshipName) {
+ setRelationshipUnavailable(new Relationship.Builder().name(relationshipName).build());
+ }
+
+ @Override
+ public void addControllerService(final String identifier, final ControllerService service) throws InitializationException {
+ addControllerService(identifier, service, new HashMap<String, String>());
+ }
+
+ @Override
+ public void addControllerService(final String identifier, final ControllerService service, final Map<String, String> properties) throws InitializationException {
+ // hold off on failing due to deprecated annotation for now... will introduce later.
+// for ( final Method method : service.getClass().getMethods() ) {
+// if ( method.isAnnotationPresent(org.apache.nifi.controller.annotation.OnConfigured.class) ) {
+// Assert.fail("Controller Service " + service + " is using deprecated Annotation " + org.apache.nifi.controller.annotation.OnConfigured.class + " for method " + method);
+// }
+// }
+
+ final ComponentLog logger = new MockProcessorLog(identifier, service);
+ final MockControllerServiceInitializationContext initContext = new MockControllerServiceInitializationContext(requireNonNull(service), requireNonNull(identifier), logger);
+ service.initialize(initContext);
+
+ final Map<PropertyDescriptor, String> resolvedProps = new HashMap<>();
+ for (final Map.Entry<String, String> entry : properties.entrySet()) {
+ resolvedProps.put(service.getPropertyDescriptor(entry.getKey()), entry.getValue());
+ }
+
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnAdded.class, service);
+ } catch (final InvocationTargetException | IllegalAccessException | IllegalArgumentException e) {
+ throw new InitializationException(e);
+ }
+
+ context.addControllerService(identifier, service, resolvedProps, null);
+ }
+
+ @Override
+ public void assertNotValid(final ControllerService service) {
+ final ValidationContext validationContext = new MockValidationContext(context).getControllerServiceValidationContext(service);
+ final Collection<ValidationResult> results = context.getControllerService(service.getIdentifier()).validate(validationContext);
+
+ for (final ValidationResult result : results) {
+ if (!result.isValid()) {
+ return;
+ }
+ }
+
+ Assert.fail("Expected Controller Service " + service + " to be invalid but it is valid");
+ }
+
+ @Override
+ public void assertValid(final ControllerService service) {
+ final ValidationContext validationContext = new MockValidationContext(context).getControllerServiceValidationContext(service);
+ final Collection<ValidationResult> results = context.getControllerService(service.getIdentifier()).validate(validationContext);
+
+ for (final ValidationResult result : results) {
+ if (!result.isValid()) {
+ Assert.fail("Expected Controller Service to be valid but it is invalid due to: " + result.toString());
+ }
+ }
+ }
+
+ @Override
+ public void disableControllerService(final ControllerService service) {
+ final ControllerServiceConfiguration configuration = context.getConfiguration(service.getIdentifier());
+ if (configuration == null) {
+ throw new IllegalArgumentException("Controller Service " + service + " is not known");
+ }
+
+ if (!configuration.isEnabled()) {
+ throw new IllegalStateException("Controller service " + service + " cannot be disabled because it is not enabled");
+ }
+
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnDisabled.class, service);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ Assert.fail("Failed to disable Controller Service " + service + " due to " + e);
+ }
+
+ configuration.setEnabled(false);
+ }
+
+ @Override
+ public void enableControllerService(final ControllerService service) {
+ final ControllerServiceConfiguration configuration = context.getConfiguration(service.getIdentifier());
+ if (configuration == null) {
+ throw new IllegalArgumentException("Controller Service " + service + " is not known");
+ }
+
+ if (configuration.isEnabled()) {
+ throw new IllegalStateException("Cannot enable Controller Service " + service + " because it is not disabled");
+ }
+
+ try {
+ final ConfigurationContext configContext = new MockConfigurationContext(service, configuration.getProperties(), context);
+ ReflectionUtils.invokeMethodsWithAnnotation(OnEnabled.class, service, configContext);
+ } catch (final InvocationTargetException ite) {
+ ite.getCause().printStackTrace();
+ Assert.fail("Failed to enable Controller Service " + service + " due to " + ite.getCause());
+ } catch (final Exception e) {
+ e.printStackTrace();
+ Assert.fail("Failed to enable Controller Service " + service + " due to " + e);
+ }
+
+ configuration.setEnabled(true);
+ }
+
+ @Override
+ public boolean isControllerServiceEnabled(final ControllerService service) {
+ final ControllerServiceConfiguration configuration = context.getConfiguration(service.getIdentifier());
+ if (configuration == null) {
+ throw new IllegalArgumentException("Controller Service " + service + " is not known");
+ }
+
+ return configuration.isEnabled();
+ }
+
+ @Override
+ public void removeControllerService(final ControllerService service) {
+ disableControllerService(service);
+
+ try {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnRemoved.class, service);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ Assert.fail("Failed to remove Controller Service " + service + " due to " + e);
+ }
+
+ context.removeControllerService(service);
+ }
+
+ @Override
+ public void setAnnotationData(final ControllerService service, final String annotationData) {
+ final ControllerServiceConfiguration configuration = getConfigToUpdate(service);
+ configuration.setAnnotationData(annotationData);
+ }
+
+ private ControllerServiceConfiguration getConfigToUpdate(final ControllerService service) {
+ final ControllerServiceConfiguration configuration = context.getConfiguration(service.getIdentifier());
+ if (configuration == null) {
+ throw new IllegalArgumentException("Controller Service " + service + " is not known");
+ }
+
+ if (configuration.isEnabled()) {
+ throw new IllegalStateException("Controller service " + service + " cannot be modified because it is not disabled");
+ }
+
+ return configuration;
+ }
+
+ @Override
+ public ValidationResult setProperty(final ControllerService service, final PropertyDescriptor property, final AllowableValue value) {
+ return setProperty(service, property, value.getValue());
+ }
+
+ @Override
+ public ValidationResult setProperty(final ControllerService service, final PropertyDescriptor property, final String value) {
+ final ControllerServiceConfiguration configuration = getConfigToUpdate(service);
+ final Map<PropertyDescriptor, String> curProps = configuration.getProperties();
+ final Map<PropertyDescriptor, String> updatedProps = new HashMap<>(curProps);
+
+ final ValidationContext validationContext = new MockValidationContext(context).getControllerServiceValidationContext(service);
+ final ValidationResult validationResult = property.validate(value, validationContext);
+
+ updatedProps.put(property, value);
+ configuration.setProperties(updatedProps);
+
+ return validationResult;
+ }
+
+ @Override
+ public ValidationResult setProperty(final ControllerService service, final String propertyName, final String value) {
+ final PropertyDescriptor descriptor = service.getPropertyDescriptor(propertyName);
+ if (descriptor == null) {
+ return new ValidationResult.Builder()
+ .input(propertyName)
+ .explanation(propertyName + " is not a known Property for Controller Service " + service)
+ .subject("Invalid property")
+ .valid(false)
+ .build();
+ }
+ return setProperty(service, descriptor, value);
+ }
+
+ @Override
+ public ControllerService getControllerService(final String identifier) {
+ return context.getControllerService(identifier);
+ }
+
+ @Override
+ public <T extends ControllerService> T getControllerService(final String identifier, final Class<T> serviceType) {
+ final ControllerService service = context.getControllerService(identifier);
+ return serviceType.cast(service);
+ }
+
+ @Override
+ public boolean removeProperty(PropertyDescriptor descriptor) {
+ return context.removeProperty(descriptor);
+ }
+
+ @Override
+ public List<ProvenanceEventRecord> getProvenanceEvents() {
+ return sharedState.getProvenanceEvents();
+ }
+
+ @Override
+ public void clearProvenanceEvents() {
+ sharedState.clearProvenanceEvents();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java
new file mode 100644
index 0000000..fb9fc78
--- /dev/null
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java
@@ -0,0 +1,765 @@
+/*
+ * 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.util;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processor.QueueSize;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceReporter;
+import org.apache.nifi.reporting.InitializationException;
+
+public interface TestRunner {
+
+ /**
+ * @return the {@link Processor} for which this <code>TestRunner</code> is
+ * configured
+ */
+ Processor getProcessor();
+
+ /**
+ * @return the {@link ProcessSessionFactory} that this
+ * <code>TestRunner</code> will use to invoke the
+ * {@link Processor#onTrigger(ProcessContext, ProcessSessionFactory)} method
+ */
+ ProcessSessionFactory getProcessSessionFactory();
+
+ /**
+ * @return the {@Link ProcessContext} that this <code>TestRunner</code> will
+ * use to invoke the
+ * {@link Processor#onTrigger(ProcessContext, ProcessSessionFactory) onTrigger}
+ * method
+ */
+ ProcessContext getProcessContext();
+
+ /**
+ * Performs exactly the same operation as calling {@link #run(int)} with a
+ * value of 1.
+ */
+ void run();
+
+ /**
+ * Performs the same operation as calling {@link #run(int, boolean)} with a
+ * value of <code>true</code>
+ *
+ * @param iterations number of iterations
+ */
+ void run(int iterations);
+
+ /**
+ * performs the same operation as calling {@link #run(int, boolean, int)}
+ * with a value of {@code iterations}, {@code stopOnFinish}, {@code true}
+ *
+ * @param iterations number of iterations
+ * @param stopOnFinish flag to stop when finished
+ */
+ void run(int iterations, boolean stopOnFinish);
+
+ /**
+ * This method runs the {@link Processor} <code>iterations</code> times,
+ * using the sequence of steps below:
+ * <ul>
+ * <li>
+ * If {@code initialize} is true, run all methods on the Processor that are
+ * annotated with the
+ * {@link nifi.processor.annotation.OnScheduled @OnScheduled} annotation. If
+ * any of these methods throws an Exception, the Unit Test will fail.
+ * </li>
+ * <li>
+ * Schedule the
+ * {@link Processor#onTrigger(ProcessContext, ProcessSessionFactory) onTrigger}
+ * method to be invoked <code>iterations</code> times. The number of threads
+ * used to run these iterations is determined by the ThreadCount of this
+ * <code>TestRunner</code>. By default, the value is set to 1, but it can be
+ * modified by calling the {@link #setThreadCount(int)} method.
+ * </li>
+ * <li>
+ * As soon as the first thread finishes its execution of
+ * {@link Processor#onTrigger(ProcessContext, ProcessSessionFactory) onTrigger},
+ * all methods on the Processor that are annotated with the
+ * {@link nifi.processor.annotation.OnUnscheduled @OnUnscheduled} annotation
+ * are invoked. If any of these methods throws an Exception, the Unit Test
+ * will fail.
+ * </li>
+ * <li>
+ * Waits for all threads to finish execution.
+ * </li>
+ * <li>
+ * If and only if the value of <code>shutdown</code> is true: Call all
+ * methods on the Processor that is annotated with the
+ * {@link nifi.processor.annotation.OnStopped @OnStopped} annotation.
+ * </li>
+ * </ul>
+ *
+ * @param iterations number of iterations
+ * @param stopOnFinish whether or not to run the Processor methods that are
+ * annotated with {@link nifi.processor.annotation.OnStopped @OnStopped}
+ * @param initialize true if must initialize
+ */
+ void run(int iterations, boolean stopOnFinish, final boolean initialize);
+
+ /**
+ * This method runs the {@link Processor} <code>iterations</code> times,
+ * using the sequence of steps below:
+ * <ul>
+ * <li>
+ * If {@code initialize} is true, run all methods on the Processor that are
+ * annotated with the
+ * {@link nifi.processor.annotation.OnScheduled @OnScheduled} annotation. If
+ * any of these methods throws an Exception, the Unit Test will fail.
+ * </li>
+ * <li>
+ * Schedule the
+ * {@link Processor#onTrigger(ProcessContext, ProcessSessionFactory) onTrigger}
+ * method to be invoked <code>iterations</code> times. The number of threads
+ * used to run these iterations is determined by the ThreadCount of this
+ * <code>TestRunner</code>. By default, the value is set to 1, but it can be
+ * modified by calling the {@link #setThreadCount(int)} method.
+ * </li>
+ * <li>
+ * As soon as the first thread finishes its execution of
+ * {@link Processor#onTrigger(ProcessContext, ProcessSessionFactory) onTrigger},
+ * all methods on the Processor that are annotated with the
+ * {@link nifi.processor.annotation.OnUnscheduled @OnUnscheduled} annotation
+ * are invoked. If any of these methods throws an Exception, the Unit Test
+ * will fail.
+ * </li>
+ * <li>
+ * Waits for all threads to finish execution.
+ * </li>
+ * <li>
+ * If and only if the value of <code>shutdown</code> is true: Call all
+ * methods on the Processor that is annotated with the
+ * {@link nifi.processor.annotation.OnStopped @OnStopped} annotation.
+ * </li>
+ * </ul>
+ *
+ * @param iterations number of iterations
+ * @param stopOnFinish whether or not to run the Processor methods that are
+ * annotated with {@link nifi.processor.annotation.OnStopped @OnStopped}
+ * @param initialize true if must initialize
+ * @param runWait indicates the amount of time in milliseconds that the framework should wait for
+ * processors to stop running before calling the {@link nifi.processor.annotation.OnUnscheduled @OnUnscheduled} annotation
+ */
+ void run(int iterations, boolean stopOnFinish, final boolean initialize, final long runWait);
+
+ /**
+ * Invokes all methods on the Processor that are annotated with the
+ * {@link nifi.processor.annotation.OnShutdown @OnShutdown} annotation. If
+ * any of these methods throws an Exception, the Unit Test will fail
+ */
+ void shutdown();
+
+ /**
+ * Updates the number of threads that will be used to run the Processor when
+ * calling the {@link #run()} or {@link #run(int)} methods.
+ *
+ * @param threadCount num threads
+ */
+ void setThreadCount(int threadCount);
+
+ /**
+ * @return the currently configured number of threads that will be used to
+ * runt he Processor when calling the {@link #run()} or {@link #run(int)}
+ * methods
+ */
+ int getThreadCount();
+
+ /**
+ * Updates the value of the property with the given PropertyDescriptor to
+ * the specified value IF and ONLY IF the value is valid according to the
+ * descriptor's validator. Otherwise, Assert.fail() is called, causing the
+ * unit test to fail
+ *
+ * @param propertyName name
+ * @param propertyValue value
+ * @return result
+ */
+ ValidationResult setProperty(String propertyName, String propertyValue);
+
+ /**
+ * Updates the value of the property with the given PropertyDescriptor to
+ * the specified value IF and ONLY IF the value is valid according to the
+ * descriptor's validator. Otherwise, Assert.fail() is called, causing the
+ * unit test to fail
+ *
+ * @param descriptor descriptor
+ * @param value value
+ * @return result
+ */
+ ValidationResult setProperty(PropertyDescriptor descriptor, String value);
+
+ /**
+ * Updates the value of the property with the given PropertyDescriptor to
+ * the specified value IF and ONLY IF the value is valid according to the
+ * descriptor's validator. Otherwise, Assert.fail() is called, causing the
+ * unit test to fail
+ *
+ * @param descriptor descriptor
+ * @param value allowable valu
+ * @return result
+ */
+ ValidationResult setProperty(PropertyDescriptor descriptor, AllowableValue value);
+
+ /**
+ * Sets the annotation data.
+ *
+ * @param annotationData data
+ */
+ void setAnnotationData(String annotationData);
+
+ /**
+ * Asserts that all FlowFiles that were transferred were transferred to the
+ * given relationship
+ *
+ * @param relationship to verify
+ */
+ void assertAllFlowFilesTransferred(String relationship);
+
+ /**
+ * Asserts that all FlowFiles that were transferred were transferred to the
+ * given relationship
+ *
+ * @param relationship to verify
+ */
+ void assertAllFlowFilesTransferred(Relationship relationship);
+
+ /**
+ * Asserts that all FlowFiles that were transferred were transferred to the
+ * given relationship and that the number of FlowFiles transferred is equal
+ * to <code>count</code>
+ *
+ * @param relationship to verify
+ * @param count number of expected transfers
+ */
+ void assertAllFlowFilesTransferred(String relationship, int count);
+
+ /**
+ * Asserts that all FlowFiles that were transferred were transferred to the
+ * given relationship and that the number of FlowFiles transferred is equal
+ * to <code>count</code>
+ *
+ * @param relationship to verify
+ * @param count number of expected transfers
+ */
+ void assertAllFlowFilesTransferred(Relationship relationship, int count);
+
+ /**
+ * Assert that the number of FlowFiles transferred to the given relationship
+ * is equal to the given count
+ *
+ * @param relationship to verify
+ * @param count number of expected transfers
+ */
+ void assertTransferCount(Relationship relationship, int count);
+
+ /**
+ * Assert that the number of FlowFiles transferred to the given relationship
+ * is equal to the given count
+ *
+ * @param relationship to verify
+ * @param count number of expected transfers
+ */
+ void assertTransferCount(String relationship, int count);
+
+ /**
+ * Assert that there are no FlowFiles left on the input queue.
+ */
+ void assertQueueEmpty();
+
+ /**
+ * @return <code>true</code> if the Input Queue to the Processor is empty,
+ * <code>false</code> otherwise
+ */
+ boolean isQueueEmpty();
+
+ /**
+ * Assert that there is at least one FlowFile left on the input queue.
+ */
+ void assertQueueNotEmpty();
+
+ /**
+ * Assert that the currently configured set of properties/annotation data
+ * are valid
+ */
+ void assertValid();
+
+ /**
+ * Assert that the currently configured set of properties/annotation data
+ * are NOT valid
+ */
+ void assertNotValid();
+
+ /**
+ * Resets the Transfer Counts that indicate how many FlowFiles have been
+ * transferred to each Relationship and removes from memory any FlowFiles
+ * that have been transferred to this Relationships. This method should be
+ * called between successive calls to {@link #run(int) run} if the output is
+ * to be examined after each run.
+ */
+ void clearTransferState();
+
+ /**
+ * Enqueues the given FlowFiles into the Processor's input queue
+ *
+ * @param flowFiles to enqueue
+ */
+ void enqueue(FlowFile... flowFiles);
+
+ /**
+ * Reads the content from the given {@link Path} into memory and creates a
+ * FlowFile from this content with no attributes and adds this FlowFile to
+ * the Processor's Input Queue
+ *
+ * @param path to read content from
+ * @throws IOException if unable to read content
+ */
+ void enqueue(Path path) throws IOException;
+
+ /**
+ * Reads the content from the given {@link Path} into memory and creates a
+ * FlowFile from this content with the given attributes and adds this
+ * FlowFile to the Processor's Input Queue
+ *
+ * @param path to read content from
+ * @param attributes attributes to use for new flow file
+ * @throws IOException if unable to read content
+ */
+ void enqueue(Path path, Map<String, String> attributes) throws IOException;
+
+ /**
+ * Copies the content from the given byte array into memory and creates a
+ * FlowFile from this content with no attributes and adds this FlowFile to
+ * the Processor's Input Queue
+ *
+ * @param data to enqueue
+ */
+ void enqueue(byte[] data);
+
+ /**
+ * Copies the content from the given byte array into memory and creates a
+ * FlowFile from this content with the given attributes and adds this
+ * FlowFile to the Processor's Input Queue
+ *
+ * @param data to enqueue
+ * @param attributes to use for enqueued items
+ */
+ void enqueue(byte[] data, Map<String, String> attributes);
+
+ /**
+ * Reads the content from the given {@link InputStream} into memory and
+ * creates a FlowFile from this content with no attributes and adds this
+ * FlowFile to the Processor's Input Queue
+ *
+ * @param data to source data from
+ */
+ void enqueue(InputStream data);
+
+ /**
+ * Reads the content from the given {@link InputStream} into memory and
+ * creates a FlowFile from this content with the given attributes and adds
+ * this FlowFile to the Processor's Input Queue
+ *
+ * @param data source of data
+ * @param attributes to use for flow files
+ */
+ void enqueue(InputStream data, Map<String, String> attributes);
+
+ /**
+ * Copies the contents of the given {@link MockFlowFile} into a byte array
+ * and returns that byte array.
+ *
+ * @param flowFile to get content for
+ * @return byte array of flowfile content
+ */
+ byte[] getContentAsByteArray(MockFlowFile flowFile);
+
+ /**
+ * Returns a List of FlowFiles in the order in which they were transferred
+ * to the given relationship
+ *
+ * @param relationship to get flowfiles for
+ * @return flowfiles transfered to given relationship
+ */
+ List<MockFlowFile> getFlowFilesForRelationship(String relationship);
+
+ /**
+ * Returns a List of FlowFiles in the order in which they were transferred
+ * to the given relationship
+ *
+ * @param relationship to get flowfiles for
+ * @return flowfiles transfered to given relationship
+ */
+ List<MockFlowFile> getFlowFilesForRelationship(Relationship relationship);
+
+ /**
+ * @return the {@link ProvenanceReporter} that will be used by the
+ * configured {@link Processor} for reporting Provenance Events
+ */
+ ProvenanceReporter getProvenanceReporter();
+
+ /**
+ * @return the current size of the Processor's Input Queue
+ */
+ QueueSize getQueueSize();
+
+ /**
+ * @param name of counter
+ * @return the current value of the counter with the specified name, or null
+ * if no counter exists with the specified name
+ */
+ Long getCounterValue(String name);
+
+ /**
+ * @return the number of FlowFiles that have been removed from the system
+ */
+ int getRemovedCount();
+
+ /**
+ * Indicates to the Framework that the given Relationship should be
+ * considered "available", meaning that the queues of all Connections that
+ * contain this Relationship are not full. This is generally used only when
+ * dealing with Processors that use the
+ * {@link nifi.processor.annotation.TriggerWhenAnyDestinationAvailable}
+ * annotation.
+ *
+ * @param relationship to mark as available
+ */
+ void setRelationshipAvailable(Relationship relationship);
+
+ /**
+ * Indicates to the Framework that the given Relationship with the given
+ * name should be considered "available", meaning that the queues of all
+ * Connections that contain this Relationship are not full. This is
+ * generally used only when dealing with Processors that use the
+ * {@link nifi.processor.annotation.TriggerWhenAnyDestinationAvailable}
+ *
+ * @param relationshipName relationship name
+ */
+ void setRelationshipAvailable(String relationshipName);
+
+ /**
+ * Indicates to the Framework that the given Relationship should NOT be
+ * considered "available", meaning that the queue of at least one Connection
+ * that contain this Relationship is full. This is generally used only when
+ * dealing with Processors that use the
+ * {@link nifi.processor.annotation.TriggerWhenAnyDestinationAvailable}
+ * annotation.
+ *
+ * @param relationship to mark as unavailable
+ */
+ void setRelationshipUnavailable(Relationship relationship);
+
+ /**
+ * Indicates to the Framework that the Relationship with the given name
+ * should NOT be considered "available", meaning that the queue of at least
+ * one Connection that contain this Relationship is full. This is generally
+ * used only when dealing with Processors that use the
+ * {@link nifi.processor.annotation.TriggerWhenAnyDestinationAvailable}
+ *
+ * @param relationshipName name of relationship.
+ */
+ void setRelationshipUnavailable(String relationshipName);
+
+ /**
+ * Adds the given {@link ControllerService} to this TestRunner so that the
+ * configured Processor can access it using the given
+ * <code>identifier</code>. The ControllerService is not expected to be
+ * initialized, as the framework will create the appropriate
+ * {@link nifi.controller.ControllerServiceInitializationContext ControllerServiceInitializationContext}
+ * and initialize the ControllerService with no specified properties.
+ *
+ * This will call any method on the given Controller Service that is
+ * annotated with the
+ * {@link org.apache.nifi.annotation.lifecycle.OnAdded @OnAdded} annotation.
+ *
+ * @param identifier of service
+ * @param service the service
+ * @throws InitializationException ie
+ */
+ void addControllerService(String identifier, ControllerService service) throws InitializationException;
+
+ /**
+ * Adds the given {@link ControllerService} to this TestRunner so that the
+ * configured Processor can access it using the given
+ * <code>identifier</code>. The ControllerService is not expected to be
+ * initialized, as the framework will create the appropriate
+ * {@link nifi.controller.ControllerServiceInitializationContext ControllerServiceInitializationContext}
+ * and initialize the ControllerService with the given properties.
+ *
+ * This will call any method on the given Controller Service that is
+ * annotated with the
+ * {@link org.apache.nifi.annotation.lifecycle.OnAdded @OnAdded} annotation.
+ *
+ * @param identifier of service
+ * @param service the service
+ * @param properties service properties
+ * @throws InitializationException ie
+ */
+ void addControllerService(String identifier, ControllerService service, Map<String, String> properties) throws InitializationException;
+
+ /**
+ * <p>
+ * Marks the Controller Service as enabled so that it can be used by other
+ * components.
+ * </p>
+ *
+ * <p>
+ * This method will result in calling any method in the Controller Service
+ * that is annotated with the
+ * {@link org.apache.nifi.annotation.lifecycle.OnEnabled @OnEnabled}
+ * annotation.
+ * </p>
+ *
+ * @param service the service to enable
+ */
+ void enableControllerService(ControllerService service);
+
+ /**
+ * <p>
+ * Marks the Controller Service as disabled so that it cannot be used by
+ * other components.
+ * </p>
+ *
+ * <p>
+ * This method will result in calling any method in the Controller Service
+ * that is annotated with the
+ * {@link org.apache.nifi.annotation.lifecycle.OnDisabled @OnDisabled}
+ * annotation.
+ * </p>
+ *
+ * @param service the service to disable
+ */
+ void disableControllerService(ControllerService service);
+
+ /**
+ * @param service the service
+ * @return {@code true} if the given Controller Service is enabled,
+ * {@code false} if it is disabled
+ *
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ */
+ boolean isControllerServiceEnabled(ControllerService service);
+
+ /**
+ * <p>
+ * Removes the Controller Service from the TestRunner. This will call any
+ * method on the ControllerService that is annotated with the
+ * {@link org.apache.nifi.annotation.lifecycle.OnRemoved @OnRemoved}
+ * annotation.
+ * </p>
+ *
+ * @param service the service
+ *
+ * @throws IllegalStateException if the ControllerService is not disabled
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ *
+ */
+ void removeControllerService(ControllerService service);
+
+ /**
+ * Sets the given property on the given ControllerService
+ *
+ * @param service to modify
+ * @param property to modify
+ * @param value value to use
+ * @return result
+ *
+ * @throws IllegalStateException if the ControllerService is not disabled
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ *
+ */
+ ValidationResult setProperty(ControllerService service, PropertyDescriptor property, String value);
+
+ /**
+ * Sets the given property on the given ControllerService
+ *
+ * @param service to modify
+ * @param property to modify
+ * @param value value to use
+ * @return result
+ *
+ * @throws IllegalStateException if the ControllerService is not disabled
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ *
+ */
+ ValidationResult setProperty(ControllerService service, PropertyDescriptor property, AllowableValue value);
+
+ /**
+ * Sets the property with the given name on the given ControllerService
+ *
+ * @param service to modify
+ * @param propertyName to modify
+ * @param value value to use
+ * @return result
+ *
+ * @throws IllegalStateException if the ControllerService is not disabled
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ *
+ */
+ ValidationResult setProperty(ControllerService service, String propertyName, String value);
+
+ /**
+ * Sets the annontation data of the given service to the provided annotation
+ * data.
+ *
+ * @param service to modify
+ * @param annotationData the data
+ *
+ * @throws IllegalStateException if the Controller Service is not disabled
+ *
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ */
+ void setAnnotationData(ControllerService service, String annotationData);
+
+ /**
+ * @param identifier of controller service
+ * @return the {@link ControllerService} that is registered with the given
+ * identifier, or <code>null</code> if no Controller Service exists with the
+ * given identifier
+ */
+ ControllerService getControllerService(String identifier);
+
+ /**
+ * Assert that the currently configured set of properties/annotation data
+ * are valid for the given Controller Service.
+ *
+ * @param service the service to validate
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ */
+ void assertValid(ControllerService service);
+
+ /**
+ * Assert that the currently configured set of properties/annotation data
+ * are NOT valid for the given Controller Service.
+ *
+ * @param service the service to validate
+ * @throws IllegalArgumentException if the given ControllerService is not
+ * known by this TestRunner (i.e., it has not been added via the
+ * {@link #addControllerService(String, ControllerService)} or
+ * {@link #addControllerService(String, ControllerService, Map)} method or
+ * if the Controller Service has been removed via the
+ * {@link #removeControllerService(ControllerService)} method.
+ *
+ */
+ void assertNotValid(ControllerService service);
+
+ /**
+ * @param <T> type of service
+ * @param identifier identifier of service
+ * @param serviceType type of service
+ * @return the {@link ControllerService} that is registered with the given
+ * identifier, cast as the provided service type, or <code>null</code> if no
+ * Controller Service exists with the given identifier
+ *
+ * @throws ClassCastException if the identifier given is registered for a
+ * Controller Service but that Controller Service is not of the type
+ * specified
+ */
+ <T extends ControllerService> T getControllerService(String identifier, Class<T> serviceType);
+
+ /**
+ * Specifies whether or not the TestRunner will validate the use of
+ * Expression Language. By default, the value is <code>true</code>, which
+ * means that an Exception will be thrown if the Processor attempts to
+ * obtain the configured value of a Property without calling
+ * {@link nifi.components.PropertyValue#evaluateAttributeExpressions evaluateAttributeExpressions}
+ * on the Property Value or if
+ * {@link nifi.components.PropertyValue#evaluateAttributeExpressions evaluateAttributeExpressions}
+ * is called but the PropertyDescriptor indicates that the Expression
+ * Language is not supported.
+ *
+ * <p>
+ * <b>See Also:
+ * </b>{@link PropertyDescriptor.Builder#expressionLanguageSupported(boolean)}
+ * </p>
+ *
+ * @param validate whether there is any need to validate the EL was used
+ */
+ void setValidateExpressionUsage(boolean validate);
+
+ /**
+ * Removes the {@link PropertyDescriptor} from the {@link ProcessContext},
+ * effectively setting its value to null.
+ *
+ * @param descriptor of property to remove
+ * @return true if removed
+ */
+ boolean removeProperty(PropertyDescriptor descriptor);
+
+ /**
+ * Returns a {@link List} of all {@link ProvenanceEventRecord}s that were
+ * emitted by the Processor
+ *
+ * @return a List of all Provenance Events that were emitted by the
+ * Processor
+ */
+ List<ProvenanceEventRecord> getProvenanceEvents();
+
+ /**
+ * Clears the Provenance Events that have been emitted by the Processor
+ */
+ void clearProvenanceEvents();
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
new file mode 100644
index 0000000..f2b0b23
--- /dev/null
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
@@ -0,0 +1,37 @@
+/*
+ * 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.util;
+
+import org.apache.nifi.processor.Processor;
+
+public class TestRunners {
+
+ public static TestRunner newTestRunner(final Processor processor) {
+ return new StandardProcessorTestRunner(processor);
+ }
+
+ public static TestRunner newTestRunner(final Class<? extends Processor> processorClass) {
+ try {
+ return newTestRunner(processorClass.newInstance());
+
+ } catch (final Exception e) {
+ System.err.println("Could not instantiate instance of class " + processorClass.getName() + " due to: " + e);
+ throw new RuntimeException(e);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java b/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java
new file mode 100644
index 0000000..6b403af
--- /dev/null
+++ b/nifi-mock/src/test/java/org/apache/nifi/util/CurrentTestStandardProcessorTestRunner.java
@@ -0,0 +1,71 @@
+/*
+ * 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.util;
+
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class CurrentTestStandardProcessorTestRunner {
+
+ /**
+ * This test will verify that all iterations of the run are finished before unscheduled is called
+ */
+ @Test
+ public void testOnScheduledCalledAfterRunFinished() {
+ SlowRunProcessor processor = new SlowRunProcessor();
+ StandardProcessorTestRunner runner = new StandardProcessorTestRunner(processor);
+ final int iterations = 5;
+ runner.run(iterations);
+ // if the counter is not equal to iterations, the the processor must have been unscheduled
+ // before all the run calls were made, that would be bad.
+ Assert.assertEquals(iterations, processor.getCounter());
+ }
+
+ /**
+ * This processor simulates a "slow" processor that checks whether it is scheduled before doing something
+ *
+ *
+ */
+ private static class SlowRunProcessor extends AbstractProcessor {
+
+ private int counter = 0;
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+
+ try {
+ // be slow
+ Thread.sleep(50);
+ // make sure we are still scheduled
+ if (isScheduled()) {
+ // increment counter
+ ++counter;
+ }
+ } catch (InterruptedException e) {
+ }
+
+ }
+
+ public int getCounter() {
+ return counter;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aa998847/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java
----------------------------------------------------------------------
diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java
new file mode 100644
index 0000000..323a357
--- /dev/null
+++ b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java
@@ -0,0 +1,90 @@
+/*
+ * 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.util;
+
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.junit.Ignore;
+import org.junit.Test;
+
+@Ignore("This should not be enabled until we actually fail processor unit tests for using deprecated methods, which should happen in 0.1.0")
+public class TestStandardProcessorTestRunner {
+
+ @Test(expected = AssertionError.class)
+ public void testFailOnDeprecatedTypeAnnotation() {
+ new StandardProcessorTestRunner(new DeprecatedAnnotation());
+ }
+
+ @Test
+ public void testDoesNotFailOnNonDeprecatedTypeAnnotation() {
+ new StandardProcessorTestRunner(new NewAnnotation());
+ }
+
+ @Test(expected = AssertionError.class)
+ public void testFailOnDeprecatedMethodAnnotation() {
+ new StandardProcessorTestRunner(new DeprecatedMethodAnnotation());
+ }
+
+ @Test
+ public void testDoesNotFailOnNonDeprecatedMethodAnnotation() {
+ new StandardProcessorTestRunner(new NewMethodAnnotation());
+ }
+
+ @SuppressWarnings("deprecation")
+ @org.apache.nifi.processor.annotation.Tags({"deprecated"})
+ private static class DeprecatedAnnotation extends AbstractProcessor {
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ }
+ }
+
+ @org.apache.nifi.annotation.documentation.Tags({"deprecated"})
+ private static class NewAnnotation extends AbstractProcessor {
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ }
+ }
+
+ private static class NewMethodAnnotation extends AbstractProcessor {
+
+ @org.apache.nifi.annotation.lifecycle.OnScheduled
+ public void dummy() {
+
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ }
+ }
+
+ private static class DeprecatedMethodAnnotation extends AbstractProcessor {
+
+ @SuppressWarnings("deprecation")
+ @org.apache.nifi.processor.annotation.OnScheduled
+ public void dummy() {
+
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ }
+ }
+}