xsteadfastx/workgroups
1
0
Fork 0
Code Issues Pull Requests Projects Releases 7 Wiki Activity

workgroups.NewDispatcher takes a context and not an sync.Errgroup

Browse Source
This commit is contained in:
Marvin Preuss 2021-09-25 14:38:47 +02:00
parent f121b05b58
commit c9fd38c245
2 changed files with 7 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
workgroups.go
View File

@ -21,12 +21,14 @@ type Dispatcher struct {
numWorkers int numWorkers int
} }
func NewDispatcher(eg *errgroup.Group, numWorkers int) *Dispatcher { func NewDispatcher(ctx context.Context, numWorkers int) (*Dispatcher, context.Context) {
eg, ctx := errgroup.WithContext(ctx)
return &Dispatcher{ return &Dispatcher{
queue: make(chan Job, numWorkers), queue: make(chan Job, numWorkers),
eg: eg, eg: eg,
numWorkers: numWorkers, numWorkers: numWorkers,
} }, ctx
} }
func (d *Dispatcher) Start(ctx context.Context) { func (d *Dispatcher) Start(ctx context.Context) {

10
workgroups_test.go
View File

@ -12,7 +12,6 @@ import (
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"go.xsfx.dev/workgroups" "go.xsfx.dev/workgroups"
"golang.org/x/sync/errgroup"
) )
func TestDispatcher(t *testing.T) { func TestDispatcher(t *testing.T) {
@ -32,8 +31,7 @@ func TestDispatcher(t *testing.T) {
return nil return nil
} }
eg, ctx := errgroup.WithContext(context.Background()) d, ctx := workgroups.NewDispatcher(context.Background(), runtime.GOMAXPROCS(0))
d := workgroups.NewDispatcher(eg, runtime.GOMAXPROCS(0))
d.Start(ctx) d.Start(ctx)
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
@ -54,8 +52,7 @@ func TestDispatcherError(t *testing.T) {
return fmt.Errorf("this is an error") //nolint:goerr113 return fmt.Errorf("this is an error") //nolint:goerr113
} }
eg, ctx := errgroup.WithContext(context.Background()) d, ctx := workgroups.NewDispatcher(context.Background(), runtime.GOMAXPROCS(0))
d := workgroups.NewDispatcher(eg, runtime.GOMAXPROCS(0))
d.Start(ctx) d.Start(ctx)
d.Append(work) d.Append(work)
d.Close() d.Close()
@ -76,8 +73,7 @@ func TestDispatcherTimeout(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second/2) ctx, cancel := context.WithTimeout(context.Background(), time.Second/2)
defer cancel() defer cancel()
eg, ctx := errgroup.WithContext(ctx) d, ctx := workgroups.NewDispatcher(ctx, runtime.GOMAXPROCS(0))
d := workgroups.NewDispatcher(eg, runtime.GOMAXPROCS(0))
d.Start(ctx) d.Start(ctx)
d.Append(work) d.Append(work)
d.Close() d.Close()