Merge pull request #2 from lxzan/dev

switch dao/deque

go.mod

@@ -3,8 +3,9 @@ module github.com/lxzan/concurrency
 go 1.20
 
 require (
+	github.com/lxzan/dao v1.1.7
 	github.com/pkg/errors v0.9.1
-	github.com/stretchr/testify v1.8.1
+	github.com/stretchr/testify v1.8.4
 )
 
 require (

go.sum

@@ -1,19 +1,14 @@
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/lxzan/dao v1.1.7 h1:I049e67buJIpr4QJ/vJbHSjKMLN4ZJlSMeK3Rq+CJl8=
+github.com/lxzan/dao v1.1.7/go.mod h1:5ChTIo7RSZ4upqRo16eicJ3XdJWhGwgMIsyuGLMUofM=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
-github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
-github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
-github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

groups/group.go

@@ -20,18 +20,18 @@ type (
 	Caller func(args any, f func(any) error) error
 
 	Group[T any] struct {
-		options    *options           // 配置
-		mu         sync.Mutex         // 锁
-		ctx        context.Context    // 上下文
-		cancelFunc context.CancelFunc // 取消函数
-		canceled   atomic.Uint32      // 是否已取消
-		errs       []error            // 错误
-		done       chan bool          // 完成信号
-		q          []T                // 任务队列
-		taskDone   int64              // 已完成任务数量
-		taskTotal  int64              // 总任务数量
-		OnMessage  func(args T) error // 任务处理
-		OnError    func(err error)    // 错误处理
+		options    *options                // 配置
+		mu         sync.Mutex              // 锁
+		ctx        context.Context         // 上下文
+		cancelFunc context.CancelFunc      // 取消函数
+		canceled   atomic.Uint32           // 是否已取消
+		errs       []error                 // 错误
+		done       chan bool               // 完成信号
+		q          []T                     // 任务队列
+		taskDone   int64                   // 已完成任务数量
+		taskTotal  int64                   // 总任务数量
+		OnMessage  func(args T) error      // 任务处理
+		OnError    func(args T, err error) // 错误处理
 	}
 )
 
@@ -53,7 +53,7 @@ func New[T any](opts ...Option) *Group[T] {
 	c.OnMessage = func(args T) error {
 		return nil
 	}
-	c.OnError = func(err error) {}
+	c.OnError = func(args T, err error) {}
 
 	return c
 }
@@ -104,7 +104,7 @@ func (c *Group[T]) do(args T) {
 		c.mu.Lock()
 		c.errs = append(c.errs, err)
 		c.mu.Unlock()
-		c.OnError(err)
+		c.OnError(args, err)
 	}
 
 	if c.incrAndIsDone() {

groups/group_test.go

@@ -96,7 +96,7 @@ func TestNewTaskGroup(t *testing.T) {
 				return nil
 			}
 		}
-		ctl.OnError = func(err error) {
+		ctl.OnError = func(args int, err error) {
 			ctl.Cancel()
 		}
 		err := ctl.Start()