runnable-gorm.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"gorm.io/driver/mysql"
 	"gorm.io/gorm"
+	"sync"
 )
 
 func NewDBRunnableFromMap(data map[string]interface{}) (*DBRunnable, error) {
@@ -50,6 +51,7 @@ type DBRunnable struct {
 	Type  string
 	DSN   string
 	Query string
+	mu    sync.Mutex
 }
 
 type dbKey struct{}
@@ -109,6 +111,8 @@ func (d *DBRunnable) GetType() string {
 }
 
 func (d *DBRunnable) GetPersistence() RunnableImport {
+	d.mu.Lock()
+	defer d.mu.Unlock()
 
 	data := JSONMap{
 		"type":  d.Type,

runnable-http.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sync"
 )
 
 func NewHTTPRunnableFromMap(data map[string]interface{}) (*HTTPRunnable, error) {
@@ -62,6 +63,7 @@ type HTTPRunnable struct {
 	Method string
 	Header map[string]string
 	Body   string
+	mu     sync.Mutex
 }
 
 func (h *HTTPRunnable) Run(ctx context.Context) (RunResult[HTTPResult], error) {
@@ -102,6 +104,8 @@ func (h *HTTPRunnable) GetType() string {
 }
 
 func (h *HTTPRunnable) GetPersistence() RunnableImport {
+	h.mu.Lock()
+	defer h.mu.Unlock()
 
 	data := JSONMap{
 		"url":    h.URL,

runnable-mail.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"net/smtp"
+	"sync"
 )
 
 func NewMailRunnableFromMap(data map[string]interface{}) (*MailRunnable, error) {
@@ -96,6 +97,7 @@ type MailRunnable struct {
 	Username string
 	Password string
 	Headers  map[string]string
+	mu       sync.Mutex
 }
 
 func (m *MailRunnable) Run(_ context.Context) (RunResult[MailResult], error) {
@@ -158,6 +160,8 @@ func (m *MailRunnable) GetType() string {
 }
 
 func (m *MailRunnable) GetPersistence() RunnableImport {
+	m.mu.Lock()
+	defer m.mu.Unlock()
 
 	data := JSONMap{
 		"to":       m.To,

runnable-sftp.go

@@ -10,6 +10,7 @@ import (
 	"golang.org/x/crypto/ssh"
 	"io"
 	"os"
+	"sync"
 )
 
 func NewSFTPRunnableFromMap(data map[string]interface{}) (*SFTPRunnable, error) {
@@ -113,6 +114,7 @@ type SFTPRunnable struct {
 	SrcDir            string
 	DstDir            string
 	TransferDirection Direction
+	mu                sync.Mutex
 }
 
 func (s *SFTPRunnable) Run(_ context.Context) (RunResult[SFTPResult], error) {
@@ -302,6 +304,8 @@ func (s *SFTPRunnable) GetType() string {
 }
 
 func (s *SFTPRunnable) GetPersistence() RunnableImport {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 
 	data := JSONMap{
 		"host":               s.Host,

runnable-shell.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"sync"
 )
 
 func NewShellRunnableFromMap(data map[string]interface{}) (*ShellRunnable, error) {
@@ -51,6 +52,7 @@ func (s *ShellResult) GetError() (string, int) {
 type ShellRunnable struct {
 	ScriptPath string
 	Script     string
+	mu         sync.Mutex
 }
 
 func (s *ShellRunnable) Run(ctx context.Context) (RunResult[ShellResult], error) {
@@ -129,6 +131,8 @@ func (s *ShellRunnable) GetType() string {
 }
 
 func (s *ShellRunnable) GetPersistence() RunnableImport {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 
 	data := JSONMap{
 		"scriptPath": s.ScriptPath,

schedule-interval.go

@@ -37,7 +37,6 @@ func (s *IntervalScheduler) Schedule(job GenericJob, eventBus *EventBus) error {
 		for {
 			select {
 			case <-ticker.C:
-
 				if !job.IsPaused() {
 					eventBus.Publish(QueueJob, job)
 				}

worker.go

@@ -181,6 +181,10 @@ func (w *LocalWorker) run(jobChannel chan GenericJob, stopChan chan bool, cancel
 		select {
 		case job := <-jobChannel:
 
+			if stopFlag {
+				break
+			}
+
 			w.statisticMu.Lock()
 			w.statistic.JobsAssigned++
 			w.statistic.ActiveThreads++
@@ -233,12 +237,14 @@ func (w *LocalWorker) run(jobChannel chan GenericJob, stopChan chan bool, cancel
 
 			cancel()
 
-			if w.manager != nil && w.manager.dbSaver != nil {
-				err = w.manager.dbSaver.SaveJob(job)
-				if err != nil {
-					Error("Error while saving job", "job_id", job.GetID(), "error", err)
-
+			if w.manager != nil {
+				go func() {
+					w.manager.mu.Lock()
+					if w.manager.jobSyncer != nil {
+						w.manager.jobSyncer.AddJob(job)
 					}
+					w.manager.mu.Unlock()
+				}()
 			}
 			
 			w.statisticMu.Lock()
@@ -246,6 +252,7 @@ func (w *LocalWorker) run(jobChannel chan GenericJob, stopChan chan bool, cancel
 			w.statisticMu.Unlock()
 
 		case <-stopChan:
+			Info("Stopping worker thread", "worker", w.ID, "thread_id", workerThreadID)
 			stopFlag = true
 			break
 		}

worker_test.go

@@ -39,6 +39,10 @@ func (j DummyJob) ResetStats() {
 
 }
 
+func (j DummyJob) GetStats() JobStats {
+	return JobStats{}
+}
+
 func (j DummyJob) GetMaxRetries() uint {
 	return 0
 }