added event based scheduling

Author: sdeoras

new.go

@@ -7,10 +7,38 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-// NewDeferredErrGroup provides a new instance of Scheduler and a contexct derived from the input context.
-func NewDeferredErrGroup(ctx context.Context, dur time.Duration) (Scheduler, context.Context) {
+// New creates a new scheduler based on an input context and trigger mode.
+// Scheduler will cancel all pending jobs if the input context is cancelled.
+// Scheduler will trigger jobs based on trigger.
+// It returns a context that should be used in each function.
+// This output context is cancelled on first error in any of the scheduled jobs.
+func New(ctx context.Context, trig Trigger) (Scheduler, context.Context) {
 	s := new(deferredErrGroupScheduler)
 	eg, ctx := errgroup.WithContext(ctx)
-	s.eg, s.dur, s.m = eg, dur, make(map[string]context.CancelFunc)
+	s.eg, s.trig, s.m = eg, trig, make(map[string]context.CancelFunc)
 	return s, ctx
 }
+
+// NewTimeoutTrigger provides a new trigger based on a timeout.
+func NewTimeoutTrigger(dur time.Duration) Trigger {
+	return Trigger{value: dur}
+}
+
+// NewContextTrigger provides a new trigger based on a context.
+// Trigger fires when input context context is done.
+func NewContextTrigger(ctx context.Context) Trigger {
+	return Trigger{value: ctx}
+}
+
+// NewChannelTrigger provides a new trigger based on a channel.
+// Trigger fires when input chan is read.
+func NewChannelTrigger(c chan struct{}) Trigger {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		select {
+		case <-c:
+			cancel()
+		}
+	}()
+	return Trigger{value: ctx}
+}

deferredEg.go scheduler.go

@@ -2,25 +2,11 @@ package scheduler
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/google/uuid"
-	"golang.org/x/sync/errgroup"
 )
 
-// deferredErrGroupScheduler implements Scheduler interface providing deferred execution for a func.
-type deferredErrGroupScheduler struct {
-	// dur is time duration to wait before executing func.
-	dur time.Duration
-	// m is map of cancel func that can be accessed using keys.
-	m map[string]context.CancelFunc
-	// mu is a lock for modifying above map.
-	mu sync.Mutex
-	// eg is error group
-	eg *errgroup.Group
-}
-
 // Func implements Scheduler interface to provide a functional literal for use in errgroup.
 // Such func will execute the input func deferred in time.
 func (d *deferredErrGroupScheduler) Go(ctx context.Context,
@@ -36,10 +22,28 @@ func (d *deferredErrGroupScheduler) Go(ctx context.Context,
 	d.m[key] = cancelFunc
 	d.mu.Unlock()
 
+	var trig context.Context
+	var trigCancelFunc context.CancelFunc
+
+	switch v := d.trig.value.(type) {
+	case time.Duration:
+		// create a context based on timeout
+		trig, trigCancelFunc = context.WithTimeout(context.Background(), v)
+		// spawn a go-routine that will call cancel func once triggered
+		go func() {
+			select {
+			case <-trig.Done():
+				trigCancelFunc()
+			}
+		}()
+	case context.Context:
+		trig = v
+	}
+
 	// spawn func within an error group.
 	d.eg.Go(func() error {
 		select {
-		case <-time.After(d.dur):
+		case <-trig.Done():
 			err := f()
 			d.Cancel(key)
 			return err

deferredEg_test.go scheduler_test.go

@@ -17,7 +17,7 @@ func Test_Exec_NormalBehavior(t *testing.T) {
 
 	// get a new deferredErrGroupScheduler with a 1 second timeout.
 	// functions will be executed after one second delay.
-	group, ctx := NewDeferredErrGroup(ctx, time.Second)
+	group, ctx := New(ctx, NewTimeoutTrigger(time.Second))
 
 	// defer execute func setting value of execTime to be 1.
 	group.Go(ctx, func() error { execTime = time.Now(); return nil })
@@ -46,7 +46,7 @@ func Test_Exec_GloabalTimeout(t *testing.T) {
 
 	// get a new deferredErrGroupScheduler with a 2 second timeout.
 	// functions will be executed after one second delay.
-	group, ctx := NewDeferredErrGroup(ctx, 2*time.Second)
+	group, ctx := New(ctx, NewTimeoutTrigger(2*time.Second))
 
 	// defer execute func.
 	group.Go(ctx, func() error { c = 1; return nil })
@@ -78,7 +78,7 @@ func Test_Exec_GlobalCancel(t *testing.T) {
 
 	// get a new deferredErrGroupScheduler with a 1 second timeout.
 	// functions will be executed after one second delay.
-	group, ctx := NewDeferredErrGroup(ctx, time.Second)
+	group, ctx := New(ctx, NewTimeoutTrigger(time.Second))
 
 	// defer execute func that sets value of c to 1.
 	group.Go(ctx, func() error { c = 1; return nil })
@@ -115,7 +115,7 @@ func Test_Exec_CancelOneFunc(t *testing.T) {
 
 	// get a new deferredErrGroupScheduler with a 1 second timeout.
 	// functions will be executed after one second delay.
-	group, ctx := NewDeferredErrGroup(ctx, time.Second)
+	group, ctx := New(ctx, NewTimeoutTrigger(time.Second))
 
 	// defer execute func setting value of c to be 1.
 	key := group.Go(ctx, func() error { c[0] = 1; return nil })
@@ -154,7 +154,7 @@ func Test_Exec_OneFuncErrorsOut(t *testing.T) {
 
 	// get a new deferredErrGroupScheduler with a 1 second timeout.
 	// functions will be executed after one second delay.
-	group, ctx := NewDeferredErrGroup(ctx, time.Second)
+	group, ctx := New(ctx, NewTimeoutTrigger(time.Second))
 
 	// defer execute func setting value of c to be 1.
 	group.Go(ctx, func() error { c[0] = 1; return fmt.Errorf("to err is human") })

types.go

@@ -0,0 +1,26 @@
+package scheduler
+
+import (
+	"context"
+	"sync"
+
+	"golang.org/x/sync/errgroup"
+)
+
+type Trigger struct {
+	value interface{}
+}
+
+// deferredErrGroupScheduler implements Scheduler interface providing deferred execution for a func.
+type deferredErrGroupScheduler struct {
+	// trig is trigger indicating how user wants function to be triggered.
+	// trigger is either a timeout or an event defined by cancellation of a context.
+	// Use NewTimeoutTrigger or NewEvent to create options in New method.
+	trig Trigger
+	// m is map of cancel func that can be accessed using keys.
+	m map[string]context.CancelFunc
+	// mu is a lock for modifying above map.
+	mu sync.Mutex
+	// eg is error group
+	eg *errgroup.Group
+}