Add suspend/resume workflow to netcore

aelassas · Feb 11, 2024 · 11b983b · 11b983b
1 parent f1c58cb
commit 11b983b
Show file tree

Hide file tree

Showing 100 changed files with 363 additions and 47 deletions.
diff --git a/src/netcore/Wexflow.Core/Task.cs b/src/netcore/Wexflow.Core/Task.cs
@@ -447,5 +447,10 @@ public void ErrorFormat(string msg, Exception e, params object[] args)
                 Logs.Add($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture)}  ERROR - {message}\r\n{e}");
             }
         }
+
+        /// <summary>
+        /// This method is necessary for suspend/resume the current task.
+        /// </summary>
+        public void WaitOne() => Workflow.WaitOne();
     }
 }
diff --git a/src/netcore/Wexflow.Core/Workflow.cs b/src/netcore/Wexflow.Core/Workflow.cs
@@ -1783,6 +1783,13 @@ public bool Stop(string stoppedBy)
             return false;
         }
 
+        private readonly ManualResetEvent _event = new(true);
+
+        /// <summary>
+        /// If "unset" the thread will wait otherwise it will continue.
+        /// </summary>
+        internal void WaitOne() => _event.WaitOne();
+
         /// <summary>
         /// Suspends this workflow.
         /// </summary>
@@ -1792,9 +1799,11 @@ public bool Suspend()
             {
                 try
                 {
-#pragma warning disable CS0618 // Le type ou le membre est obsolète
-                    _thread.Suspend();
-#pragma warning restore CS0618 // Le type ou le membre est obsolète
+                    //#pragma warning disable CS0618 // Le type ou le membre est obsolète
+                    //                    _thread.Suspend();
+                    //#pragma warning restore CS0618 // Le type ou le membre est obsolète
+                    // unset the reset event which will cause the workflow to pause
+                    _event.Reset();
                     IsPaused = true;
                     Database.IncrementPendingCount();
                     Database.DecrementRunningCount();
@@ -1827,9 +1836,11 @@ public void Resume()
             {
                 try
                 {
-#pragma warning disable CS0618 // Le type ou le membre est obsolète
-                    _thread.Resume();
-#pragma warning restore CS0618 // Le type ou le membre est obsolète
+                    //#pragma warning disable CS0618 // Le type ou le membre est obsolète
+                    //                    _thread.Resume();
+                    //#pragma warning restore CS0618 // Le type ou le membre est obsolète
+                    // // set the reset event which will cause the workflow to continue
+                    _event.Set();
                     Database.IncrementRunningCount();
                     Database.DecrementPendingCount();
                     var entry = Database.GetEntry(Id, InstanceId);

diff --git a/src/netcore/Wexflow.Server/WexflowService.cs b/src/netcore/Wexflow.Server/WexflowService.cs
@@ -621,6 +621,8 @@ private void SuspendWorkflow()
         {
             _ = _endpoints.MapPost(GetPattern("suspend"), async context =>
             {
+                var res = false;
+
                 var auth = GetAuth(context.Request);
                 var username = auth.Username;
                 var password = auth.Password;
@@ -633,19 +635,19 @@ private void SuspendWorkflow()
                 {
                     if (user.UserProfile == Core.Db.UserProfile.SuperAdministrator)
                     {
-                        WexflowServer.WexflowEngine.SuspendWorkflow(workflowId, instanceId);
-                        await context.Response.WriteAsync(string.Empty);
+                        res = WexflowServer.WexflowEngine.SuspendWorkflow(workflowId, instanceId);
                     }
                     else if (user.UserProfile == Core.Db.UserProfile.Administrator)
                     {
                         var workflowDbId = WexflowServer.WexflowEngine.Workflows.First(w => w.Id == workflowId).DbId;
                         var check = WexflowServer.WexflowEngine.CheckUserWorkflow(user.GetDbId(), workflowDbId);
                         if (check)
                         {
-                            WexflowServer.WexflowEngine.SuspendWorkflow(workflowId, instanceId);
-                            await context.Response.WriteAsync(string.Empty);
+                            res = WexflowServer.WexflowEngine.SuspendWorkflow(workflowId, instanceId);
                         }
                     }
+
+                    await context.Response.WriteAsync(JsonConvert.SerializeObject(res));
                 }
                 else
                 {

diff --git a/src/netcore/Wexflow.Tasks.Approval/Approval.cs b/src/netcore/Wexflow.Tasks.Approval/Approval.cs
@@ -30,6 +30,7 @@ public override TaskStatus Run()
                     while (!File.Exists(trigger) && !Workflow.IsRejected && !IsStopped)
                     {
                         Thread.Sleep(1000);
+                        WaitOne();
                     }
 
                     IsWaitingForApproval = false;

diff --git a/src/netcore/Wexflow.Tasks.ApprovalRecordsCreator/ApprovalRecordsCreator.cs b/src/netcore/Wexflow.Tasks.ApprovalRecordsCreator/ApprovalRecordsCreator.cs
@@ -57,6 +57,10 @@ public override TaskStatus Run()
                         ErrorFormat("An error occured while importing the record {0}.", e, file.Path);
                         success = false;
                     }
+                    finally
+                    {
+                        WaitOne();
+                    }
                 }
 
                 var smKey = "ApprovalRecordsCreator.RecordIds";

diff --git a/src/netcore/Wexflow.Tasks.ApprovalWorkflowsCreator/ApprovalWorkflowsCreator.cs b/src/netcore/Wexflow.Tasks.ApprovalWorkflowsCreator/ApprovalWorkflowsCreator.cs
@@ -114,6 +114,10 @@ public override TaskStatus Run()
                                     e, recordId);
                                 success = false;
                             }
+                            finally
+                            {
+                                WaitOne();
+                            }
                         }
                     }
                 }

diff --git a/src/netcore/Wexflow.Tasks.ApproveRecord/ApproveRecord.cs b/src/netcore/Wexflow.Tasks.ApproveRecord/ApproveRecord.cs
@@ -701,6 +701,10 @@ record = Workflow.Database.GetRecord(RecordId);
                 Error("An error occured during approval process.", e);
                 status = Core.Status.Error;
             }
