Merge pull request #4 from yankooo/dev

impl pool

Author: yankooo

.gitignore

@@ -0,0 +1 @@
+.idea/

.travis.yml

@@ -0,0 +1,13 @@
+language: go
+
+go:
+  - 1.12.x
+
+before_install:
+  - go get -t -v ./...
+
+script:
+  - go test -race -coverprofile=coverage.txt -covermode=atomic
+
+after_success:
+  - bash <(curl -s https://codecov.io/bash)

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020 Michael Liu
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,2 +1,75 @@
 # wasps
-A lightweight Goroutine pool for Go
+
+[![Build Status](https://travis-ci.com/yankooo/wasps.svg?branch=master)](https://travis-ci.com/yankooo/wasps) [![Go Report Card](https://goreportcard.com/badge/github.com/yankooo/wasps)](https://goreportcard.com/report/github.com/yankooo/wasps) [![Codecov](https://img.shields.io/codecov/c/github/yankooo/wasps/master)](https://codecov.io/gh/yankooo/wasps) [![Doc for wasps](https://img.shields.io/badge/go.dev-doc-007d9c?style=flat&logo=appveyor)](https://pkg.go.dev/github.com/yankooo/wasps?tab=doc)
+
+English | [中文](README_ZH.md)
+
+## Introduction
+
+`wasps` is a lightweight goroutine pool that implements scheduling management for multiple goroutines.
+
+## Features:
+
+- Automatic scheduling goroutine.
+- Provides commonly-used interfaces: task submission, getting the number of running goroutines, dynamically adjusting the size of the pool, and releasing the pool.
+- Provide callback type goroutine pool, serialization work goroutine pool, custom work goroutine pool.
+- Support custom work goroutine, support panic processing of task goroutine, and custom pass parameter of closure function.
+- Asynchronous mechanism.
+
+## Docs
+
+https://godoc.org/github.com/yankooo/wasps
+
+## Installation
+
+``` go
+go get github.com/yankooo/wasps
+```
+
+## Use
+``` go
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/yankooo/wasps"
+	"log"
+	"sync"
+	"time"
+)
+
+func main()  {
+	pool := wasps.NewCallback(5)
+	defer func() {
+		pool.Release()
+	}()
+
+	var num = 10
+	ctx, _ := context.WithTimeout(context.TODO(), 1*time.Second)
+
+	var wg sync.WaitGroup
+
+	wg.Add(3)
+	pool.SubmitWithContext(ctx, func(args ...interface{}) {
+		defer wg.Done()
+		num := args[0].(int)
+		log.Printf("first submit %d", num*2)
+	}, wasps.WithArgs(num))
+
+	pool.Submit(func(args ...interface{}) {
+		defer wg.Done()
+		num := args[0].(int)
+		log.Printf("second submit %d", num)
+	}, wasps.WithArgs(num), wasps.WithRecoverFn(func(r interface{}) { fmt.Printf("catch panic: %+v\n", r) }))
+
+	num = 200
+	pool.Submit(func(args ...interface{}) {
+		defer wg.Done()
+		log.Printf("third submit %d", num)
+	})
+
+	wg.Wait()
+}
+```
+

README_ZH.md

@@ -0,0 +1,74 @@
+# wasps
+
+[![Build Status](https://travis-ci.com/yankooo/wasps.svg?branch=master)](https://travis-ci.com/yankooo/wasps) [![Go Report Card](https://goreportcard.com/badge/github.com/yankooo/wasps)](https://goreportcard.com/report/github.com/yankooo/wasps) [![Codecov](https://img.shields.io/codecov/c/github/yankooo/wasps/master)](https://codecov.io/gh/yankooo/wasps) [![Doc for wasps](https://img.shields.io/badge/go.dev-doc-007d9c?style=flat&logo=appveyor)](https://pkg.go.dev/github.com/yankooo/wasps?tab=doc)
+
+[英文](README.md) | 中文
+
+## 简介
+
+`wasps`是一个轻量级的 goroutine 池，实现了对多个 goroutine 的调度管理。
+
+## 功能
+
+- 自动调度goroutine。
+- 提供了常用的接口：任务提交、获取运行中的 goroutine 数量、动态调整 Pool 大小、释放 Pool。
+- 提供了回调类型的协程池、串行化工作的协程池、自定义工作协程的协程池。
+- 支持自定义工作协程，支持任务协程的panic处理以及闭包函数的自定义传参。
+- 异步机制
+
+## 文档
+
+https://godoc.org/github.com/yankooo/wasps
+
+## 安装
+
+``` go
+go get github.com/yankooo/wasps
+```
+
+## 使用
+``` go
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/yankooo/wasps"
+	"log"
+	"sync"
+	"time"
+)
+
+func main()  {
+	pool := wasps.NewCallback(5)
+	defer func() {
+		pool.Release()
+	}()
+
+	var num = 10
+	ctx, _ := context.WithTimeout(context.TODO(), 1*time.Second)
+
+	var wg sync.WaitGroup
+
+	wg.Add(3)
+	pool.SubmitWithContext(ctx, func(args ...interface{}) {
+		defer wg.Done()
+		num := args[0].(int)
+		log.Printf("first submit %d", num*2)
+	}, wasps.WithArgs(num))
+
+	pool.Submit(func(args ...interface{}) {
+		defer wg.Done()
+		num := args[0].(int)
+		log.Printf("second submit %d", num)
+	}, wasps.WithArgs(num), wasps.WithRecoverFn(func(r interface{}) { fmt.Printf("catch panic: %+v\n", r) }))
+
+	num = 200
+	pool.Submit(func(args ...interface{}) {
+		defer wg.Done()
+		log.Printf("third submit %d", num)
+	})
+
+	wg.Wait()
+}
+```

example/custom_worker.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"context"
+	"github.com/yankooo/wasps"
+	"log"
+	"time"
+)
+
+type customWorker struct {
+	job   chan *wasps.Job
+	close chan struct{}
+}
+
+type customTypeFn func() error
+
+func (c *customWorker) Do(job *wasps.Job) {
+	f := job.Task.(customTypeFn)
+	_ = f()
+}
+
+func (c *customWorker) JobChan() chan *wasps.Job {
+	return c.job
+}
+
+func (c *customWorker) StopChan() chan struct{} {
+	return c.close
+}
+
+func customWorkerExample() {
+	pool := wasps.New(5, func() wasps.Worker {
+		return &customWorker{
+			job:   make(chan *wasps.Job),
+			close: make(chan struct{}),
+		}
+	})
+	defer pool.Release()
+
+	ctx, _ := context.WithTimeout(context.TODO(), time.Second*2)
+	var task customTypeFn = func() error {
+		panic("custome")
+		return nil
+	}
+
+	pool.SubmitWithContext(ctx, task, wasps.WithRecoverFn(func(r interface{}) {
+		log.Printf("catch panic : %+v", r)
+	}))
+
+	time.Sleep(time.Second * 2)
+}

example/simple.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/yankooo/wasps"
+	"log"
+	"sync"
+	"time"
+)
+
+func main() {
+	pool := wasps.NewCallback(5)
+	defer func() {
+		pool.Release()
+	}()
+
+	var num = 10
+	ctx, _ := context.WithTimeout(context.TODO(), 1*time.Second)
+
+	var wg sync.WaitGroup
+
+	wg.Add(3)
+	pool.SubmitWithContext(ctx, func(args ...interface{}) {
+		defer wg.Done()
+		num := args[0].(int)
+		log.Printf("first submit %d", num*2)
+	}, wasps.WithArgs(num))
+
+	pool.Submit(func(args ...interface{}) {
+		defer wg.Done()
+		num := args[0].(int)
+		log.Printf("second submit %d", num)
+	}, wasps.WithArgs(num), wasps.WithRecoverFn(func(r interface{}) { fmt.Printf("catch panic: %+v\n", r) }))
+
+	num = 200
+	pool.Submit(func(args ...interface{}) {
+		defer wg.Done()
+		log.Printf("third submit %d", num)
+	})
+
+	wg.Wait()
+}

go.mod

@@ -0,0 +1,11 @@
+module github.com/yankooo/wasps
+
+go 1.12
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
+	github.com/stretchr/testify v1.6.1
+	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect
+	gopkg.in/yaml.v3 v3.0.0-20200605160147-a5ece683394c // indirect
+)

go.sum

@@ -0,0 +1,15 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
+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/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20200605160147-a5ece683394c h1:grhR+C34yXImVGp7EzNk+DTIk+323eIUWOmEevy6bDo=
+gopkg.in/yaml.v3 v3.0.0-20200605160147-a5ece683394c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

option.go

@@ -0,0 +1,71 @@
+package wasps
+
+type taskOptions struct {
+	RecoverFn func(r interface{})
+	Args      []interface{}
+}
+
+// TaskOption configures how we set up the connection.
+type TaskOption interface {
+	apply(*taskOptions)
+}
+
+// funcTaskOption wraps a function that modifies taskOptions into an
+// implementation of the TaskOption interface.
+type funcTaskOption struct {
+	f func(*taskOptions)
+}
+
+func (fdo *funcTaskOption) apply(do *taskOptions) {
+	fdo.f(do)
+}
+
+func newFuncTaskOption(f func(*taskOptions)) *funcTaskOption {
+	return &funcTaskOption{
+		f: f,
+	}
+}
+
+// WithRecoverFn returns task option for recover to catch panic
+func WithRecoverFn(f func(r interface{})) TaskOption {
+	return newFuncTaskOption(func(o *taskOptions) {
+		o.RecoverFn = f
+	})
+}
+
+// WithArgs returns task option for callback func args
+func WithArgs(args ...interface{}) TaskOption {
+	return newFuncTaskOption(func(o *taskOptions) {
+		o.Args = args
+	})
+}
+
+var defaultTaskOptions = &taskOptions{
+	RecoverFn: func(r interface{}) {},
+}
+
+type poolOptions struct {
+}
+
+// PoolOption configures how we set up the connection.
+type PoolOption interface {
+	apply(*poolOptions)
+}
+
+// funcPoolOption wraps a function that modifies poolOptions into an
+// implementation of the PoolOption interface.
+type funcPoolOption struct {
+	f func(*poolOptions)
+}
+
+func (fdo *funcPoolOption) apply(do *poolOptions) {
+	fdo.f(do)
+}
+
+func newFuncPoolOption(f func(*poolOptions)) *funcPoolOption {
+	return &funcPoolOption{
+		f: f,
+	}
+}
+
+var defaultPoolOptions = &poolOptions{}