Refactors nFlow codebase to use modern Java 17 functional patterns

nflow-engine/src/main/java/io/nflow/engine/internal/dao/WorkflowInstanceDao.java

@@ -284,13 +284,12 @@ void insertVariables(final long id, final long actionId, Map<String, String> cha
   }
 
   private void insertVariablesWithMultipleUpdates(final long id, final long actionId, Map<String, String> changedStateVariables) {
-    for (Entry<String, String> entry : changedStateVariables.entrySet()) {
-      int updated = jdbc.update(insertWorkflowInstanceStateSql() + " values (?,?,?,?)", id, actionId, entry.getKey(),
-          entry.getValue());
+    changedStateVariables.forEach((key, value) -> {
+      int updated = jdbc.update(insertWorkflowInstanceStateSql() + " values (?,?,?,?)", id, actionId, key, value);
       if (updated != 1) {
-        throw new IllegalStateException("Failed to insert state variable " + entry.getKey());
+        throw new IllegalStateException("Failed to insert state variable " + key);
       }
-    }
+    });
   }
 
   private void insertVariablesWithBatchUpdate(final long id, final long actionId, Map<String, String> changedStateVariables) {
@@ -310,8 +309,8 @@ protected boolean setValuesIfAvailable(PreparedStatement ps, int i) throws SQLEx
             return true;
           }
         });
-    int updatedRows = 0;
     boolean unknownResults = false;
+    AtomicInteger updatedRows = new AtomicInteger(0);
     for (int i = 0; i < updateStatus.length; ++i) {
       if (updateStatus[i] == Statement.SUCCESS_NO_INFO) {
         unknownResults = true;
@@ -320,12 +319,13 @@ protected boolean setValuesIfAvailable(PreparedStatement ps, int i) throws SQLEx
       if (updateStatus[i] == Statement.EXECUTE_FAILED) {
         throw new IllegalStateException("Failed to insert/update state variable at index " + i + " (" + updateStatus[i] + ")");
       }
-      updatedRows += updateStatus[i];
+      updatedRows.addAndGet(updateStatus[i]);
     }
+    int updatedRowsCount = updatedRows.get();
     int changedVariables = changedStateVariables.size();
-    if (!unknownResults && updatedRows != changedVariables) {
+    if (!unknownResults && updatedRowsCount != changedVariables) {
       throw new IllegalStateException(
-          "Failed to insert/update state variables, expected update count " + changedVariables + ", actual " + updatedRows);
+          "Failed to insert/update state variables, expected update count " + changedVariables + ", actual " + updatedRowsCount);
     }
   }
 
@@ -385,14 +385,12 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
         }
         long parentActionId = insertWorkflowInstanceAction(action);
         insertVariables(action.workflowInstanceId, parentActionId, changedStateVariables);
-        for (WorkflowInstance childTemplate : childWorkflows) {
+        childWorkflows.forEach(childTemplate -> {
           WorkflowInstance childWorkflow = new WorkflowInstance.Builder(childTemplate).setParentWorkflowId(instance.id)
               .setParentActionId(parentActionId).build();
           insertWorkflowInstance(childWorkflow);
-        }
-        for (WorkflowInstance workflow : workflows) {
-          insertWorkflowInstance(workflow);
-        }
+        });
+        workflows.forEach(workflow -> insertWorkflowInstance(workflow));
       }
     });
   }