+            finally
+            {
+                WaitOne();
+            }
 
             Info("Approval process finished.");
             return new TaskStatus(status);

diff --git a/src/netcore/Wexflow.Tasks.CsvToJson/CsvToJson.cs b/src/netcore/Wexflow.Tasks.CsvToJson/CsvToJson.cs
@@ -80,6 +80,10 @@ private bool ConvertFiles(ref bool atLeastOneSuccess)
                     ErrorFormat("An error occured while converting the CSV file {0}: {1}", csvFile.Path, e.Message);
                     success = false;
                 }
+                finally
+                {
+                    WaitOne();
+                }
             }
 
             return success;

diff --git a/src/netcore/Wexflow.Tasks.CsvToSql/CsvToSql.cs b/src/netcore/Wexflow.Tasks.CsvToSql/CsvToSql.cs
@@ -116,6 +116,10 @@ private bool ConvertCsvToSql(string csvPath, string sqlPath, string tableName, s
                 ErrorFormat("An error occured while converting the CSV {0} to SQL: {1}", csvPath, e.Message);
                 return false;
             }
+            finally
+            {
+                WaitOne();
+            }
         }
     }
 }
diff --git a/src/netcore/Wexflow.Tasks.CsvToXml/CsvToXml.cs b/src/netcore/Wexflow.Tasks.CsvToXml/CsvToXml.cs
@@ -43,6 +43,10 @@ public override TaskStatus Run()
                     ErrorFormat("An error occured while creating the XML from {0} Please check this XML file according to the documentation of the task. Error: {1}", file.Path, e.Message);
                     success = false;
                 }
+                finally
+                {
+                    WaitOne();
+                }
             }
 
             var status = Status.Success;

diff --git a/src/netcore/Wexflow.Tasks.CsvToYaml/CsvToYaml.cs b/src/netcore/Wexflow.Tasks.CsvToYaml/CsvToYaml.cs
@@ -80,6 +80,10 @@ private bool ConvertFiles(ref bool atLeastOneSuccess)
                     ErrorFormat("An error occured while converting the CSV file {0}: {1}", csvFile.Path, e.Message);
                     success = false;
                 }
+                finally
+                {
+                    WaitOne();
+                }
             }
 
             return success;

diff --git a/src/netcore/Wexflow.Tasks.EnvironmentVariable/EnvironmentVariable.cs b/src/netcore/Wexflow.Tasks.EnvironmentVariable/EnvironmentVariable.cs
@@ -34,6 +34,10 @@ public override TaskStatus Run()
             {
                 ErrorFormat("An error occured while retrieving environment variable. Error: {0}", e.Message);
             }
+            finally
+            {
+                WaitOne();
+            }
 
             Info("Task finished.");
             return new TaskStatus(succeeded ? Status.Success : Status.Error, value);

diff --git a/src/netcore/Wexflow.Tasks.ExecPython/ExecPython.cs b/src/netcore/Wexflow.Tasks.ExecPython/ExecPython.cs
@@ -42,6 +42,10 @@ public override Core.TaskStatus Run()
                     ErrorFormat("An error occured while executing the script {0}: {1}", pythonFile.Path, e.Message);
                     success = false;
                 }
