refactored processor state management to use action callbacks

Author: Mihhai

minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java

@@ -23,6 +23,7 @@
 import org.apache.nifi.controller.ProcessorNode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import java.util.function.BiConsumer;
 
 
 import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
@@ -40,33 +41,37 @@ public DefaultProcessorStateStrategy(FlowController flowController) {
 
     @Override
     public OperationState startProcessor(String processorId) {
-        return changeState(processorId, true);
+        return changeState(processorId, this::start);
     }
 
     @Override
     public OperationState stopProcessor(String processorId) {
-        return changeState(processorId, false);
+        return changeState(processorId, this::stop);
     }
 
-    private OperationState changeState(String processorId, boolean start) {
+    private OperationState changeState(String processorId, BiConsumer<String, String> action) {
         try {
             ProcessorNode node = flowController.getFlowManager().getProcessorNode(processorId);
             if (node == null) {
                 LOGGER.warn("Processor with id {} not found", processorId);
                 return NOT_APPLIED;
             }
             String parentGroupId = node.getProcessGroupIdentifier();
-            if (start) {
-                flowController.startProcessor(parentGroupId, processorId, true);
-                LOGGER.info("Started processor {} (group={})", processorId, parentGroupId);
-            } else {
-                flowController.stopProcessor(parentGroupId, processorId);
-                LOGGER.info("Stopped processor {} (group={})", processorId, parentGroupId);
-            }
+            action.accept(processorId, parentGroupId);
             return FULLY_APPLIED;
         } catch (Exception e) {
             LOGGER.error("Failed to change state for processor {}", processorId, e);
             return NOT_APPLIED;
         }
     }
+
+    private void start(String processorId, String parentGroupId) {
+        flowController.startProcessor(parentGroupId, processorId, true);
+        LOGGER.info("Started processor {} (group={})", processorId, parentGroupId);
+    }
+
+    private void stop(String processorId, String parentGroupId) {
+        flowController.stopProcessor(parentGroupId, processorId);
+        LOGGER.info("Stopped processor {} (group={})", processorId, parentGroupId);
+    }
 }