@@ -655,18 +653,18 @@ private List<Long> updateNextWorkflowInstancesWithBatchUpdate(List<OptimisticLoc
     List<Object[]> batchArgs = instances.stream()
         .map(instance -> new Object[] { instance.id, sqlVariants.tuneTimestampForDb(instance.modified) }).collect(toList());
     int[] updateStatuses = jdbc.batchUpdate(sql, batchArgs);
-    List<Long> ids = new ArrayList<>(instances.size());
-    for (int i = 0; i < updateStatuses.length; ++i) {
-      int status = updateStatuses[i];
-      if (status == 1) {
-        ids.add(instances.get(i).id);
-      } else if (status != 0) {
-        disableBatchUpdates.set(true);
-        throw new PollingBatchException(
-            "Database was unable to provide information about affected rows in a batch update. Disabling batch updates.");
-      }
-    }
-    return ids;
+    return Stream.iterate(0, i -> i < updateStatuses.length, i -> i + 1)
+        .peek(i -> {
+          int status = updateStatuses[i];
+          if (status != 1 && status != 0) {
+            disableBatchUpdates.set(true);
+            throw new PollingBatchException(
+                "Database was unable to provide information about affected rows in a batch update. Disabling batch updates.");
+          }
+        })
+        .filter(i -> updateStatuses[i] == 1)
+        .map(i -> instances.get(i).id)
+        .collect(toList());
   }
 
   private static class OptimisticLockKey extends ModelObject implements Comparable<OptimisticLockKey> {
@@ -799,10 +797,9 @@ private void fillChildWorkflowIds(final WorkflowInstance instance, boolean query
   }
 
   private long getMaxResults(Long maxResults) {
-    if (maxResults == null) {
-      return workflowInstanceQueryMaxResultsDefault;
-    }
-    return min(maxResults, workflowInstanceQueryMaxResults);
+    return Optional.ofNullable(maxResults)
+        .map(m -> min(m, workflowInstanceQueryMaxResults))
+        .orElse(workflowInstanceQueryMaxResultsDefault);
   }
 
   private void fillActions(WorkflowInstance instance, boolean includeStateVariables, Long requestedMaxActions) {
@@ -815,20 +812,17 @@ private void fillActions(WorkflowInstance instance, boolean includeStateVariable
     if (includeStateVariables) {
       Map<Long, Map<String, String>> actionStates = fetchActionStateVariables(instance, actionBuilders.size(), maxActions);
       actionBuilders.forEach(builder -> {
-        Map<String, String> actionState = actionStates.get(builder.getId());
-        if (actionState != null) {
-          builder.setUpdatedStateVariables(actionState);
-        }
+          Map<String, String> actionState = actionStates.get(builder.getId());
+          Optional.ofNullable(actionState).ifPresent(builder::setUpdatedStateVariables);
       });
     }
     actionBuilders.stream().map(WorkflowInstanceAction.Builder::build).forEach(instance.actions::add);
   }
 
   private long getMaxActions(Long maxActions) {
-    if (maxActions == null) {
-      return workflowInstanceQueryMaxActionsDefault;
-    }
-    return min(maxActions, workflowInstanceQueryMaxActions);
+    return Optional.ofNullable(maxActions)
+        .map(m -> min(m, workflowInstanceQueryMaxActions))
+        .orElse(workflowInstanceQueryMaxActionsDefault);
   }
 
   private Map<Long, Map<String, String>> fetchActionStateVariables(WorkflowInstance instance, long actions, long maxActions) {

nflow-engine/src/main/java/io/nflow/engine/internal/executor/WorkflowDispatcher.java

@@ -183,9 +183,7 @@ private void dispatch(List<Long> nextInstanceIds) {
       return;
     }
     logger.debug("Found {} workflow instances, dispatching executors.", nextInstanceIds.size());
-    for (Long instanceId : nextInstanceIds) {
-      executor.execute(stateProcessorFactory.createProcessor(instanceId, shutdownRequested::get));
-    }
+    nextInstanceIds.forEach(instanceId -> executor.execute(stateProcessorFactory.createProcessor(instanceId, shutdownRequested::get)));
   }
 
   private List<Long> getNextInstanceIds() {

nflow-engine/src/main/java/io/nflow/engine/internal/executor/WorkflowStateProcessor.java

@@ -9,6 +9,7 @@
 import static io.nflow.engine.workflow.instance.WorkflowInstanceAction.WorkflowActionType.stateExecutionFailed;
 import static java.lang.Thread.currentThread;
 import static java.util.Arrays.asList;
+import static java.util.Arrays.stream;
 import static java.util.Collections.emptyList;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.commons.lang3.exception.ExceptionUtils.getStackTrace;
@@ -24,7 +25,9 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
@@ -563,33 +566,33 @@ public NextAction processState() {
   }
 
   private void processBeforeListeners() {
-    for (WorkflowExecutorListener listener : executorListeners) {
+    executorListeners.forEach(listener -> {
       try {
         listener.beforeProcessing(listenerContext);
       } catch (Throwable t) {
         logger.error("Error in {}.beforeProcessing ({})", listener.getClass().getName(), t.getMessage(), t);
       }
-    }
+    });
   }
 
   private void processAfterListeners() {
-    for (WorkflowExecutorListener listener : executorListeners) {
+    executorListeners.forEach(listener -> {
       try {
         listener.afterProcessing(listenerContext);
       } catch (Throwable t) {
         logger.error("Error in {}.afterProcessing ({})", listener.getClass().getName(), t.getMessage(), t);
       }
-    }
+    });
   }
 
   private void processAfterFailureListeners(Throwable ex) {
-    for (WorkflowExecutorListener listener : executorListeners) {
+    executorListeners.forEach(listener -> {
       try {
         listener.afterFailure(listenerContext, ex);
       } catch (Throwable t) {
         logger.error("Error in {}.afterFailure ({})", listener.getClass().getName(), t.getMessage(), t);
       }
-    }
+    });
   }
 
   public DateTime getStartTime() {
@@ -602,25 +605,26 @@ public void logPotentiallyStuck(long processingTimeSeconds) {
   }
 
   private StringBuilder getStackTraceAsString() {
-    StringBuilder sb = new StringBuilder(2000);
-    for (StackTraceElement element : thread.getStackTrace()) {
-      sb.append(element).append('\n');
+    String stack = stream(thread.getStackTrace()).map(Object::toString).collect(Collectors.joining("\n"));
+    StringBuilder sb = new StringBuilder(stack.length() + 2);
+    if (!stack.isEmpty()) {
+      sb.append(stack).append('\n');
     }
     return sb;
   }
 
   public void handlePotentiallyStuck(Duration processingTime) {
-    boolean interrupt = false;
-    for (WorkflowExecutorListener listener : executorListeners) {
+    AtomicBoolean interrupt = new AtomicBoolean(false);
+    executorListeners.forEach(listener -> {
       try {
         if (listener.handlePotentiallyStuck(listenerContext, processingTime)) {
-          interrupt = true;
+          interrupt.set(true);
         }
       } catch (Throwable t) {
         logger.error("Error in " + listener.getClass().getName() + ".handleStuck (" + t.getMessage() + ")", t);
       }
-    }
-    if (interrupt) {
+    });
+    if (interrupt.get()) {
       thread.interrupt();
     }
   }

nflow-engine/src/main/java/io/nflow/engine/internal/executor/WorkflowStateProcessorFactory.java

@@ -4,6 +4,7 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 
 import jakarta.inject.Inject;
@@ -65,16 +66,16 @@ public WorkflowStateProcessor createProcessor(long instanceId, Supplier<Boolean>
 
   public int getPotentiallyStuckProcessors() {
     DateTime currentTime = now();
-    int potentiallyStuck = 0;
-    for (WorkflowStateProcessor processor : processingInstances.values()) {
+    AtomicInteger potentiallyStuck = new AtomicInteger(0);
+    processingInstances.values().forEach(processor -> {
       Duration processingTime = new Duration(processor.getStartTime(), currentTime);
       long processingTimeSeconds = processingTime.getStandardSeconds();
       if (processingTimeSeconds > stuckThreadThresholdSeconds) {
-        potentiallyStuck++;
+        potentiallyStuck.incrementAndGet();
         processor.logPotentiallyStuck(processingTimeSeconds);
         processor.handlePotentiallyStuck(processingTime);
       }
-    }
-    return potentiallyStuck;
+    });
+    return potentiallyStuck.get();
   }
 }

nflow-engine/src/main/java/io/nflow/engine/internal/workflow/ObjectStringMapper.java

@@ -4,6 +4,7 @@
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Type;
+import java.util.Optional;
 
 import io.nflow.engine.config.EngineConfiguration.EngineObjectMapperSupplier;
 import jakarta.inject.Inject;
@@ -81,22 +82,22 @@ public void storeArguments(StateExecution execution,
         continue;
       }
       Object value = args[i + 1];
-      if (value == null) {
-        continue;
-      }
-      String sVal;
-      if (param.mutable) {
-        value = ((Mutable<Object>) value).val;
-        if (value == null) {
-          continue;
+      Optional.ofNullable(value).ifPresent(v -> {
+        Object actual = v;
+        if (param.mutable) {
+          actual = ((Mutable<Object>) actual).val;
+          if (actual == null) {
+            return;
+          }
         }
-      }
-      if (String.class.equals(param.type)) {
-        sVal = (String) value;
-      } else {
-        sVal = convertFromObject(param.key, value);
-      }
-      execution.setVariable(param.key, sVal);
+        String sVal;
+        if (String.class.equals(param.type)) {
+          sVal = (String) actual;
+        } else {
+          sVal = convertFromObject(param.key, actual);
+        }
+        execution.setVariable(param.key, sVal);
+      });
     }
   }
 

nflow-engine/src/main/java/io/nflow/engine/internal/workflow/WorkflowInstancePreProcessor.java

@@ -13,6 +13,8 @@
 import io.nflow.engine.workflow.definition.WorkflowDefinition;
 import io.nflow.engine.workflow.instance.WorkflowInstance;
 
+import java.util.Optional;
+
 @Component
 public class WorkflowInstancePreProcessor {
 
@@ -28,10 +30,8 @@ public WorkflowInstancePreProcessor(WorkflowDefinitionService workflowDefinition
   }
 
   public WorkflowInstance process(WorkflowInstance instance) {
-    WorkflowDefinition def = workflowDefinitionService.getWorkflowDefinition(instance.type);
-    if (def == null) {
-      throw new IllegalArgumentException("No workflow definition found for type [" + instance.type + "]");
-    }
+    WorkflowDefinition def = Optional.ofNullable(workflowDefinitionService.getWorkflowDefinition(instance.type))
+        .orElseThrow(() -> new IllegalArgumentException("No workflow definition found for type [" + instance.type + "]"));
     WorkflowInstance.Builder builder = new WorkflowInstance.Builder(instance);
     if (instance.state == null) {
       builder.setState(def.getInitialState());

nflow-rest-api-common/src/main/java/io/nflow/rest/v1/converter/CreateWorkflowConverter.java

@@ -1,6 +1,7 @@
 package io.nflow.rest.v1.converter;
 
 import static java.lang.Boolean.FALSE;
+import static java.util.Optional.ofNullable;
 import static org.apache.commons.lang3.StringUtils.isNotEmpty;
 
 import java.util.Map.Entry;
@@ -27,24 +28,22 @@ public WorkflowInstance convert(CreateWorkflowInstanceRequest req) {
     WorkflowInstance.Builder builder = factory.newWorkflowInstanceBuilder().setType(req.type).setBusinessKey(req.businessKey)
         .setExternalId(req.externalId);
     if (!FALSE.equals(req.activate)) {
-      if (req.activationTime != null) {
-        builder.setNextActivation(req.activationTime);
-      }
+      ofNullable(req.activationTime).ifPresent(builder::setNextActivation);
     } else {
       builder.setNextActivation(null);
     }
     builder.setParentWorkflowId(req.parentWorkflowId);
     if (isNotEmpty(req.startState)) {
       builder.setState(req.startState);
     }
-    for (Entry<String, Object> entry : req.stateVariables.entrySet()) {
+    req.stateVariables.entrySet().forEach(entry -> {
       Object value = entry.getValue();
       if (value instanceof String) {
         builder.putStateVariable(entry.getKey(), (String) value);
       } else {
         builder.putStateVariable(entry.getKey(), value);
       }
-    }
+    });
     return builder.build();
   }