release v2.0.5

README.md

@@ -4,7 +4,7 @@
 <br/>
 
 ![Go version](https://img.shields.io/badge/go-%3E%3D1.16<recommend>-9cf)
-[![Release](https://img.shields.io/badge/release-2.0.4-green.svg)](https://github.com/duke-git/lancet/releases)
+[![Release](https://img.shields.io/badge/release-2.0.5-green.svg)](https://github.com/duke-git/lancet/releases)
 [![GoDoc](https://godoc.org/github.com/duke-git/lancet/v2?status.svg)](https://pkg.go.dev/github.com/duke-git/lancet/v2)
 [![Go Report Card](https://goreportcard.com/badge/github.com/duke-git/lancet/v2)](https://goreportcard.com/report/github.com/duke-git/lancet/v2)
 [![test](https://github.com/duke-git/lancet/actions/workflows/codecov.yml/badge.svg?branch=main&event=push)](https://github.com/duke-git/lancet/actions/workflows/codecov.yml)
@@ -23,7 +23,7 @@ English | [简体中文](./README_zh-CN.md)
 ## Feature
 
 - 👏 Comprehensive, efficient and reusable.
-- 💪 250+ go util functions, support string, slice, datetime, net, crypt...
+- 💪 300+ go util functions, support string, slice, datetime, net, crypt...
 - 💅 Only depend on the go standard library.
 - 🌍 Unit test for every exported function.
 
@@ -89,6 +89,24 @@ import "github.com/duke-git/lancet/v2/algorithm"
 - [LRUCache](https://github.com/duke-git/lancet/blob/main/docs/algorithm.md#LRUCache)
 
 
+
+### Concurrency package contain some functions to support concurrent programming. eg, goroutine, channel, async.
+
+```go
+import "github.com/duke-git/lancet/v2/concurrency"
+```
+#### Function list:
+- [NewChannel](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#NewChannel)
+- [Bridge](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#Bridge)
+- [FanIn](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#FanIn)
+- [Generate](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#Generate)
+- [Or](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#Or)
+- [OrDone](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#OrDone)
+- [Repeat](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#Repeat)
+- [RepeatFn](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#RepeatFn)
+- [Take](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#Take)
+- [Tee](https://github.com/duke-git/lancet/blob/main/docs/concurrency.md#Tee)
+
 ### Convertor package contains some functions for data convertion.
 
 ```go

README_zh-CN.md

@@ -4,7 +4,7 @@
 <br/>
 
 ![Go version](https://img.shields.io/badge/go-%3E%3D1.16<recommend>-9cf)
-[![Release](https://img.shields.io/badge/release-2.0.4-green.svg)](https://github.com/duke-git/lancet/releases)
+[![Release](https://img.shields.io/badge/release-2.0.5-green.svg)](https://github.com/duke-git/lancet/releases)
 [![GoDoc](https://godoc.org/github.com/duke-git/lancet/v2?status.svg)](https://pkg.go.dev/github.com/duke-git/lancet/v2)
 [![Go Report Card](https://goreportcard.com/badge/github.com/duke-git/lancet/v2)](https://goreportcard.com/report/github.com/duke-git/lancet/v2)
 [![test](https://github.com/duke-git/lancet/actions/workflows/codecov.yml/badge.svg?branch=main&event=push)](https://github.com/duke-git/lancet/actions/workflows/codecov.yml)
@@ -23,7 +23,7 @@
 ## 特性
 
 - 👏 全面、高效、可复用
-- 💪 250+常用go工具函数，支持string、slice、datetime、net、crypt...
+- 💪 300+常用go工具函数，支持string、slice、datetime、net、crypt...
 - 💅 只依赖go标准库
 - 🌍 所有导出函数单元测试覆盖率100%
 
@@ -88,6 +88,24 @@ import "github.com/duke-git/lancet/v2/algorithm"
 - [LinearSearch](https://github.com/duke-git/lancet/blob/main/docs/algorithm_zh-CN.md#LinearSearch)
 - [LRUCache](https://github.com/duke-git/lancet/blob/main/docs/algorithm_zh-CN.md#LRUCache)
 
+
+### 并发包包含一些支持并发编程的功能。例如：goroutine, channel, async等。
+
+```go
+import "github.com/duke-git/lancet/v2/concurrency"
+```
+#### Function list:
+- [NewChannel](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#NewChannel)
+- [Bridge](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#Bridge)
+- [FanIn](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#FanIn)
+- [Generate](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#Generate)
+- [Or](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#Or)
+- [OrDone](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#OrDone)
+- [Repeat](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#Repeat)
+- [RepeatFn](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#RepeatFn)
+- [Take](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#Take)
+- [Tee](https://github.com/duke-git/lancet/blob/main/docs/concurrency_zh-CN.md#Tee)
+  
 ### convertor转换器包支持一些常见的数据类型转换。
 
 ```go

concurrency/channel.go

@@ -0,0 +1,243 @@
+// Copyright 2021 dudaodong@gmail.com. All rights reserved.
+// Use of this source code is governed by MIT license
+
+// Package concurrency contain some functions to support concurrent programming. eg, goroutine, channel, async.
+package concurrency
+
+import (
+	"context"
+	"sync"
+)
+
+// Channel is a logic object which can generate or manipulate go channel
+// all methods of Channel are in the book tilted《Concurrency in Go》
+type Channel struct {
+}
+
+// NewChannel return a Channel instance
+func NewChannel() *Channel {
+	return &Channel{}
+}
+
+// Generate a data of type any chan, put param `values` into the chan
+func (c *Channel) Generate(ctx context.Context, values ...any) <-chan any {
+	dataStream := make(chan any)
+
+	go func() {
+		defer close(dataStream)
+
+		for _, v := range values {
+			select {
+			case <-ctx.Done():
+				return
+			case dataStream <- v:
+			}
+		}
+	}()
+
+	return dataStream
+}
+
+// Repeat return a data of type any chan, put param `values` into the chan repeatly until cancel the context.
+func (c *Channel) Repeat(ctx context.Context, values ...any) <-chan any {
+	dataStream := make(chan any)
+
+	go func() {
+		defer close(dataStream)
+		for {
+			for _, v := range values {
+				select {
+				case <-ctx.Done():
+					return
+				case dataStream <- v:
+				}
+			}
+		}
+	}()
+	return dataStream
+}
+
+// RepeatFn return a chan, excutes fn repeatly, and put the result into retruned chan
+// until close the `done` channel
+func (c *Channel) RepeatFn(ctx context.Context, fn func() any) <-chan any {
+	dataStream := make(chan any)
+
+	go func() {
+		defer close(dataStream)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case dataStream <- fn():
+			}
+		}
+	}()
+	return dataStream
+}
+
+// Take return a chan whose values are tahken from another chan
+func (c *Channel) Take(ctx context.Context, valueStream <-chan any, number int) <-chan any {
+	takeStream := make(chan any)
+
+	go func() {
+		defer close(takeStream)
+
+		for i := 0; i < number; i++ {
+			select {
+			case <-ctx.Done():
+				return
+			case takeStream <- <-valueStream:
+			}
+		}
+	}()
+
+	return takeStream
+}
+
+// FanIn merge multiple channels into one channel
+func (c *Channel) FanIn(ctx context.Context, channels ...<-chan any) <-chan any {
+	out := make(chan any)
+
+	go func() {
+		var wg sync.WaitGroup
+		wg.Add(len(channels))
+
+		for _, c := range channels {
+			go func(c <-chan any) {
+				defer wg.Done()
+				for v := range c {
+					select {
+					case <-ctx.Done():
+						return
+					case out <- v:
+					}
+				}
+			}(c)
+		}
+		wg.Wait()
+		close(out)
+	}()
+
+	return out
+}
+
+// Tee split one chanel into two channels
+func (c *Channel) Tee(ctx context.Context, in <-chan any) (<-chan any, <-chan any) {
+	out1 := make(chan any)
+	out2 := make(chan any)
+
+	go func() {
+		defer close(out1)
+		defer close(out2)
+
+		for val := range c.OrDone(ctx, in) {
+			var out1, out2 = out1, out2
+			for i := 0; i < 2; i++ {
+				select {
+				case <-ctx.Done():
+				case out1 <- val:
+					out1 = nil
+				case out2 <- val:
+					out2 = nil
+				}
+			}
+		}
+	}()
+
+	return out1, out2
+}
+
+// Bridge link multiply channels into one channel
+func (c *Channel) Bridge(ctx context.Context, chanStream <-chan <-chan any) <-chan any {
+	valStream := make(chan any)
+
+	go func() {
+		defer close(valStream)
+
+		for {
+			var stream <-chan any
+			select {
+			case maybeStream, ok := <-chanStream:
+				if ok == false {
+					return
+				}
+				stream = maybeStream
+			case <-ctx.Done():
+				return
+			}
+
+			for val := range c.OrDone(ctx, stream) {
+				select {
+				case valStream <- val:
+				case <-ctx.Done():
+				}
+			}
+		}
+	}()
+
+	return valStream
+}
+
+// Or read one or more channels into one channel, will close when any readin channel is closed
+func (c *Channel) Or(channels ...<-chan any) <-chan any {
+	switch len(channels) {
+	case 0:
+		return nil
+	case 1:
+		return channels[0]
+	}
+
+	orDone := make(chan any)
+
+	go func() {
+		defer close(orDone)
+
+		switch len(channels) {
+		case 2:
+			select {
+			case <-channels[0]:
+			case <-channels[1]:
+			}
+		default:
+			m := len(channels) / 2
+			select {
+			case <-c.Or(channels[:m]...):
+			case <-c.Or(channels[m:]...):
+			}
+			// select {
+			// case <-channels[0]:
+			// case <-channels[1]:
+			// case <-channels[2]:
+			// case <-c.Or(append(channels[3:], orDone)...):
+			// }
+		}
+	}()
+
+	return orDone
+}
+
+// OrDone read a channel into another channel, will close until cancel context.
+func (c *Channel) OrDone(ctx context.Context, channel <-chan any) <-chan any {
+	valStream := make(chan any)
+
+	go func() {
+		defer close(valStream)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case v, ok := <-channel:
+				if !ok {
+					return
+				}
+				select {
+				case valStream <- v:
+				case <-ctx.Done():
+				}
+			}
+		}
+
+	}()
+
+	return valStream
+}

concurrency/channel_test.go

@@ -0,0 +1,206 @@
+package concurrency
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/duke-git/lancet/v2/internal"
+)
+
+func TestGenerate(t *testing.T) {
+	assert := internal.NewAssert(t, "TestGenerate")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := NewChannel()
+	intStream := c.Generate(ctx, 1, 2, 3)
+
+	// for v := range intStream {
+	// 	t.Log(v) //1, 2, 3
+	// }
+	assert.Equal(1, <-intStream)
+	assert.Equal(2, <-intStream)
+	assert.Equal(3, <-intStream)
+}
+
+func TestRepeat(t *testing.T) {
+	assert := internal.NewAssert(t, "TestRepeat")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := NewChannel()
+	intStream := c.Take(ctx, c.Repeat(ctx, 1, 2), 5)
+
+	// for v := range intStream {
+	// 	t.Log(v) //1, 2, 1, 2, 1
+	// }
+	assert.Equal(1, <-intStream)
+	assert.Equal(2, <-intStream)
+	assert.Equal(1, <-intStream)
+	assert.Equal(2, <-intStream)
+	assert.Equal(1, <-intStream)
+}
+
+func TestRepeatFn(t *testing.T) {
+	assert := internal.NewAssert(t, "TestRepeatFn")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	fn := func() any {
+		s := "a"
+		return s
+	}
+	c := NewChannel()
+	dataStream := c.Take(ctx, c.RepeatFn(ctx, fn), 3)
+
+	// for v := range dataStream {
+	// 	t.Log(v) //a, a, a
+	// }
+
+	assert.Equal("a", <-dataStream)
+	assert.Equal("a", <-dataStream)
+	assert.Equal("a", <-dataStream)
+}
+
+func TestTake(t *testing.T) {
+	assert := internal.NewAssert(t, "TestTake")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	numbers := make(chan any, 5)
+	numbers <- 1
+	numbers <- 2
+	numbers <- 3
+	numbers <- 4
+	numbers <- 5
+	defer close(numbers)
+
+	c := NewChannel()
+	intStream := c.Take(ctx, numbers, 3)
+
+	assert.Equal(1, <-intStream)
+	assert.Equal(2, <-intStream)
+	assert.Equal(3, <-intStream)
+}
+
+func TestFanIn(t *testing.T) {
+	assert := internal.NewAssert(t, "TestFanIn")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := NewChannel()
+	channels := make([]<-chan any, 3)
+
+	for i := 0; i < 3; i++ {
+		channels[i] = c.Take(ctx, c.Repeat(ctx, i), 3)
+	}
+
+	mergedChannel := c.FanIn(ctx, channels...)
+
+	for val := range mergedChannel {
+		t.Logf("\t%d\n", val)
+	}
+
+	assert.Equal(1, 1)
+}
+
+func TestOr(t *testing.T) {
+	assert := internal.NewAssert(t, "TestOr")
+
+	sig := func(after time.Duration) <-chan any {
+		c := make(chan interface{})
+		go func() {
+			defer close(c)
+			time.Sleep(after)
+		}()
+		return c
+	}
+
+	start := time.Now()
+
+	c := NewChannel()
+	<-c.Or(
+		sig(1*time.Second),
+		sig(2*time.Second),
+		sig(3*time.Second),
+		sig(4*time.Second),
+		sig(5*time.Second),
+	)
+
+	t.Logf("done after %v", time.Since(start))
+
+	assert.Equal(1, 1)
+}
+
+func TestOrDone(t *testing.T) {
+	assert := internal.NewAssert(t, "TestOrDone")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := NewChannel()
+	intStream := c.Take(ctx, c.Repeat(ctx, 1), 3)
+
+	var res any
+	for val := range c.OrDone(ctx, intStream) {
+		t.Logf("%v", val)
+		res = val
+	}
+
+	assert.Equal(1, res)
+}
+
+func TestTee(t *testing.T) {
+	assert := internal.NewAssert(t, "TestTee")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := NewChannel()
+	inStream := c.Take(ctx, c.Repeat(ctx, 1), 4)
+
+	out1, out2 := c.Tee(ctx, inStream)
+	for val := range out1 {
+		val1 := val
+		val2 := <-out2
+		// t.Log("val1 is", val1)
+		// t.Log("val2 is", val2)
+		assert.Equal(1, val1)
+		assert.Equal(1, val2)
+	}
+}
+
+func TestBridge(t *testing.T) {
+	assert := internal.NewAssert(t, "TestBridge")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := NewChannel()
+	genVals := func() <-chan <-chan any {
+		chanStream := make(chan (<-chan any))
+		go func() {
+			defer close(chanStream)
+			for i := 0; i < 10; i++ {
+				stream := make(chan any, 1)
+				stream <- i
+				close(stream)
+				chanStream <- stream
+			}
+		}()
+		return chanStream
+	}
+
+	index := 0
+	for val := range c.Bridge(ctx, genVals()) {
+		// t.Logf("%v ", val) //0 1 2 3 4 5 6 7 8 9
+		assert.Equal(index, val)
+		index++
+	}
+}

docs/concurrency.md

@@ -0,0 +1,390 @@
+# Concurrency
+Package concurrency contain some functions to support concurrent programming. eg, goroutine, channel, async.
+
+<div STYLE="page-break-after: always;"></div>
+
+## Source:
+
+- [https://github.com/duke-git/lancet/blob/main/concurrency/channel.go](https://github.com/duke-git/lancet/blob/main/concurrency/channel.go)
+
+<div STYLE="page-break-after: always;"></div>
+
+## Usage:
+```go
+import (
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+```
+
+<div STYLE="page-break-after: always;"></div>
+
+## Index
+### Channel
+- [NewChannel](#NewChannel)
+- [Bridge](#Bridge)
+- [FanIn](#FanIn)
+- [Generate](#Generate)
+- [Or](#Or)
+- [OrDone](#OrDone)
+- [Repeat](#Repeat)
+- [RepeatFn](#RepeatFn)
+- [Take](#Take)
+- [Tee](#Tee)
+
+<div STYLE="page-break-after: always;"></div>
+
+## Documentation
+
+
+## Channel
+### <span id="NewChannel">NewChannel</span>
+<p>return a Channel pointer instance.</p>
+
+<b>Signature:</b>
+
+```go
+type Channel struct {}
+func NewChannel() *Channel
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    c := concurrency.NewChannel()
+}
+```
+
+
+
+### <span id="Bridge">Bridge</span>
+
+<p>Link multiple channels into one channel until cancel the context.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) Bridge(ctx context.Context, chanStream <-chan <-chan any) <-chan any
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	genVals := func() <-chan <-chan any {
+		chanStream := make(chan (<-chan any))
+		go func() {
+			defer close(chanStream)
+			for i := 0; i < 10; i++ {
+				stream := make(chan any, 1)
+				stream <- i
+				close(stream)
+				chanStream <- stream
+			}
+		}()
+		return chanStream
+	}
+
+	index := 0
+	for val := range c.Bridge(ctx, genVals()) {
+		fmt.Printf("%v ", val) //0 1 2 3 4 5 6 7 8 9
+	}
+}
+```
+
+
+
+
+### <span id="FanIn">FanIn</span>
+
+<p>merge multiple channels into one channel until cancel the context.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) FanIn(ctx context.Context, channels ...<-chan any) <-chan any
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	channels := make([]<-chan any, 3)
+
+	for i := 0; i < 3; i++ {
+		channels[i] = c.Take(ctx, c.Repeat(ctx, i), 3)
+	}
+
+	mergedChannel := c.FanIn(ctx, channels...)
+
+	for val := range mergedChannel {
+		fmt.Println("\t%d\n", val) //1,2,1,0,0,1,0,2,2 (order not for sure)
+	}
+}
+```
+
+
+### <span id="Repeat">Repeat</span>
+
+<p>Return a chan, put param `values` into the chan repeatly until cancel the context.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) Repeat(ctx context.Context, values ...any) <-chan any
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	intStream := c.Take(ctx, c.Repeat(ctx, 1, 2), 5)
+
+	for v := range intStream {
+     	fmt.Println(v) //1, 2, 1, 2, 1
+	}
+}
+```
+
+
+
+
+### <span id="RepeatFn">RepeatFn</span>
+
+<p>Return a chan, excutes fn repeatly, and put the result into retruned chan until cancel context.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) RepeatFn(ctx context.Context, fn func() any) <-chan any
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	fn := func() any {
+		s := "a"
+		return s
+	}
+	c := concurrency.NewChannel()
+	dataStream := c.Take(ctx, c.RepeatFn(ctx, fn), 3)
+
+	for v := range dataStream {
+	    fmt.Println(v) //a, a, a
+	}
+}
+```
+
+
+
+### <span id="Or">Or</span>
+
+<p>Read one or more channels into one channel, will close when any readin channel is closed.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) Or(channels ...<-chan any) <-chan any
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    sig := func(after time.Duration) <-chan any {
+		c := make(chan interface{})
+		go func() {
+			defer close(c)
+			time.Sleep(after)
+		}()
+		return c
+	}
+
+	start := time.Now()
+
+	c := concurrency.NewChannel()
+	<-c.Or(
+		sig(1*time.Second),
+		sig(2*time.Second),
+		sig(3*time.Second),
+		sig(4*time.Second),
+		sig(5*time.Second),
+	)
+
+	fmt.Println("done after %v", time.Since(start)) //1.003s
+}
+```
+
+
+
+
+### <span id="OrDone">OrDone</span>
+
+<p>Read a channel into another channel, will close until cancel context.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) OrDone(ctx context.Context, channel <-chan any) <-chan any
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	intStream := c.Take(ctx, c.Repeat(ctx, 1), 3)
+
+	for val := range c.OrDone(ctx, intStream) {
+		fmt.Println(val)  //1
+	}
+}
+```
+
+
+
+
+### <span id="Take">Take</span>
+
+<p>Return a chan whose values are tahken from another chan until cancel context.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) Take(ctx context.Context, valueStream <-chan any, number int) <-chan any
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	numbers := make(chan any, 5)
+	numbers <- 1
+	numbers <- 2
+	numbers <- 3
+	numbers <- 4
+	numbers <- 5
+	defer close(numbers)
+
+	c := concurrency.NewChannel()
+	intStream := c.Take(ctx, numbers, 3)
+
+	for val := range intStream {
+        fmt.Println(val) //1, 2, 3
+	}
+}
+```
+
+
+
+### <span id="Tee">Tee</span>
+
+<p>Split one chanel into two channels until cancel context.</p>
+
+<b>Signature:</b>
+
+```go
+func (c *Channel) Tee(ctx context.Context, in <-chan any) (<-chan any, <-chan any)
+```
+<b>Example:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	inStream := c.Take(ctx, c.Repeat(ctx, 1, 2), 4)
+
+	out1, out2 := c.Tee(ctx, inStream)
+	for val := range out1 {
+        fmt.Println(val) //1
+        fmt.Println(<-out2) //1
+	}
+}
+```

docs/concurrency_zh-CN.md

@@ -0,0 +1,390 @@
+# Concurrency
+并发包包含一些支持并发编程的功能。例如：goroutine, channel, async等。
+
+<div STYLE="page-break-after: always;"></div>
+
+## 源码:
+
+- [https://github.com/duke-git/lancet/blob/main/concurrency/channel.go](https://github.com/duke-git/lancet/blob/main/concurrency/channel.go)
+
+<div STYLE="page-break-after: always;"></div>
+
+## 用法:
+```go
+import (
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+```
+
+<div STYLE="page-break-after: always;"></div>
+
+## 目录
+### Channel
+- [NewChannel](#NewChannel)
+- [Bridge](#Bridge)
+- [FanIn](#FanIn)
+- [Generate](#Generate)
+- [Or](#Or)
+- [OrDone](#OrDone)
+- [Repeat](#Repeat)
+- [RepeatFn](#RepeatFn)
+- [Take](#Take)
+- [Tee](#Tee)
+
+<div STYLE="page-break-after: always;"></div>
+
+## 文档
+
+
+### Channel
+### <span id="NewChannel">NewChannel</span>
+<p>返回一个 Channel 指针实例</p>
+
+<b>函数签名:</b>
+
+```go
+type Channel struct {}
+func NewChannel() *Channel
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    c := concurrency.NewChannel()
+}
+```
+
+
+
+### <span id="Bridge">Bridge</span>
+
+<p>将多个通道链接到一个通道，直到取消上下文。</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) Bridge(ctx context.Context, chanStream <-chan <-chan any) <-chan any
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	genVals := func() <-chan <-chan any {
+		chanStream := make(chan (<-chan any))
+		go func() {
+			defer close(chanStream)
+			for i := 0; i < 10; i++ {
+				stream := make(chan any, 1)
+				stream <- i
+				close(stream)
+				chanStream <- stream
+			}
+		}()
+		return chanStream
+	}
+
+	index := 0
+	for val := range c.Bridge(ctx, genVals()) {
+		fmt.Printf("%v ", val) //0 1 2 3 4 5 6 7 8 9
+	}
+}
+```
+
+
+
+
+### <span id="FanIn">FanIn</span>
+
+<p>将多个通道合并为一个通道，直到取消上下文</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) FanIn(ctx context.Context, channels ...<-chan any) <-chan any
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	channels := make([]<-chan any, 3)
+
+	for i := 0; i < 3; i++ {
+		channels[i] = c.Take(ctx, c.Repeat(ctx, i), 3)
+	}
+
+	mergedChannel := c.FanIn(ctx, channels...)
+
+	for val := range mergedChannel {
+		fmt.Println("\t%d\n", val) //1,2,1,0,0,1,0,2,2 (order not for sure)
+	}
+}
+```
+
+
+### <span id="Repeat">Repeat</span>
+
+<p>返回一个chan，将参数`values`重复放入chan，直到取消上下文。</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) Repeat(ctx context.Context, values ...any) <-chan any
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	intStream := c.Take(ctx, c.Repeat(ctx, 1, 2), 5)
+
+	for v := range intStream {
+     	fmt.Println(v) //1, 2, 1, 2, 1
+	}
+}
+```
+
+
+
+
+### <span id="RepeatFn">RepeatFn</span>
+
+<p>返回一个chan，重复执行函数fn，并将结果放入返回的chan，直到取消上下文。</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) RepeatFn(ctx context.Context, fn func() any) <-chan any
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	fn := func() any {
+		s := "a"
+		return s
+	}
+	c := concurrency.NewChannel()
+	dataStream := c.Take(ctx, c.RepeatFn(ctx, fn), 3)
+
+	for v := range dataStream {
+	    fmt.Println(v) //a, a, a
+	}
+}
+```
+
+
+
+### <span id="Or">Or</span>
+
+<p>将一个或多个通道读取到一个通道中，当任何读取通道关闭时将结束读取。</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) Or(channels ...<-chan any) <-chan any
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    sig := func(after time.Duration) <-chan any {
+		c := make(chan interface{})
+		go func() {
+			defer close(c)
+			time.Sleep(after)
+		}()
+		return c
+	}
+
+	start := time.Now()
+
+	c := concurrency.NewChannel()
+	<-c.Or(
+		sig(1*time.Second),
+		sig(2*time.Second),
+		sig(3*time.Second),
+		sig(4*time.Second),
+		sig(5*time.Second),
+	)
+
+	fmt.Println("done after %v", time.Since(start)) //1.003s
+}
+```
+
+
+
+
+### <span id="OrDone">OrDone</span>
+
+<p>将一个通道读入另一个通道，直到取消上下文。</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) OrDone(ctx context.Context, channel <-chan any) <-chan any
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	intStream := c.Take(ctx, c.Repeat(ctx, 1), 3)
+
+	for val := range c.OrDone(ctx, intStream) {
+		fmt.Println(val)  //1
+	}
+}
+```
+
+
+
+
+### <span id="Take">Take</span>
+
+<p>返回一个chan，其值从另一个chan获取，直到取消上下文。</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) Take(ctx context.Context, valueStream <-chan any, number int) <-chan any
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	numbers := make(chan any, 5)
+	numbers <- 1
+	numbers <- 2
+	numbers <- 3
+	numbers <- 4
+	numbers <- 5
+	defer close(numbers)
+
+	c := concurrency.NewChannel()
+	intStream := c.Take(ctx, numbers, 3)
+
+	for val := range intStream {
+        fmt.Println(val) //1, 2, 3
+	}
+}
+```
+
+
+
+### <span id="Tee">Tee</span>
+
+<p>将一个通道分成两个通道，直到取消上下文。</p>
+
+<b>函数签名:</b>
+
+```go
+func (c *Channel) Tee(ctx context.Context, in <-chan any) (<-chan any, <-chan any)
+```
+<b>例子:</b>
+
+```go
+package main
+
+import (
+    "context"
+    "fmt"
+    "github.com/duke-git/lancet/v2/concurrency"
+)
+
+func main() {
+    ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := concurrency.NewChannel()
+	inStream := c.Take(ctx, c.Repeat(ctx, 1, 2), 4)
+
+	out1, out2 := c.Tee(ctx, inStream)
+	for val := range out1 {
+        fmt.Println(val) //1
+        fmt.Println(<-out2) //1
+	}
+}
+```