Merge branch 'main' into issue-937

Author: mcruzdev

.gitattributes

@@ -1 +1,2 @@
-* text eol=lf
+* text eol=lf
+*.db -text -eol -working-tree-encoding -merge -diff

.github/project.yml

@@ -1,4 +1,4 @@
-# 7.4.0 - third round
+# 7.4.0 - (4 round)
 release:
   current-version: 7.4.0.Final
   next-version: 8.0.0-SNAPSHOT

.github/workflows/release.yml

@@ -5,6 +5,7 @@ on:
     types: [closed]
     paths:
       - '.github/project.yml'
+  workflow_dispatch: {}
 
 permissions:
   contents: write
@@ -61,6 +62,11 @@ jobs:
         run: |
           git checkout -b release
           mvn -B release:prepare -DperformRelease -Prelease -DreleaseVersion=${{steps.metadata.outputs.current-version}} -DdevelopmentVersion=${{steps.metadata.outputs.next-version}}
+          
+          git restore --staged --worktree -- impl/test/db-samples/** 2>/dev/null || true
+          git clean -fd -- impl/test/db-samples 2>/dev/null || true
+    
+          git status --porcelain
           cat release.properties
           git checkout ${{github.base_ref}}
           git rebase release

experimental/fluent/agentic/src/test/java/io/serverlessworkflow/fluent/agentic/AgenticWorkflowHelperIT.java

@@ -178,10 +178,11 @@ public void errorHandling() {
 
     Map<String, Object> input =
         Map.of(
-            "style", "fantasy",
+            "topic", "fantasy",
+            "style", "funny",
             "audience", "young adults");
 
-    Map<String, Object> result;
+    Map<String, Object> result = null;
     try (WorkflowApplication app = WorkflowApplication.builder().build()) {
       result = app.workflowDefinition(wf).instance(input).start().get().asMap().orElseThrow();
     } catch (Exception e) {
@@ -210,7 +211,7 @@ public void conditionalWorkflow() {
                     conditional(RequestCategory.LEGAL::equals, legalExpert)))
             .build();
 
-    Map<String, Object> input = Map.of("question", "What is the best treatment for a common cold?");
+    Map<String, Object> input = Map.of("request", "What is the best treatment for a common cold?");
 
     Map<String, Object> result;
     try (WorkflowApplication app = WorkflowApplication.builder().build()) {

experimental/fluent/agentic/src/test/java/io/serverlessworkflow/fluent/agentic/Agents.java

@@ -15,7 +15,6 @@
  */
 package io.serverlessworkflow.fluent.agentic;
 
-import dev.langchain4j.agent.tool.Tool;
 import dev.langchain4j.agentic.Agent;
 import dev.langchain4j.agentic.internal.AgentSpecification;
 import dev.langchain4j.agentic.scope.AgenticScopeAccess;
@@ -371,7 +370,7 @@ interface MedicalExpert {
             Analyze the following user request under a medical point of view and provide the best possible answer.
             The user request is {{it}}.
             """)
-    @Tool("A medical expert")
+    @Agent("A medical expert")
     String medicalRequest(String request);
   }
 
@@ -383,7 +382,7 @@ interface LegalExpert {
             Analyze the following user request under a legal point of view and provide the best possible answer.
             The user request is {{it}}.
             """)
-    @Tool("A legal expert")
+    @Agent("A legal expert")
     String legalRequest(String request);
   }
 