+                finally
+                {
+                    WaitOne();
+                }
             }
 
             if (!success && atLeastOneSuccess)

diff --git a/src/netcore/Wexflow.Tasks.FileContentMatch/FileContentMatch.cs b/src/netcore/Wexflow.Tasks.FileContentMatch/FileContentMatch.cs
@@ -43,6 +43,8 @@ public override TaskStatus Run()
                     {
                         InfoFormat("No content matching the pattern {0} was found in the file {1}.", Pattern, file);
                     }
+
+                    WaitOne();
                 }
 
                 // Checking folders
@@ -63,6 +65,8 @@ public override TaskStatus Run()
                         {
                             InfoFormat("No content matching the pattern {0} was found in the file {1}.", Pattern, file);
                         }
+
+                        WaitOne();
                     }
                 }
 

diff --git a/src/netcore/Wexflow.Tasks.FileExists/FileExists.cs b/src/netcore/Wexflow.Tasks.FileExists/FileExists.cs
@@ -35,6 +35,10 @@ public override TaskStatus Run()
                 ErrorFormat("An error occured while checking file {0}. Error: {1}", File, e.Message);
                 return new TaskStatus(Status.Error, false);
             }
+            finally
+            {
+                WaitOne();
+            }
 
             Info("Task finished");
 

diff --git a/src/netcore/Wexflow.Tasks.FileMatch/FileMatch.cs b/src/netcore/Wexflow.Tasks.FileMatch/FileMatch.cs
@@ -40,6 +40,8 @@ public override TaskStatus Run()
                         success = true;
                         break;
                     }
+
+                    WaitOne();
                 }
 
                 if (success)
@@ -61,6 +63,7 @@ public override TaskStatus Run()
                 return new TaskStatus(Status.Error, false);
             }
 
+
             Info("Task finished");
 
             return new TaskStatus(Status.Success, success);

diff --git a/src/netcore/Wexflow.Tasks.FileSystemWatcher/FileSystemWatcher.cs b/src/netcore/Wexflow.Tasks.FileSystemWatcher/FileSystemWatcher.cs
@@ -85,6 +85,7 @@ public override TaskStatus Run()
                     finally
                     {
                         Info("FileSystemWatcher.OnFound finished.");
+                        WaitOne();
                     }
                     try
                     {

diff --git a/src/netcore/Wexflow.Tasks.FilesConcat/FilesConcat.cs b/src/netcore/Wexflow.Tasks.FilesConcat/FilesConcat.cs
@@ -68,6 +68,10 @@ public override TaskStatus Run()
                             ErrorFormat("An error occured while concatenating the file {0}", e, file.Path);
                             success = false;
                         }
+                        finally
+                        {
+                            WaitOne();
+                        }
                     }
                 }
 

diff --git a/src/netcore/Wexflow.Tasks.FilesCopier/FilesCopier.cs b/src/netcore/Wexflow.Tasks.FilesCopier/FilesCopier.cs
@@ -78,6 +78,10 @@ public override TaskStatus Run()
                     ErrorFormat("An error occured while copying the file {0} to {1}.", e, file.Path, destPath);
                     success = false;
                 }
+                finally
+                {
+                    WaitOne();
+                }
             }
 
             var status = Status.Success;

diff --git a/src/netcore/Wexflow.Tasks.FilesDecryptor/FilesDecryptor.cs b/src/netcore/Wexflow.Tasks.FilesDecryptor/FilesDecryptor.cs
@@ -32,6 +32,8 @@ public override TaskStatus Run()
                     {
                         atLeastOneSuccess = true;
                     }
+
+                    WaitOne();
                 }
 
                 if (!succeeded && atLeastOneSuccess)

diff --git a/src/netcore/Wexflow.Tasks.FilesDiff/FilesDiff.cs b/src/netcore/Wexflow.Tasks.FilesDiff/FilesDiff.cs
@@ -28,6 +28,7 @@ public override TaskStatus Run()
             try
             {
                 CheckFiles();
+                WaitOne();
                 ts = new TaskStatus(Status.Success);
             }
             catch (ThreadInterruptedException)

diff --git a/src/netcore/Wexflow.Tasks.FilesEncryptor/FilesEncryptor.cs b/src/netcore/Wexflow.Tasks.FilesEncryptor/FilesEncryptor.cs
@@ -32,6 +32,7 @@ public override TaskStatus Run()
                     {
                         atLeastOneSuccess = true;
                     }
+                    WaitOne();
                 }
 
                 if (!succeeded && atLeastOneSuccess)

