Merge ffabd2c into 01a01e6

CHANGELOG.md

@@ -3,10 +3,12 @@
 **Highlights**
 - Use constructor injection instead of field or setter injection in nFlow classes
 - Separate workflow definition scanning from `WorkflowDefinitionService`
+- Replace nflow.autoinit and nflow.definition.persist configuration options with nflow.definition.autopersist.
 
 **Details**
 - `nflow-engine`
   - Separate workflow definition scanning from `WorkflowDefinitionService` by introducing `WorkflowDefinitionSpringBeanScanner` and `WorkflowDefinitionClassNameScanner`. This allows breaking the circular dependency when a workflow definition uses `WorkflowInstanceService` (which depends on `WorkflowDefinitionService`, which depended on all workflow definitions). This enabled using constructor injection in all nFlow classes. 
+  - nflow.definition.autopersist=true works as having nflow.autoinit=true and nflow.definition.persist=true, e.g. definitions are persisted to database when they are registered to nFlow engine. When nflow.definition.autopersist is false, definitions are not automatically persisted when they are added, nor when nFlow engine starts. Instead, the definitions can be persisted by calling `WorkflowDefinitionService.persistWorkflowDefinitions` manually.
 
 ## 5.7.0 (2019-06-06)
 

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

@@ -18,7 +18,6 @@
 import io.nflow.engine.internal.dao.PollingRaceConditionException;
 import io.nflow.engine.internal.dao.WorkflowInstanceDao;
 import io.nflow.engine.internal.util.PeriodicLogger;
-import io.nflow.engine.service.WorkflowDefinitionService;
 
 @Component
 @SuppressFBWarnings(value = "MDM_RANDOM_SEED", justification = "rand does not need to be secure")
