StartWorkflow/EnqueueWorkflow (#25)

Author: manojdbos

src/main/java/dev/dbos/transact/DBOS.java

@@ -3,12 +3,12 @@
 import dev.dbos.transact.config.DBOSConfig;
 import dev.dbos.transact.database.SystemDatabase;
 import dev.dbos.transact.execution.DBOSExecutor;
+import dev.dbos.transact.execution.WorkflowFunction;
 import dev.dbos.transact.execution.RecoveryService;
 import dev.dbos.transact.http.HttpServer;
 import dev.dbos.transact.http.controllers.AdminController;
 import dev.dbos.transact.interceptor.AsyncInvocationHandler;
 import dev.dbos.transact.interceptor.QueueInvocationHandler;
-import dev.dbos.transact.interceptor.TransactInvocationHandler;
 import dev.dbos.transact.interceptor.UnifiedInvocationHandler;
 import dev.dbos.transact.migrations.MigrationManager;
 import dev.dbos.transact.notifications.NotificationService;
@@ -21,7 +21,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class DBOS {
@@ -273,7 +272,7 @@ public void shutdown() {
         }
     }
 
-    public static WorkflowHandle<?> retrieveWorkflow(String workflowId) {
+    public static <T> WorkflowHandle<T> retrieveWorkflow(String workflowId) {
         return DBOS.getInstance().dbosExecutor.retrieveWorkflow(workflowId);
     }
 
@@ -355,7 +354,7 @@ public void sleep(float seconds) {
      * @param workflowId id of the workflow
      * @return A handle to the workflow
      */
-    public WorkflowHandle<?> resumeWorkflow(String workflowId) {
+    public <T> WorkflowHandle<T> resumeWorkflow(String workflowId) {
         return this.dbosExecutor.resumeWorkflow(workflowId) ;
     }
 
@@ -380,10 +379,23 @@ public void cancelWorkflow(String workflowId) {
      * @param options {@link ForkOptions} containing forkedWorkflowId, applicationVersion, timeout
      * @return handle to the workflow
      */
-    public WorkflowHandle<?> forkWorkflow(String workflowId, int startStep, ForkOptions options) {
+    public <T> WorkflowHandle<T> forkWorkflow(String workflowId, int startStep, ForkOptions options) {
         return this.dbosExecutor.forkWorkflow(workflowId, startStep, options) ;
     }
 
+    /**
+     * Start a workflow asynchronously.
+     * If a queue is specified with DBOSOptions, the workflow is queued.
+     *
+     * @param func A function annotated with @Workflow
+     * @return handle {@link WorkflowHandle} to the workflow
+     * @param <T> type returned by the function
+     */
+
+    public <T> WorkflowHandle<T> startWorkflow(WorkflowFunction<T> func) {
+        return this.dbosExecutor.startWorkflow(func) ;
+    }
+
 
 }
 

src/main/java/dev/dbos/transact/context/DBOSContext.java

@@ -163,5 +163,17 @@ public Queue getQueue() {
     public long getWorkflowTimeoutMs() {
         return workflowTimeoutMs ;
     }
+
+    public DBOSContext copyWithAsync() {
+        return new DBOSContext(workflowId, functionId, parentWorkflowId, parentFunctionId, inWorkflow, true, queue, workflowTimeoutMs);
+    }
+
+    public DBOSContext copyWithQueue(Queue q) {
+        return new DBOSContext(workflowId, functionId, parentWorkflowId, parentFunctionId, inWorkflow, async, q, workflowTimeoutMs);
+    }
+
+    public DBOSContext copyWithWorkflowId(String id) {
+        return new DBOSContext(id, functionId, parentWorkflowId, parentFunctionId, inWorkflow, async, queue, workflowTimeoutMs);
+    }
 }
 

src/main/java/dev/dbos/transact/context/DBOSOptions.java

@@ -46,10 +46,10 @@ public Builder(String workflowId) {
             this.workflowId = workflowId;
         }
 
-        public Builder async() {
+        /* public Builder async() {
             this.async = true;
             return this;
-        }
+        } */
 
         public Builder queue(Queue queue) {
             this.queue = queue;

src/main/java/dev/dbos/transact/execution/DBOSExecutor.java

@@ -10,7 +10,6 @@
 import dev.dbos.transact.exceptions.*;
 import dev.dbos.transact.json.JSONUtil;
 import dev.dbos.transact.queue.Queue;
-import dev.dbos.transact.queue.QueueRegistry;
 import dev.dbos.transact.queue.QueueService;
 import dev.dbos.transact.workflow.ForkOptions;
 import dev.dbos.transact.workflow.WorkflowHandle;
@@ -25,7 +24,6 @@
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
-import java.sql.SQLException;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.*;
@@ -150,7 +148,7 @@ public <T> T syncWorkflow(String workflowName,
                              String targetClassName,
                              Object target,
                              Object[] args,
-                             WorkflowFunction function,
+                             WorkflowFunctionReflect function,
                              String workflowId) throws Throwable {
 
         String wfid = workflowId ;
@@ -213,7 +211,7 @@ public <T> T syncWorkflow(String workflowName,
     <T> T runAndSaveResult(
           Object target,
           Object[] args,
-          WorkflowFunction function,
+          WorkflowFunctionReflect function,
           String workflowId) throws Throwable {
 
         try {
@@ -249,7 +247,7 @@ public <T> WorkflowHandle<T> submitWorkflow(String workflowName,
                                                 String targetClassName,
                                                 Object target,
                                                 Object[] args,
-                                                WorkflowFunction function) throws Throwable {
+                                                WorkflowFunctionReflect function) throws Throwable {
 
         DBOSContext ctx = DBOSContextHolder.get();
         String workflowId = ctx.getWorkflowId() ;
@@ -332,7 +330,6 @@ public void enqueueWorkflow(String workflowName,
                                                 ) throws Throwable {
 
 
-
         DBOSContext ctx = DBOSContextHolder.get();
         String wfid = ctx.getWorkflowId() ;
 
@@ -363,7 +360,7 @@ public <T> T runStep(String stepName,
                              int maxAttempts,
                              float backOffRate,
                              Object[] args,
-                             DBOSFunction<T> function
+                             WorkflowFunction<T> function
                          ) throws Throwable {
 
 
@@ -435,7 +432,7 @@ public <T> T runStep(String stepName,
      * Retrieve the workflowHandle for the workflowId
      *
      */
-    public WorkflowHandle<?> retrieveWorkflow(String workflowId) {
+    public <R> WorkflowHandle<R> retrieveWorkflow(String workflowId) {
         return new WorkflowHandleDBPoll(workflowId, systemDatabase) ;
     }
 
@@ -488,7 +485,7 @@ public void sleep(float seconds) {
 
     }
 
-    public WorkflowHandle<?> resumeWorkflow(String workflowId)  {
+    public <T> WorkflowHandle<T> resumeWorkflow(String workflowId)  {
 
         Supplier<Void> resumeFunction = () -> {
             logger.info("Resuming workflow: ", workflowId);
@@ -512,7 +509,7 @@ public void cancelWorkflow(String workflowId)  {
 
     }
 
-    public WorkflowHandle<?> forkWorkflow(String workflowId, int startStep, ForkOptions options) {
+    public <T> WorkflowHandle<T> forkWorkflow(String workflowId, int startStep, ForkOptions options) {
 
         Supplier<String> forkFunction = () -> {
             logger.info(String.format("Forking workflow:%s from step:%d ", workflowId, startStep));
@@ -523,4 +520,29 @@ public WorkflowHandle<?> forkWorkflow(String workflowId, int startStep, ForkOpti
         String forkedId = systemDatabase.callFunctionAsStep(forkFunction, "DBOS.forkedWorkflow");
         return retrieveWorkflow(forkedId);
     }
+
+    public <T> WorkflowHandle<T> startWorkflow(WorkflowFunction<T> func) {
+        DBOSContext oldctx = DBOSContextHolder.get();
+        DBOSContext newCtx = oldctx ;
+
+        if (newCtx.getWorkflowId() == null) {
+            newCtx = newCtx.copyWithWorkflowId(UUID.randomUUID().toString()) ;
+        }
+
+        if (newCtx.getQueue() == null) {
+            newCtx = oldctx.copyWithAsync() ;
+        }
+
+        try {
+            DBOSContextHolder.set(newCtx);
+            func.execute();
+            return retrieveWorkflow(newCtx.getWorkflowId());
+        } catch(Throwable t) {
+            throw new DBOSException(UNEXPECTED.getCode(), t.getMessage());
+        } finally {
+            DBOSContextHolder.set(oldctx);
+        }
+
+    }
+
 }

src/main/java/dev/dbos/transact/execution/DBOSFunction.java

@@ -1,6 +1,6 @@
 package dev.dbos.transact.execution;
 
 @FunctionalInterface
-public interface WorkflowFunction {
-    Object invoke(Object target, Object[] args) throws Exception;
+public interface WorkflowFunction<T> {
+    T execute() throws Throwable;
 }

src/main/java/dev/dbos/transact/execution/WorkflowFunction.java

@@ -0,0 +1,6 @@
+package dev.dbos.transact.execution;
+
+@FunctionalInterface
+public interface WorkflowFunction0<R> {
+    R run() ;
+}

src/main/java/dev/dbos/transact/execution/WorkflowFunction0.java

@@ -0,0 +1,6 @@
+package dev.dbos.transact.execution;
+
+@FunctionalInterface
+public interface WorkflowFunction1<T1, R> {
+    R run(T1 arg1);
+}

src/main/java/dev/dbos/transact/execution/WorkflowFunction1.java

@@ -0,0 +1,6 @@
+package dev.dbos.transact.execution;
+
+@FunctionalInterface
+public interface WorkflowFunction2<T1, T2, R> {
+    R run(T1 arg1, T2 arg2);
+}

src/main/java/dev/dbos/transact/execution/WorkflowFunction2.java

@@ -0,0 +1,6 @@
+package dev.dbos.transact.execution;
+
+@FunctionalInterface
+public interface WorkflowFunctionReflect {
+    Object invoke(Object target, Object[] args) throws Exception;
+}