FilesExpand file tree

develop

/

queue.go

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

57 lines (51 loc) · 1.25 KB

develop

/

queue.go

Copy path

Top

File metadata and controls

• Code
• Blame

57 lines (51 loc) · 1.25 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

package queue

import (

"sync"

"github.com/go-errors/errors"

)

// JobQueue implements a background job processor using a single channel and wait group.

//

// The channel is initialised with a maximum number of concurrent workers. Adding a new job

// consumes a worker from the channel. After finishing a job, the worker is added back to

// the channel. When all workers are consumed, adding new job will block.

//

// Example usage:

//

// jq := NewJobQueue(5)

// err := jq.Put(func() error {

// return nil

// })

// errors.Join(err, jq.Collect())

type JobQueue struct {

wg sync.WaitGroup

errCh chan error

}

func NewJobQueue(maxConcurrency uint) *JobQueue {

q := &JobQueue{

errCh: make(chan error, maxConcurrency),

}

for i := 0; i < cap(q.errCh); i++ {

q.errCh <- nil

}

return q

}

// Put runs a job in the background, returning any error from the previous job.

func (q *JobQueue) Put(job func() error) error {

err := <-q.errCh

q.wg.Add(1)

go func() {

defer q.wg.Done()

q.errCh <- job()

}()

return err

}

// Collect waits for all jobs to finish, returning any errors.

func (q *JobQueue) Collect() error {

q.wg.Wait()

var err []error

for i := 0; i < cap(q.errCh); i++ {

err = append(err, <-q.errCh)

q.errCh <- nil

}

return errors.Join(err...)

}