@@ -395,7 +394,7 @@ interface TechnicalExpert {
             Analyze the following user request under a technical point of view and provide the best possible answer.
             The user request is {{it}}.
             """)
-    @Tool("A technical expert")
+    @Agent("A technical expert")
     String technicalRequest(String request);
   }
 

impl/core/pom.xml

@@ -24,5 +24,9 @@
             <groupId>de.huxhorn.sulky</groupId>
             <artifactId>de.huxhorn.sulky.ulid</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.cronutils</groupId>
+            <artifactId>cron-utils</artifactId>
+        </dependency>
     </dependencies>
 </project>

impl/core/src/main/java/io/serverlessworkflow/impl/SchedulerListener.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed 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 io.serverlessworkflow.impl;
+
+import io.serverlessworkflow.impl.lifecycle.WorkflowCompletedEvent;
+import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener;
+import io.serverlessworkflow.impl.scheduler.Cancellable;
+import io.serverlessworkflow.impl.scheduler.WorkflowScheduler;
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+class SchedulerListener implements WorkflowExecutionListener, AutoCloseable {
+
+  private final WorkflowScheduler scheduler;
+  private final Map<WorkflowDefinition, WorkflowValueResolver<Duration>> afterMap =
+      new ConcurrentHashMap<>();
+  private final Map<WorkflowDefinition, Cancellable> cancellableMap = new ConcurrentHashMap<>();
+
+  public SchedulerListener(WorkflowScheduler scheduler) {
+    this.scheduler = scheduler;
+  }
+
+  public void addAfter(WorkflowDefinition definition, WorkflowValueResolver<Duration> after) {
+    afterMap.put(definition, after);
+  }
+
+  @Override
+  public void onWorkflowCompleted(WorkflowCompletedEvent ev) {
+    WorkflowDefinition workflowDefinition = (WorkflowDefinition) ev.workflowContext().definition();
+    WorkflowValueResolver<Duration> after = afterMap.get(workflowDefinition);
+    if (after != null) {
+      cancellableMap.put(
+          workflowDefinition,
+          scheduler.scheduleAfter(
+              workflowDefinition,
+              after.apply((WorkflowContext) ev.workflowContext(), null, ev.output())));
+    }
+  }
+
+  public void removeAfter(WorkflowDefinition definition) {
+    if (afterMap.remove(definition) != null) {
+      Cancellable cancellable = cancellableMap.remove(definition);
+      if (cancellable != null) {
+        cancellable.cancel();
+      }
+    }
+  }
+
+  @Override
+  public void close() {
+    cancellableMap.values().forEach(c -> c.cancel());
+    cancellableMap.clear();
+    afterMap.clear();
+  }
+}

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java

@@ -36,6 +36,7 @@
 import io.serverlessworkflow.impl.resources.ExternalResourceHandler;
 import io.serverlessworkflow.impl.resources.ResourceLoaderFactory;
 import io.serverlessworkflow.impl.scheduler.DefaultWorkflowScheduler;
+import io.serverlessworkflow.impl.scheduler.WorkflowScheduler;
 import io.serverlessworkflow.impl.schema.SchemaValidator;
 import io.serverlessworkflow.impl.schema.SchemaValidatorFactory;
 import java.util.ArrayList;
@@ -71,6 +72,7 @@ public class WorkflowApplication implements AutoCloseable {
   private final Map<String, WorkflowAdditionalObject<?>> additionalObjects;
   private final ConfigManager configManager;
   private final SecretManager secretManager;
+  private final SchedulerListener schedulerListener;
 
   private WorkflowApplication(Builder builder) {
     this.taskFactory = builder.taskFactory;
@@ -81,13 +83,14 @@ private WorkflowApplication(Builder builder) {
     this.idFactory = builder.idFactory;
     this.runtimeDescriptorFactory = builder.descriptorFactory;
     this.executorFactory = builder.executorFactory;
-    this.listeners = builder.listeners != null ? builder.listeners : Collections.emptySet();
+    this.listeners = builder.listeners;
     this.definitions = new ConcurrentHashMap<>();
     this.eventConsumer = builder.eventConsumer;
     this.eventPublishers = builder.eventPublishers;
     this.lifeCycleCEPublishingEnabled = builder.lifeCycleCEPublishingEnabled;
     this.modelFactory = builder.modelFactory;
     this.scheduler = builder.scheduler;
+    this.schedulerListener = builder.schedulerListener;
     this.additionalObjects = builder.additionalObjects;
     this.configManager = builder.configManager;
     this.secretManager = builder.secretManager;
@@ -169,6 +172,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private Map<String, WorkflowAdditionalObject<?>> additionalObjects;
     private SecretManager secretManager;
     private ConfigManager configManager;
+    private SchedulerListener schedulerListener;
 
     private Builder() {}
 
@@ -304,6 +308,8 @@ public WorkflowApplication build() {
       if (scheduler == null) {
         scheduler = new DefaultWorkflowScheduler();
       }
+      schedulerListener = new SchedulerListener(scheduler);
+      listeners.add(schedulerListener);
       if (additionalObjects == null) {
         additionalObjects = Collections.emptyMap();
       }
@@ -388,6 +394,10 @@ public SecretManager secretManager() {
     return secretManager;
   }
 
+  SchedulerListener schedulerListener() {
+    return schedulerListener;
+  }
+
   public <T> Optional<T> additionalObject(
       String name, WorkflowContext workflowContext, TaskContext taskContext) {
     return Optional.ofNullable(additionalObjects.get(name))

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java

@@ -20,24 +20,27 @@
 import static io.serverlessworkflow.impl.WorkflowUtils.safeClose;
 
 import io.serverlessworkflow.api.types.Input;
-import io.serverlessworkflow.api.types.ListenTo;
 import io.serverlessworkflow.api.types.Output;
 import io.serverlessworkflow.api.types.Schedule;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.events.EventRegistrationBuilderInfo;
 import io.serverlessworkflow.impl.executors.TaskExecutor;
 import io.serverlessworkflow.impl.executors.TaskExecutorHelper;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
+import io.serverlessworkflow.impl.scheduler.Cancellable;
 import io.serverlessworkflow.impl.scheduler.ScheduledEventConsumer;
+import io.serverlessworkflow.impl.scheduler.WorkflowScheduler;
 import io.serverlessworkflow.impl.schema.SchemaValidator;
 import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 
 public class WorkflowDefinition implements AutoCloseable, WorkflowDefinitionData {
 
   private final Workflow workflow;
+  private final WorkflowDefinitionId definitionId;
   private Optional<SchemaValidator> inputSchemaValidator = Optional.empty();
   private Optional<SchemaValidator> outputSchemaValidator = Optional.empty();
   private Optional<WorkflowFilter> inputFilter = Optional.empty();
@@ -47,10 +50,13 @@ public class WorkflowDefinition implements AutoCloseable, WorkflowDefinitionData
   private final ResourceLoader resourceLoader;
   private final Map<String, TaskExecutor<?>> executors = new HashMap<>();
   private ScheduledEventConsumer scheculedConsumer;
+  private Cancellable everySchedule;
+  private Cancellable cronSchedule;
 
   private WorkflowDefinition(
       WorkflowApplication application, Workflow workflow, ResourceLoader resourceLoader) {
     this.workflow = workflow;
+    this.definitionId = WorkflowDefinitionId.of(workflow);
     this.application = application;
     this.resourceLoader = resourceLoader;
 
@@ -84,15 +90,28 @@ static WorkflowDefinition of(WorkflowApplication application, Workflow workflow,
             application.resourceLoaderFactory().getResourceLoader(application, path));
     Schedule schedule = workflow.getSchedule();
     if (schedule != null) {
-      ListenTo to = schedule.getOn();
-      if (to != null) {
+      WorkflowScheduler scheduler = application.scheduler();
+      if (schedule.getOn() != null) {
         definition.scheculedConsumer =
-            application
-                .scheduler()
-                .eventConsumer(
-                    definition,
-                    application.modelFactory()::from,
-                    EventRegistrationBuilderInfo.from(application, to, x -> null));
+            scheduler.eventConsumer(
+                definition,
+                application.modelFactory()::from,
+                EventRegistrationBuilderInfo.from(application, schedule.getOn(), x -> null));
+      }
+      if (schedule.getAfter() != null) {
+        application
+            .schedulerListener()
+            .addAfter(definition, WorkflowUtils.fromTimeoutAfter(application, schedule.getAfter()));
+      }
+      if (schedule.getCron() != null) {
+        definition.cronSchedule = scheduler.scheduleCron(definition, schedule.getCron());
+      }
+      if (schedule.getEvery() != null) {
+        definition.everySchedule =
+            scheduler.scheduleEvery(
+                definition,
+                WorkflowUtils.fromTimeoutAfter(application, schedule.getEvery())
+                    .apply(null, null, application.modelFactory().fromNull()));
       }
     }
     return definition;
@@ -148,7 +167,28 @@ public void addTaskExecutor(WorkflowMutablePosition position, TaskExecutor<?> ta
 
   @Override
   public void close() {
-    safeClose(scheculedConsumer);
     safeClose(resourceLoader);
+    safeClose(scheculedConsumer);
+    application.schedulerListener().removeAfter(this);
+    if (everySchedule != null) {
+      everySchedule.cancel();
+    }
+    if (cronSchedule != null) {
+      cronSchedule.cancel();
+    }
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(definitionId);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) return true;
+    if (obj == null) return false;
+    if (getClass() != obj.getClass()) return false;
+    WorkflowDefinition other = (WorkflowDefinition) obj;
+    return Objects.equals(definitionId, other.definitionId);
   }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowError.java

@@ -56,6 +56,10 @@ public static Builder runtime(TaskContext context, Exception ex) {
     return runtime(Errors.RUNTIME.status(), context, ex);
   }
 
+  public static Builder timeout() {
+    return error(Errors.TIMEOUT.toString(), Errors.TIMEOUT.status());
+  }
+
   public static class Builder {
 
     private final String type;