@@ -35,7 +34,6 @@ public class WorkflowDispatcher implements Runnable {
   private final WorkflowInstanceExecutor executor;
   private final WorkflowInstanceDao workflowInstances;
   private final WorkflowStateProcessorFactory stateProcessorFactory;
-  private final WorkflowDefinitionService workflowDefinitions;
   private final ExecutorDao executorDao;
   private final long sleepTimeMillis;
   private final int stuckThreadThresholdSeconds;
@@ -44,12 +42,10 @@ public class WorkflowDispatcher implements Runnable {
   @Inject
   @SuppressFBWarnings(value = "WEM_WEAK_EXCEPTION_MESSAGING", justification = "Transaction support exception message is fine")
   public WorkflowDispatcher(WorkflowInstanceExecutor executor, WorkflowInstanceDao workflowInstances,
-      WorkflowStateProcessorFactory stateProcessorFactory, WorkflowDefinitionService workflowDefinitions, ExecutorDao executorDao,
-      Environment env) {
+      WorkflowStateProcessorFactory stateProcessorFactory, ExecutorDao executorDao, Environment env) {
     this.executor = executor;
     this.workflowInstances = workflowInstances;
     this.stateProcessorFactory = stateProcessorFactory;
-    this.workflowDefinitions = workflowDefinitions;
     this.executorDao = executorDao;
     this.sleepTimeMillis = env.getRequiredProperty("nflow.dispatcher.sleep.ms", Long.class);
     this.stuckThreadThresholdSeconds = env.getRequiredProperty("nflow.executor.stuckThreadThreshold.seconds", Integer.class);
@@ -63,7 +59,6 @@ public WorkflowDispatcher(WorkflowInstanceExecutor executor, WorkflowInstanceDao
   public void run() {
     logger.info("Starting.");
     try {
-      workflowDefinitions.postProcessWorkflowDefinitions();
       running = true;
       while (!shutdownRequested) {
         if (paused) {

nflow-engine/src/main/java/io/nflow/engine/service/WorkflowDefinitionService.java

@@ -27,14 +27,12 @@ public class WorkflowDefinitionService {
 
   private volatile Map<String, AbstractWorkflowDefinition<? extends WorkflowState>> workflowDefinitions = new LinkedHashMap<>();
   private final WorkflowDefinitionDao workflowDefinitionDao;
-  private final boolean persistWorkflowDefinitions;
-  private final boolean autoInit;
+  private final boolean autoPersistDefinitions;
 
   @Inject
   public WorkflowDefinitionService(WorkflowDefinitionDao workflowDefinitionDao, Environment env) {
     this.workflowDefinitionDao = workflowDefinitionDao;
-    this.persistWorkflowDefinitions = env.getRequiredProperty("nflow.definition.persist", Boolean.class);
-    this.autoInit = env.getRequiredProperty("nflow.autoinit", Boolean.class);
+    this.autoPersistDefinitions = env.getRequiredProperty("nflow.definition.autopersist", Boolean.class);
   }
 
   /**
@@ -58,18 +56,19 @@ public List<AbstractWorkflowDefinition<? extends WorkflowState>> getWorkflowDefi
   }
 
   /**
-   * Persist all loaded workflow definitions if nflow.autoinit is false and nflow.definition.persist is true. If nflow.autoinit is
-   * true, definitions are persisted when they are added to managed definitions.
+   * Persist all loaded workflow definitions to database if nflow.definition.autopersist is false. If nflow.definition.autopersist
+   * is true, definitions are persisted automatically when they are added to managed definitions.
    */
-  public void postProcessWorkflowDefinitions() {
-    if (!autoInit && persistWorkflowDefinitions) {
+  public void persistWorkflowDefinitions() {
+    if (!autoPersistDefinitions) {
       workflowDefinitions.values().forEach(workflowDefinitionDao::storeWorkflowDefinition);
     }
   }
 
   /**
-   * Add given workflow definition to managed definitions. Persist given definition if nflow.autoinit and nflow.definition.persist
-   * are true.
+   * Add given workflow definition to managed definitions. Persist given definition to database if nflow.definition.autopersist is
+   * true. If nflow.definition.autopersist is false, call persistWorkflowDefinitions manually if needed to persist the
+   * definitions.
    *
    * @param wd
    *          The workflow definition to be added.
@@ -86,7 +85,7 @@ public void addWorkflowDefinition(AbstractWorkflowDefinition<? extends WorkflowS
       }
       workflowDefinitions = newDefinitions;
     }
-    if (autoInit && persistWorkflowDefinitions) {
+    if (autoPersistDefinitions) {
       workflowDefinitionDao.storeWorkflowDefinition(wd);
     }
     logger.info("Added workflow type: {} ({})", wd.getType(), wd.getClass().getName());

nflow-engine/src/main/resources/nflow-engine.properties

@@ -59,4 +59,4 @@ nflow.db.max_pool_size=4
 nflow.db.idle_timeout_seconds=600
 nflow.db.create_on_startup=true
 
-nflow.definition.persist=true
+nflow.definition.autopersist=true

nflow-engine/src/test/java/io/nflow/engine/internal/executor/WorkflowDispatcherTest.java

@@ -48,7 +48,6 @@
 import io.nflow.engine.internal.dao.ExecutorDao;
 import io.nflow.engine.internal.dao.WorkflowInstanceDao;
 import io.nflow.engine.listener.WorkflowExecutorListener;
-import io.nflow.engine.service.WorkflowDefinitionService;
 
 @ExtendWith(MockitoExtension.class)
 public class WorkflowDispatcherTest {
@@ -59,8 +58,6 @@ public class WorkflowDispatcherTest {
   @Mock
   WorkflowInstanceDao workflowInstances;
   @Mock
-  WorkflowDefinitionService workflowDefinitions;
-  @Mock
   ExecutorDao executorDao;
   @Mock
   WorkflowStateProcessorFactory executorFactory;
@@ -71,7 +68,6 @@ public class WorkflowDispatcherTest {
 
   @BeforeEach
   public void setup() {
-    env.setProperty("nflow.autoinit", "true");
     env.setProperty("nflow.dispatcher.sleep.ms", "0");
     env.setProperty("nflow.dispatcher.executor.queue.wait_until_threshold", "0");
     env.setProperty("nflow.illegal.state.change.action", "ignore");
@@ -82,7 +78,7 @@ public void setup() {
     env.setProperty("nflow.executor.stateSaveRetryDelay.seconds", "60");
     when(executorDao.isTransactionSupportEnabled()).thenReturn(true);
     executor = new WorkflowInstanceExecutor(3, 2, 0, 10, 0, new CustomizableThreadFactory("nflow-executor-"));
-    dispatcher = new WorkflowDispatcher(executor, workflowInstances, executorFactory, workflowDefinitions, executorDao, env);
+    dispatcher = new WorkflowDispatcher(executor, workflowInstances, executorFactory, executorDao, env);
     Logger logger = (Logger) getLogger(ROOT_LOGGER_NAME);
     logger.addAppender(mockAppender);
   }
@@ -96,7 +92,8 @@ public void teardown() {
   @Test
   public void workflowDispatcherCreationFailsWithoutTransactionSupport() {
     when(executorDao.isTransactionSupportEnabled()).thenReturn(false);
-    assertThrows(BeanCreationException.class, () -> new WorkflowDispatcher(executor, workflowInstances, executorFactory, workflowDefinitions, executorDao, env));
+    assertThrows(BeanCreationException.class,
+        () -> new WorkflowDispatcher(executor, workflowInstances, executorFactory, executorDao, env));
   }
 
   @Test
@@ -252,7 +249,7 @@ class ExceptionOnPoolShutdownIsNotPropagated extends MultithreadedTestCase {
       @Override
       public void initialize() {
         poolSpy = Mockito.spy(executor);
-        dispatcher = new WorkflowDispatcher(poolSpy, workflowInstances, executorFactory, workflowDefinitions, executorDao, env);
+        dispatcher = new WorkflowDispatcher(poolSpy, workflowInstances, executorFactory, executorDao, env);
       }
 
       public void threadDispatcher() {

nflow-engine/src/test/java/io/nflow/engine/service/WorkflowDefinitionServiceTest.java

@@ -34,27 +34,25 @@ public class WorkflowDefinitionServiceTest extends BaseNflowTest {
   @BeforeEach
   public void setup() {
     lenient().when(workflowDefinition.getType()).thenReturn("dummy");
+    initializeService(true);
   }
 
-  private void initializeService(boolean definitionPersist, boolean autoInit) {
-    when(env.getRequiredProperty("nflow.definition.persist", Boolean.class)).thenReturn(definitionPersist);
-    when(env.getRequiredProperty("nflow.autoinit", Boolean.class)).thenReturn(autoInit);
+  private void initializeService(boolean autoPersistDefinition) {
+    when(env.getRequiredProperty("nflow.definition.autopersist", Boolean.class)).thenReturn(autoPersistDefinition);
     service = new WorkflowDefinitionService(workflowDefinitionDao, env);
   }
 
   @Test
-  public void addedDefinitionIsStoredWhenAutoInitIsTrue() {
-    initializeService(true, true);
-
+  public void addedDefinitionIsStoredWhenAutoPersistDefinitionIsTrue() {
     service.addWorkflowDefinition(workflowDefinition);
 
     verify(workflowDefinitionDao).storeWorkflowDefinition(workflowDefinition);
     assertThat(service.getWorkflowDefinitions().size(), is(equalTo(1)));
   }
 
   @Test
-  public void addedDefinitionIsNotStoredWhenAutoInitIsFalse() {
-    initializeService(true, false);
+  public void addedDefinitionIsNotStoredWhenAutoPersistDefinitionIsFalse() {
+    initializeService(false);
 
     service.addWorkflowDefinition(workflowDefinition);
 
@@ -63,53 +61,31 @@ public void addedDefinitionIsNotStoredWhenAutoInitIsFalse() {
   }
 
   @Test
-  public void addedDefinitionIsNotStoredWhenDefinitionPersistIsFalse() {
-    initializeService(false, true);
-
+  public void persistWorkflowDefinitionStoresDefinitionsWhenAutoPersistDefinitionIsFalse() {
+    initializeService(false);
     service.addWorkflowDefinition(workflowDefinition);
 
-    verifyZeroInteractions(workflowDefinitionDao);
-    assertThat(service.getWorkflowDefinitions().size(), is(equalTo(1)));
-  }
-
-  @Test
-  public void definitionsAreStoredDuringPostProcessingWhenAutoInitIsFalse() {
-    initializeService(true, false);
-    service.addWorkflowDefinition(workflowDefinition);
-
-    service.postProcessWorkflowDefinitions();
+    service.persistWorkflowDefinitions();
 
     verify(workflowDefinitionDao).storeWorkflowDefinition(workflowDefinition);
     assertThat(service.getWorkflowDefinitions().size(), is(equalTo(1)));
   }
 
   @Test
-  public void definitionsAreNotStoredDuringPostProcessingWhenAutoInitIsTrue() {
-    initializeService(true, true);
+  public void persistWorkflowDefinitionDoesNotStoreDefinitionsWhenAutoPersistDefinitionIsTrue() {
     service.addWorkflowDefinition(workflowDefinition);
     verify(workflowDefinitionDao).storeWorkflowDefinition(workflowDefinition);
 
-    service.postProcessWorkflowDefinitions();
+    service.persistWorkflowDefinitions();
 
     verifyNoMoreInteractions(workflowDefinitionDao);
     assertThat(service.getWorkflowDefinitions().size(), is(equalTo(1)));
   }
 
-  @Test
-  public void definitionsAreNotStoredDuringPostProcessingWhenDefinitionPersistIsFalse() {
-    initializeService(false, false);
-    service.addWorkflowDefinition(workflowDefinition);
-
-    service.postProcessWorkflowDefinitions();
-
-    verifyZeroInteractions(workflowDefinitionDao);
-    assertThat(service.getWorkflowDefinitions().size(), is(equalTo(1)));
-  }
-
   @Test
   public void addingDuplicatDefinitionThrowsException() {
-    initializeService(true, true);
     service.addWorkflowDefinition(workflowDefinition);
+    verify(workflowDefinitionDao).storeWorkflowDefinition(workflowDefinition);
 
     IllegalStateException thrown = assertThrows(IllegalStateException.class,
         () -> service.addWorkflowDefinition(workflowDefinition));
@@ -118,22 +94,19 @@ public void addingDuplicatDefinitionThrowsException() {
     assertThat(thrown.getMessage(),
         containsString("Both " + className + " and " + className + " define same workflow type: dummy"));
     assertThat(service.getWorkflowDefinitions().size(), is(equalTo(1)));
-  }
-
-  @Test
-  public void getWorkflowDefinitionReturnsNullWhenTypeIsNotFound() {
-    initializeService(true, true);
-
-    assertThat(service.getWorkflowDefinition("notFound"), is(nullValue()));
+    verifyNoMoreInteractions(workflowDefinitionDao);
   }
 
   @Test
   public void getWorkflowDefinitionReturnsDefinitionWhenTypeIsFound() {
-    initializeService(true, true);
-
     service.addWorkflowDefinition(workflowDefinition);
 
     assertThat(service.getWorkflowDefinition("dummy"), is(instanceOf(DummyTestWorkflow.class)));
   }
 
+  @Test
+  public void getWorkflowDefinitionReturnsNullWhenTypeIsNotFound() {
+    assertThat(service.getWorkflowDefinition("dummy"), is(nullValue()));
+  }
+
 }

nflow-engine/src/test/java/io/nflow/engine/service/WorkflowDefinitionServiceWithSpringTest.java

@@ -57,7 +57,7 @@ static class ContextConfiguration {
     @Bean
     @Primary
     public Environment env() {
-      return new MockEnvironment().withProperty("nflow.definition.persist", "true").withProperty("nflow.autoinit", "true");
+      return new MockEnvironment().withProperty("nflow.definition.autopersist", "true");
     }
 
     @Bean

nflow-netty/src/test/java/io/nflow/netty/StartNflowTest.java

@@ -30,7 +30,7 @@ public void startNflowNetty() throws Exception {
     Map<String, Object> properties = new HashMap<>();
     properties.put("nflow.db.create_on_startup", false);
     properties.put("nflow.autostart", false);
-    properties.put("nflow.autoinit", false);
+    properties.put("nflow.definition.autopersist", false);
     ApplicationContext ctx = startNflow.startNetty(7500, "local", "", properties);
 
     assertNotNull(testListener.applicationContextEvent);
@@ -39,7 +39,7 @@ public void startNflowNetty() throws Exception {
     assertEquals("externallyDefinedExecutorGroup", ctx.getEnvironment().getProperty("nflow.executor.group"));
     assertEquals("false", ctx.getEnvironment().getProperty("nflow.db.create_on_startup"));
     assertEquals("false", ctx.getEnvironment().getProperty("nflow.autostart"));
-    assertEquals("false", ctx.getEnvironment().getProperty("nflow.autoinit"));
+    assertEquals("false", ctx.getEnvironment().getProperty("nflow.definition.autopersist"));
   }
 
 }

nflow-tests/src/test/java/io/nflow/tests/SkipAutoStartTest.java

@@ -2,26 +2,28 @@
 
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
-import io.nflow.tests.extension.NflowServerConfig;
-import io.nflow.tests.extension.NflowServerExtension;
 import org.junit.jupiter.api.MethodOrderer;
 import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestMethodOrder;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import io.nflow.tests.extension.NflowServerConfig;
+import io.nflow.tests.extension.NflowServerExtension;
+
 @ExtendWith(NflowServerExtension.class)
 @TestMethodOrder(MethodOrderer.OrderAnnotation.class)
 public class SkipAutoStartTest extends AbstractNflowTest {
 
-  // When nflow.autoinit, nflow.autostart and nflow.db.create_on_startup are false
-  // no database access should happen. This test fails if SQL statements are
-  // issued during bean initialization.
-  public static NflowServerConfig server = new NflowServerConfig.Builder()
-          .prop("nflow.autoinit", "false")
-          .prop("nflow.autostart", "false")
-          .prop("nflow.db.create_on_startup", "false")
-          .build();
+  /**
+   * When nflow.definition.autopersist, nflow.autostart and nflow.db.create_on_startup are false no database access should happen.
+   * This test fails if SQL statements are issued during bean initialization.
+   */
+  public static NflowServerConfig server = new NflowServerConfig.Builder() //
+      .prop("nflow.definition.autopersist", "false") //
+      .prop("nflow.autostart", "false") //
+      .prop("nflow.db.create_on_startup", "false") //
+      .build();
 
   public SkipAutoStartTest() {
     super(server);