diff --git a/src/netcore/Wexflow.Tasks.FilesEqual/FilesEqual.cs b/src/netcore/Wexflow.Tasks.FilesEqual/FilesEqual.cs
@@ -61,6 +61,7 @@ public override TaskStatus Run()
 
                 xdoc.Save(xmlPath);
                 Files.Add(new FileInf(xmlPath, Id));
+                WaitOne();
                 InfoFormat("The result has been written in: {0}", xmlPath);
             }
             catch (ThreadInterruptedException)

diff --git a/src/netcore/Wexflow.Tasks.FilesExist/FilesExist.cs b/src/netcore/Wexflow.Tasks.FilesExist/FilesExist.cs
@@ -38,6 +38,7 @@ public override TaskStatus Run()
                         new XAttribute("path", file),
                         new XAttribute("name", Path.GetFileName(file)),
                         new XAttribute("exists", File.Exists(file))));
+                    WaitOne();
                 }
 
                 foreach (var folder in Folders)
@@ -46,6 +47,7 @@ public override TaskStatus Run()
                         new XAttribute("path", folder),
                         new XAttribute("name", Path.GetFileName(folder)),
                         new XAttribute("exists", Directory.Exists(folder))));
+                    WaitOne();
                 }
 
                 if (xdoc.Root != null)

diff --git a/src/netcore/Wexflow.Tasks.FilesInfo/FilesInfo.cs b/src/netcore/Wexflow.Tasks.FilesInfo/FilesInfo.cs
@@ -73,6 +73,10 @@ public override TaskStatus Run()
                         ErrorFormat("An error occured while generating file information of the file {0}", e, file.Path);
                         success = false;
                     }
+                    finally
+                    {
+                        WaitOne();
+                    }
                 }
                 xdoc.Save(filesInfoPath);
                 Files.Add(new FileInf(filesInfoPath, Id));

diff --git a/src/netcore/Wexflow.Tasks.FilesJoiner/FilesJoiner.cs b/src/netcore/Wexflow.Tasks.FilesJoiner/FilesJoiner.cs
@@ -137,6 +137,10 @@ private bool JoinFiles(string fileName, FileInf[] files)
                             ErrorFormat("An error occured while concatenating the file {0}", e, file.Path);
                             success = false;
                         }
+                        finally
+                        {
+                            WaitOne();
+                        }
                     }
 
                     if (success)

diff --git a/src/netcore/Wexflow.Tasks.FilesLoader/FilesLoader.cs b/src/netcore/Wexflow.Tasks.FilesLoader/FilesLoader.cs
@@ -44,6 +44,7 @@ public override TaskStatus Run()
                                 Files.Add(fi);
                                 InfoFormat("File loaded: {0}", file);
                             }
+                            WaitOne();
                         }
                     }
                 }
@@ -59,6 +60,7 @@ public override TaskStatus Run()
                                 Files.Add(fi);
                                 InfoFormat("File loaded: {0}", file);
                             }
+                            WaitOne();
                         }
                     }
                 }

diff --git a/src/netcore/Wexflow.Tasks.FilesLoaderEx/FilesLoaderEx.cs b/src/netcore/Wexflow.Tasks.FilesLoaderEx/FilesLoaderEx.cs
@@ -66,6 +66,7 @@ public override TaskStatus Run()
                                 FileInf fi = new(file, Id);
                                 folderFiles.Add(fi);
                             }
+                            WaitOne();
                         }
                     }
                 }
@@ -80,6 +81,7 @@ public override TaskStatus Run()
                                 FileInf fi = new(file, Id);
                                 folderFiles.Add(fi);
                             }
+                            WaitOne();
                         }
                     }
                 }

diff --git a/src/netcore/Wexflow.Tasks.FilesMover/FilesMover.cs b/src/netcore/Wexflow.Tasks.FilesMover/FilesMover.cs
@@ -104,6 +104,10 @@ public override TaskStatus Run()
                     ErrorFormat("An error occured while moving the file {0} to {1}", e, file.Path, destFilePath);
                     success = false;
                 }
+                finally
+                {
+                    WaitOne();
+                }
             }
 
             var status = Status.Success;

diff --git a/src/netcore/Wexflow.Tasks.FilesRemover/FilesRemover.cs b/src/netcore/Wexflow.Tasks.FilesRemover/FilesRemover.cs
@@ -44,6 +44,10 @@ public override TaskStatus Run()
                     ErrorFormat("An error occured while deleting the file {0}", e, file.Path);
                     success = false;
                 }
+                finally
+                {
+                    WaitOne();
+                }
             }
 
             var status = Status.Success;