Merge pull request #277 from cschleiden/ignore-poll-interval

Ignore polling interval for redis backend

Author: cschleiden

bench/main.go

@@ -48,6 +48,12 @@ func main() {
 
 	wo := worker.DefaultOptions
 	wo.WorkflowExecutorCacheSize = *cacheSize
+
+	if *b == "redis" {
+		wo.ActivityPollingInterval = 0
+		wo.WorkflowPollingInterval = 0
+	}
+
 	w := worker.New(ba, &wo)
 
 	w.RegisterWorkflow(Root)
@@ -150,7 +156,10 @@ func getBackend(b string, opt ...backend.BackendOption) backend.Backend {
 			ReadTimeout:  time.Second * 30,
 		})
 
-		rclient.FlushAll(context.Background()).Result()
+		_, err := rclient.FlushAll(context.Background()).Result()
+		if err != nil {
+			panic(err)
+		}
 
 		b, err := redis.NewRedisBackend(rclient, redis.WithBackendOptions(opt...))
 		if err != nil {

internal/worker/worker.go

@@ -80,10 +80,20 @@ func (w *Worker[Task, TaskResult]) WaitForCompletion() error {
 func (w *Worker[Task, TaskResult]) poller(ctx context.Context) {
 	defer w.pollersWg.Done()
 
-	ticker := time.NewTicker(w.options.PollingInterval)
-	defer ticker.Stop()
+	var ticker *time.Ticker
+
+	if w.options.PollingInterval > 0 {
+		ticker = time.NewTicker(w.options.PollingInterval)
+		defer ticker.Stop()
+	}
 
 	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
 		task, err := w.poll(ctx, 30*time.Second)
 		if err != nil {
 			w.logger.ErrorContext(ctx, "error polling task", "error", err)
@@ -92,10 +102,12 @@ func (w *Worker[Task, TaskResult]) poller(ctx context.Context) {
 			continue // check for new tasks right away
 		}
 
-		select {
-		case <-ticker.C:
-		case <-ctx.Done():
-			return
+		if w.options.PollingInterval > 0 {
+			select {
+			case <-ticker.C:
+			case <-ctx.Done():
+				return
+			}
 		}
 	}
 }