docs: update roadmap (#1405)

* feat: support tls for etcd client

* chore: fix typo

* refactor: rename TrustedCAFile to CACertFile

* docs: add comments

* fix: missing tls registration

* feat: add InsecureSkipVerify config for testing

.github/workflows/release.yaml

@@ -1,30 +0,0 @@
-on:
-  release:
-    types: [created]
-
-jobs:
-  releases-matrix:
-    name: Release goctl binary
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        # build and publish in parallel: linux/386, linux/amd64, linux/arm64,
-        # windows/386, windows/amd64, windows/arm64, darwin/amd64, darwin/arm64
-        goos: [linux, windows, darwin]
-        goarch: ["386", amd64, arm64]
-        exclude:
-          - goarch: "386"
-            goos: darwin
-          - goarch: "386"
-            goos: windows
-    steps:
-      - uses: actions/checkout@v2
-      - uses: wangyoucao577/go-release-action@v1.22
-        with:
-          github_token: ${{ secrets.GITHUB_TOKEN }}
-          goos: ${{ matrix.goos }}
-          goarch: ${{ matrix.goarch }}
-          goversion: "https://dl.google.com/go/go1.17.5.linux-amd64.tar.gz"
-          project_path: "tools/goctl"
-          binary_name: "goctl"
-          extra_files: tools/goctl/goctl.md

ROADMAP.md

@@ -14,8 +14,10 @@ We hope that the items listed below will inspire further engagement from the com
 
 ## 2021 Q4
 - [x] Support `username/password` authentication in ETCD
+- [x] Support `SSL/TLS` in ETCD
 - [x] Support `SSL/TLS` in `zRPC`
 - [x] Support `TLS` in redis connections
+- [x] Support `goctl bug` to report bugs conveniently
 
 ## 2022
 - [ ] Support `goctl mock` command to start a mocking server with given `.api` file

core/discov/accountregistry.go

@@ -2,6 +2,13 @@ package discov
 
 import "github.com/tal-tech/go-zero/core/discov/internal"
 
+// RegisterAccount registers the username/password to the given etcd cluster.
 func RegisterAccount(endpoints []string, user, pass string) {
 	internal.AddAccount(endpoints, user, pass)
 }
+
+// RegisterTLS registers the CertFile/CertKeyFile/CACertFile to the given etcd.
+func RegisterTLS(endpoints []string, certFile, certKeyFile, caFile string,
+	insecureSkipVerify bool) error {
+	return internal.AddTLS(endpoints, certFile, certKeyFile, caFile, insecureSkipVerify)
+}

core/discov/config.go

@@ -4,10 +4,14 @@ import "errors"
 
 // EtcdConf is the config item with the given key on etcd.
 type EtcdConf struct {
-	Hosts []string
-	Key   string
-	User  string `json:",optional"`
-	Pass  string `json:",optional"`
+	Hosts              []string
+	Key                string
+	User               string `json:",optional"`
+	Pass               string `json:",optional"`
+	CertFile           string `json:",optional"`
+	CertKeyFile        string `json:",optional=CertFile"`
+	CACertFile         string `json:",optional=CertFile"`
+	InsecureSkipVerify bool   `json:",optional"`
 }
 
 // HasAccount returns if account provided.
@@ -15,6 +19,11 @@ func (c EtcdConf) HasAccount() bool {
 	return len(c.User) > 0 && len(c.Pass) > 0
 }
 
+// HasTLS returns if TLS CertFile/CertKeyFile/CACertFile are provided.
+func (c EtcdConf) HasTLS() bool {
+	return len(c.CertFile) > 0 && len(c.CertKeyFile) > 0 && len(c.CACertFile) > 0
+}
+
 // Validate validates c.
 func (c EtcdConf) Validate() error {
 	if len(c.Hosts) == 0 {

core/discov/internal/accountmanager.go

@@ -1,17 +1,25 @@
 package internal
 
-import "sync"
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"io/ioutil"
+	"sync"
+)
 
+var (
+	accounts   = make(map[string]Account)
+	tlsConfigs = make(map[string]*tls.Config)
+	lock       sync.RWMutex
+)
+
+// Account holds the username/password for an etcd cluster.
 type Account struct {
 	User string
 	Pass string
 }
 
-var (
-	accounts = make(map[string]Account)
-	lock     sync.RWMutex
-)
-
+// AddAccount adds the username/password for the given etcd cluster.
 func AddAccount(endpoints []string, user, pass string) {
 	lock.Lock()
 	defer lock.Unlock()
@@ -22,6 +30,33 @@ func AddAccount(endpoints []string, user, pass string) {
 	}
 }
 
+// AddTLS adds the tls cert files for the given etcd cluster.
+func AddTLS(endpoints []string, certFile, certKeyFile, caFile string, insecureSkipVerify bool) error {
+	cert, err := tls.LoadX509KeyPair(certFile, certKeyFile)
+	if err != nil {
+		return err
+	}
+
+	caData, err := ioutil.ReadFile(caFile)
+	if err != nil {
+		return err
+	}
+
+	pool := x509.NewCertPool()
+	pool.AppendCertsFromPEM(caData)
+
+	lock.Lock()
+	defer lock.Unlock()
+	tlsConfigs[getClusterKey(endpoints)] = &tls.Config{
+		Certificates:       []tls.Certificate{cert},
+		RootCAs:            pool,
+		InsecureSkipVerify: insecureSkipVerify,
+	}
+
+	return nil
+}
+
+// GetAccount gets the username/password for the given etcd cluster.
 func GetAccount(endpoints []string) (Account, bool) {
 	lock.RLock()
 	defer lock.RUnlock()
@@ -29,3 +64,12 @@ func GetAccount(endpoints []string) (Account, bool) {
 	account, ok := accounts[getClusterKey(endpoints)]
 	return account, ok
 }
+
+// GetTLS gets the tls config for the given etcd cluster.
+func GetTLS(endpoints []string) (*tls.Config, bool) {
+	lock.RLock()
+	defer lock.RUnlock()
+
+	cfg, ok := tlsConfigs[getClusterKey(endpoints)]
+	return cfg, ok
+}

core/discov/internal/registry.go

@@ -337,6 +337,9 @@ func DialClient(endpoints []string) (EtcdClient, error) {
 		cfg.Username = account.User
 		cfg.Password = account.Pass
 	}
+	if tlsCfg, ok := GetTLS(endpoints); ok {
+		cfg.TLS = tlsCfg
+	}
 
 	return clientv3.New(cfg)
 }

core/discov/publisher.go

@@ -145,16 +145,23 @@ func (p *Publisher) revoke(cli internal.EtcdClient) {
 	}
 }
 
-// WithPubEtcdAccount provides the etcd username/password.
-func WithPubEtcdAccount(user, pass string) PubOption {
-	return func(pub *Publisher) {
-		internal.AddAccount(pub.endpoints, user, pass)
-	}
-}
-
 // WithId customizes a Publisher with the id.
 func WithId(id int64) PubOption {
 	return func(publisher *Publisher) {
 		publisher.id = id
 	}
 }
+
+// WithPubEtcdAccount provides the etcd username/password.
+func WithPubEtcdAccount(user, pass string) PubOption {
+	return func(pub *Publisher) {
+		RegisterAccount(pub.endpoints, user, pass)
+	}
+}
+
+// WithPubEtcdTLS provides the etcd CertFile/CertKeyFile/CACertFile.
+func WithPubEtcdTLS(certFile, certKeyFile, caFile string, insecureSkipVerify bool) PubOption {
+	return func(pub *Publisher) {
+		logx.Must(RegisterTLS(pub.endpoints, certFile, certKeyFile, caFile, insecureSkipVerify))
+	}
+}

core/discov/subscriber.go

@@ -5,6 +5,7 @@ import (
 	"sync/atomic"
 
 	"github.com/tal-tech/go-zero/core/discov/internal"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/syncx"
 )
 
@@ -58,9 +59,17 @@ func Exclusive() SubOption {
 	}
 }
 
+// WithSubEtcdAccount provides the etcd username/password.
 func WithSubEtcdAccount(user, pass string) SubOption {
 	return func(sub *Subscriber) {
-		internal.AddAccount(sub.endpoints, user, pass)
+		RegisterAccount(sub.endpoints, user, pass)
+	}
+}
+
+// WithSubEtcdTLS provides the etcd CertFile/CertKeyFile/CACertFile.
+func WithSubEtcdTLS(certFile, certKeyFile, caFile string, insecureSkipVerify bool) SubOption {
+	return func(sub *Subscriber) {
+		logx.Must(RegisterTLS(sub.endpoints, certFile, certKeyFile, caFile, insecureSkipVerify))
 	}
 }
 

core/fx/stream.go

@@ -90,6 +90,8 @@ func Range(source <-chan interface{}) Stream {
 func (s Stream) AllMach(predicate func(item interface{}) bool) bool {
 	for item := range s.source {
 		if !predicate(item) {
+			// make sure the former goroutine not block, and current func returns fast.
+			go drain(s.source)
 			return false
 		}
 	}
@@ -103,6 +105,8 @@ func (s Stream) AllMach(predicate func(item interface{}) bool) bool {
 func (s Stream) AnyMach(predicate func(item interface{}) bool) bool {
 	for item := range s.source {
 		if predicate(item) {
+			// make sure the former goroutine not block, and current func returns fast.
+			go drain(s.source)
 			return true
 		}
 	}
@@ -186,8 +190,7 @@ func (s Stream) Distinct(fn KeyFunc) Stream {
 
 // Done waits all upstreaming operations to be done.
 func (s Stream) Done() {
-	for range s.source {
-	}
+	drain(s.source)
 }
 
 // Filter filters the items by the given FilterFunc.
@@ -199,9 +202,22 @@ func (s Stream) Filter(fn FilterFunc, opts ...Option) Stream {
 	}, opts...)
 }
 
+// First returns the first item, nil if no items.
+func (s Stream) First() interface{} {
+	for item := range s.source {
+		// make sure the former goroutine not block, and current func returns fast.
+		go drain(s.source)
+		return item
+	}
+
+	return nil
+}
+
 // ForAll handles the streaming elements from the source and no later streams.
 func (s Stream) ForAll(fn ForAllFunc) {
 	fn(s.source)
+	// avoid goroutine leak on fn not consuming all items.
+	go drain(s.source)
 }
 
 // ForEach seals the Stream with the ForEachFunc on each item, no successive operations.
@@ -246,11 +262,14 @@ func (s Stream) Head(n int64) Stream {
 			}
 			if n == 0 {
 				// let successive method go ASAP even we have more items to skip
-				// why we don't just break the loop, because if breaks,
-				// this former goroutine will block forever, which will cause goroutine leak.
 				close(source)
+				// why we don't just break the loop, and drain to consume all items.
+				// because if breaks, this former goroutine will block forever,
+				// which will cause goroutine leak.
+				drain(s.source)
 			}
 		}
+		// not enough items in s.source, but we need to let successive method to go ASAP.
 		if n > 0 {
 			close(source)
 		}
@@ -259,6 +278,13 @@ func (s Stream) Head(n int64) Stream {
 	return Range(source)
 }
 
+// Last returns the last item, or nil if no items.
+func (s Stream) Last() (item interface{}) {
+	for item = range s.source {
+	}
+	return
+}
+
 // Map converts each item to another corresponding item, which means it's a 1:1 model.
 func (s Stream) Map(fn MapFunc, opts ...Option) Stream {
 	return s.Walk(func(item interface{}, pipe chan<- interface{}) {
@@ -280,6 +306,21 @@ func (s Stream) Merge() Stream {
 	return Range(source)
 }
 
+// NoneMatch returns whether all elements of this stream don't match the provided predicate.
+// May not evaluate the predicate on all elements if not necessary for determining the result.
+// If the stream is empty then true is returned and the predicate is not evaluated.
+func (s Stream) NoneMatch(predicate func(item interface{}) bool) bool {
+	for item := range s.source {
+		if predicate(item) {
+			// make sure the former goroutine not block, and current func returns fast.
+			go drain(s.source)
+			return false
+		}
+	}
+
+	return true
+}
+
 // Parallel applies the given ParallelFunc to each item concurrently with given number of workers.
 func (s Stream) Parallel(fn ParallelFunc, opts ...Option) {
 	s.Walk(func(item interface{}, pipe chan<- interface{}) {
@@ -411,15 +452,12 @@ func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
 		var wg sync.WaitGroup
 		pool := make(chan lang.PlaceholderType, option.workers)
 
-		for {
+		for item := range s.source {
+			// important, used in another goroutine
+			val := item
 			pool <- lang.Placeholder
-			item, ok := <-s.source
-			if !ok {
-				<-pool
-				break
-			}
-
 			wg.Add(1)
+
 			// better to safely run caller defined method
 			threading.GoSafe(func() {
 				defer func() {
@@ -427,7 +465,7 @@ func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
 					<-pool
 				}()
 
-				fn(item, pipe)
+				fn(val, pipe)
 			})
 		}
 
@@ -439,22 +477,19 @@ func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
 }
 
 func (s Stream) walkUnlimited(fn WalkFunc, option *rxOptions) Stream {
-	pipe := make(chan interface{}, defaultWorkers)
+	pipe := make(chan interface{}, option.workers)
 
 	go func() {
 		var wg sync.WaitGroup
 
-		for {
-			item, ok := <-s.source
-			if !ok {
-				break
-			}
-
+		for item := range s.source {
+			// important, used in another goroutine
+			val := item
 			wg.Add(1)
 			// better to safely run caller defined method
 			threading.GoSafe(func() {
 				defer wg.Done()
-				fn(item, pipe)
+				fn(val, pipe)
 			})
 		}
 
@@ -465,14 +500,14 @@ func (s Stream) walkUnlimited(fn WalkFunc, option *rxOptions) Stream {
 	return Range(pipe)
 }
 
-// UnlimitedWorkers lets the caller to use as many workers as the tasks.
+// UnlimitedWorkers lets the caller use as many workers as the tasks.
 func UnlimitedWorkers() Option {
 	return func(opts *rxOptions) {
 		opts.unlimitedWorkers = true
 	}
 }
 
-// WithWorkers lets the caller to customize the concurrent workers.
+// WithWorkers lets the caller customize the concurrent workers.
 func WithWorkers(workers int) Option {
 	return func(opts *rxOptions) {
 		if workers < minWorkers {
@@ -483,6 +518,7 @@ func WithWorkers(workers int) Option {
 	}
 }
 
+// buildOptions returns a rxOptions with given customizations.
 func buildOptions(opts ...Option) *rxOptions {
 	options := newOptions()
 	for _, opt := range opts {
@@ -492,6 +528,13 @@ func buildOptions(opts ...Option) *rxOptions {
 	return options
 }
 
+// drain drains the given channel.
+func drain(channel <-chan interface{}) {
+	for range channel {
+	}
+}
+
+// newOptions returns a default rxOptions.
 func newOptions() *rxOptions {
 	return &rxOptions{
 		workers: defaultWorkers,

core/fx/stream_test.go

@@ -17,320 +17,489 @@ import (
 )
 
 func TestBuffer(t *testing.T) {
-	const N = 5
-	var count int32
-	var wait sync.WaitGroup
-	wait.Add(1)
-	From(func(source chan<- interface{}) {
-		ticker := time.NewTicker(10 * time.Millisecond)
-		defer ticker.Stop()
+	runCheckedTest(t, func(t *testing.T) {
+		const N = 5
+		var count int32
+		var wait sync.WaitGroup
+		wait.Add(1)
+		From(func(source chan<- interface{}) {
+			ticker := time.NewTicker(10 * time.Millisecond)
+			defer ticker.Stop()
 
-		for i := 0; i < 2*N; i++ {
-			select {
-			case source <- i:
-				atomic.AddInt32(&count, 1)
-			case <-ticker.C:
-				wait.Done()
-				return
+			for i := 0; i < 2*N; i++ {
+				select {
+				case source <- i:
+					atomic.AddInt32(&count, 1)
+				case <-ticker.C:
+					wait.Done()
+					return
+				}
 			}
-		}
-	}).Buffer(N).ForAll(func(pipe <-chan interface{}) {
-		wait.Wait()
-		// why N+1, because take one more to wait for sending into the channel
-		assert.Equal(t, int32(N+1), atomic.LoadInt32(&count))
+		}).Buffer(N).ForAll(func(pipe <-chan interface{}) {
+			wait.Wait()
+			// why N+1, because take one more to wait for sending into the channel
+			assert.Equal(t, int32(N+1), atomic.LoadInt32(&count))
+		})
 	})
 }
 
 func TestBufferNegative(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4).Buffer(-1).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-		for item := range pipe {
-			result += item.(int)
-		}
-		return result, nil
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4).Buffer(-1).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			for item := range pipe {
+				result += item.(int)
+			}
+			return result, nil
+		})
+		assert.Equal(t, 10, result)
 	})
-	assert.Equal(t, 10, result)
 }
 
 func TestCount(t *testing.T) {
-	tests := []struct {
-		name     string
-		elements []interface{}
-	}{
-		{
-			name: "no elements with nil",
-		},
-		{
-			name:     "no elements",
-			elements: []interface{}{},
-		},
-		{
-			name:     "1 element",
-			elements: []interface{}{1},
-		},
-		{
-			name:     "multiple elements",
-			elements: []interface{}{1, 2, 3},
-		},
-	}
+	runCheckedTest(t, func(t *testing.T) {
+		tests := []struct {
+			name     string
+			elements []interface{}
+		}{
+			{
+				name: "no elements with nil",
+			},
+			{
+				name:     "no elements",
+				elements: []interface{}{},
+			},
+			{
+				name:     "1 element",
+				elements: []interface{}{1},
+			},
+			{
+				name:     "multiple elements",
+				elements: []interface{}{1, 2, 3},
+			},
+		}
 
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			val := Just(test.elements...).Count()
-			assert.Equal(t, len(test.elements), val)
-		})
-	}
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				val := Just(test.elements...).Count()
+				assert.Equal(t, len(test.elements), val)
+			})
+		}
+	})
 }
 
 func TestDone(t *testing.T) {
-	var count int32
-	Just(1, 2, 3).Walk(func(item interface{}, pipe chan<- interface{}) {
-		time.Sleep(time.Millisecond * 100)
-		atomic.AddInt32(&count, int32(item.(int)))
-	}).Done()
-	assert.Equal(t, int32(6), count)
+	runCheckedTest(t, func(t *testing.T) {
+		var count int32
+		Just(1, 2, 3).Walk(func(item interface{}, pipe chan<- interface{}) {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, int32(item.(int)))
+		}).Done()
+		assert.Equal(t, int32(6), count)
+	})
 }
 
 func TestJust(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-		for item := range pipe {
-			result += item.(int)
-		}
-		return result, nil
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			for item := range pipe {
+				result += item.(int)
+			}
+			return result, nil
+		})
+		assert.Equal(t, 10, result)
 	})
-	assert.Equal(t, 10, result)
 }
 
 func TestDistinct(t *testing.T) {
-	var result int
-	Just(4, 1, 3, 2, 3, 4).Distinct(func(item interface{}) interface{} {
-		return item
-	}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-		for item := range pipe {
-			result += item.(int)
-		}
-		return result, nil
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(4, 1, 3, 2, 3, 4).Distinct(func(item interface{}) interface{} {
+			return item
+		}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			for item := range pipe {
+				result += item.(int)
+			}
+			return result, nil
+		})
+		assert.Equal(t, 10, result)
 	})
-	assert.Equal(t, 10, result)
 }
 
 func TestFilter(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
-		return item.(int)%2 == 0
-	}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-		for item := range pipe {
-			result += item.(int)
-		}
-		return result, nil
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
+			return item.(int)%2 == 0
+		}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			for item := range pipe {
+				result += item.(int)
+			}
+			return result, nil
+		})
+		assert.Equal(t, 6, result)
+	})
+}
+
+func TestFirst(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		assert.Nil(t, Just().First())
+		assert.Equal(t, "foo", Just("foo").First())
+		assert.Equal(t, "foo", Just("foo", "bar").First())
 	})
-	assert.Equal(t, 6, result)
 }
 
 func TestForAll(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
-		return item.(int)%2 == 0
-	}).ForAll(func(pipe <-chan interface{}) {
-		for item := range pipe {
-			result += item.(int)
-		}
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
+			return item.(int)%2 == 0
+		}).ForAll(func(pipe <-chan interface{}) {
+			for item := range pipe {
+				result += item.(int)
+			}
+		})
+		assert.Equal(t, 6, result)
 	})
-	assert.Equal(t, 6, result)
 }
 
 func TestGroup(t *testing.T) {
-	var groups [][]int
-	Just(10, 11, 20, 21).Group(func(item interface{}) interface{} {
-		v := item.(int)
-		return v / 10
-	}).ForEach(func(item interface{}) {
-		v := item.([]interface{})
-		var group []int
-		for _, each := range v {
-			group = append(group, each.(int))
-		}
-		groups = append(groups, group)
-	})
+	runCheckedTest(t, func(t *testing.T) {
+		var groups [][]int
+		Just(10, 11, 20, 21).Group(func(item interface{}) interface{} {
+			v := item.(int)
+			return v / 10
+		}).ForEach(func(item interface{}) {
+			v := item.([]interface{})
+			var group []int
+			for _, each := range v {
+				group = append(group, each.(int))
+			}
+			groups = append(groups, group)
+		})
 
-	assert.Equal(t, 2, len(groups))
-	for _, group := range groups {
-		assert.Equal(t, 2, len(group))
-		assert.True(t, group[0]/10 == group[1]/10)
-	}
+		assert.Equal(t, 2, len(groups))
+		for _, group := range groups {
+			assert.Equal(t, 2, len(group))
+			assert.True(t, group[0]/10 == group[1]/10)
+		}
+	})
 }
 
 func TestHead(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4).Head(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-		for item := range pipe {
-			result += item.(int)
-		}
-		return result, nil
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4).Head(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			for item := range pipe {
+				result += item.(int)
+			}
+			return result, nil
+		})
+		assert.Equal(t, 3, result)
 	})
-	assert.Equal(t, 3, result)
 }
 
 func TestHeadZero(t *testing.T) {
-	assert.Panics(t, func() {
-		Just(1, 2, 3, 4).Head(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-			return nil, nil
+	runCheckedTest(t, func(t *testing.T) {
+		assert.Panics(t, func() {
+			Just(1, 2, 3, 4).Head(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+				return nil, nil
+			})
 		})
 	})
 }
 
 func TestHeadMore(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4).Head(6).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-		for item := range pipe {
-			result += item.(int)
-		}
-		return result, nil
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4).Head(6).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			for item := range pipe {
+				result += item.(int)
+			}
+			return result, nil
+		})
+		assert.Equal(t, 10, result)
+	})
+}
+
+func TestLast(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		goroutines := runtime.NumGoroutine()
+		assert.Nil(t, Just().Last())
+		assert.Equal(t, "foo", Just("foo").Last())
+		assert.Equal(t, "bar", Just("foo", "bar").Last())
+		// let scheduler schedule first
+		runtime.Gosched()
+		assert.Equal(t, goroutines, runtime.NumGoroutine())
 	})
-	assert.Equal(t, 10, result)
 }
 
 func TestMap(t *testing.T) {
-	log.SetOutput(ioutil.Discard)
+	runCheckedTest(t, func(t *testing.T) {
+		log.SetOutput(ioutil.Discard)
 
-	tests := []struct {
-		mapper MapFunc
-		expect int
-	}{
-		{
-			mapper: func(item interface{}) interface{} {
-				v := item.(int)
-				return v * v
+		tests := []struct {
+			mapper MapFunc
+			expect int
+		}{
+			{
+				mapper: func(item interface{}) interface{} {
+					v := item.(int)
+					return v * v
+				},
+				expect: 30,
 			},
-			expect: 30,
-		},
-		{
-			mapper: func(item interface{}) interface{} {
-				v := item.(int)
-				if v%2 == 0 {
-					return 0
-				}
-				return v * v
-			},
-			expect: 10,
-		},
-		{
-			mapper: func(item interface{}) interface{} {
-				v := item.(int)
-				if v%2 == 0 {
-					panic(v)
-				}
-				return v * v
-			},
-			expect: 10,
-		},
-	}
-
-	// Map(...) works even WithWorkers(0)
-	for i, test := range tests {
-		t.Run(stringx.Rand(), func(t *testing.T) {
-			var result int
-			var workers int
-			if i%2 == 0 {
-				workers = 0
-			} else {
-				workers = runtime.NumCPU()
-			}
-			From(func(source chan<- interface{}) {
-				for i := 1; i < 5; i++ {
-					source <- i
-				}
-			}).Map(test.mapper, WithWorkers(workers)).Reduce(
-				func(pipe <-chan interface{}) (interface{}, error) {
-					for item := range pipe {
-						result += item.(int)
+			{
+				mapper: func(item interface{}) interface{} {
+					v := item.(int)
+					if v%2 == 0 {
+						return 0
 					}
-					return result, nil
-				})
+					return v * v
+				},
+				expect: 10,
+			},
+			{
+				mapper: func(item interface{}) interface{} {
+					v := item.(int)
+					if v%2 == 0 {
+						panic(v)
+					}
+					return v * v
+				},
+				expect: 10,
+			},
+		}
 
-			assert.Equal(t, test.expect, result)
-		})
-	}
+		// Map(...) works even WithWorkers(0)
+		for i, test := range tests {
+			t.Run(stringx.Rand(), func(t *testing.T) {
+				var result int
+				var workers int
+				if i%2 == 0 {
+					workers = 0
+				} else {
+					workers = runtime.NumCPU()
+				}
+				From(func(source chan<- interface{}) {
+					for i := 1; i < 5; i++ {
+						source <- i
+					}
+				}).Map(test.mapper, WithWorkers(workers)).Reduce(
+					func(pipe <-chan interface{}) (interface{}, error) {
+						for item := range pipe {
+							result += item.(int)
+						}
+						return result, nil
+					})
+
+				assert.Equal(t, test.expect, result)
+			})
+		}
+	})
 }
 
 func TestMerge(t *testing.T) {
-	Just(1, 2, 3, 4).Merge().ForEach(func(item interface{}) {
-		assert.ElementsMatch(t, []interface{}{1, 2, 3, 4}, item.([]interface{}))
+	runCheckedTest(t, func(t *testing.T) {
+		Just(1, 2, 3, 4).Merge().ForEach(func(item interface{}) {
+			assert.ElementsMatch(t, []interface{}{1, 2, 3, 4}, item.([]interface{}))
+		})
 	})
 }
 
 func TestParallelJust(t *testing.T) {
-	var count int32
-	Just(1, 2, 3).Parallel(func(item interface{}) {
-		time.Sleep(time.Millisecond * 100)
-		atomic.AddInt32(&count, int32(item.(int)))
-	}, UnlimitedWorkers())
-	assert.Equal(t, int32(6), count)
+	runCheckedTest(t, func(t *testing.T) {
+		var count int32
+		Just(1, 2, 3).Parallel(func(item interface{}) {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, int32(item.(int)))
+		}, UnlimitedWorkers())
+		assert.Equal(t, int32(6), count)
+	})
 }
 
 func TestReverse(t *testing.T) {
-	Just(1, 2, 3, 4).Reverse().Merge().ForEach(func(item interface{}) {
-		assert.ElementsMatch(t, []interface{}{4, 3, 2, 1}, item.([]interface{}))
+	runCheckedTest(t, func(t *testing.T) {
+		Just(1, 2, 3, 4).Reverse().Merge().ForEach(func(item interface{}) {
+			assert.ElementsMatch(t, []interface{}{4, 3, 2, 1}, item.([]interface{}))
+		})
 	})
 }
 
 func TestSort(t *testing.T) {
-	var prev int
-	Just(5, 3, 7, 1, 9, 6, 4, 8, 2).Sort(func(a, b interface{}) bool {
-		return a.(int) < b.(int)
-	}).ForEach(func(item interface{}) {
-		next := item.(int)
-		assert.True(t, prev < next)
-		prev = next
+	runCheckedTest(t, func(t *testing.T) {
+		var prev int
+		Just(5, 3, 7, 1, 9, 6, 4, 8, 2).Sort(func(a, b interface{}) bool {
+			return a.(int) < b.(int)
+		}).ForEach(func(item interface{}) {
+			next := item.(int)
+			assert.True(t, prev < next)
+			prev = next
+		})
 	})
 }
 
 func TestSplit(t *testing.T) {
-	assert.Panics(t, func() {
-		Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(0).Done()
+	runCheckedTest(t, func(t *testing.T) {
+		assert.Panics(t, func() {
+			Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(0).Done()
+		})
+		var chunks [][]interface{}
+		Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(4).ForEach(func(item interface{}) {
+			chunk := item.([]interface{})
+			chunks = append(chunks, chunk)
+		})
+		assert.EqualValues(t, [][]interface{}{
+			{1, 2, 3, 4},
+			{5, 6, 7, 8},
+			{9, 10},
+		}, chunks)
 	})
-	var chunks [][]interface{}
-	Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(4).ForEach(func(item interface{}) {
-		chunk := item.([]interface{})
-		chunks = append(chunks, chunk)
-	})
-	assert.EqualValues(t, [][]interface{}{
-		{1, 2, 3, 4},
-		{5, 6, 7, 8},
-		{9, 10},
-	}, chunks)
 }
 
 func TestTail(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4).Tail(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-		for item := range pipe {
-			result += item.(int)
-		}
-		return result, nil
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4).Tail(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			for item := range pipe {
+				result += item.(int)
+			}
+			return result, nil
+		})
+		assert.Equal(t, 7, result)
 	})
-	assert.Equal(t, 7, result)
 }
 
 func TestTailZero(t *testing.T) {
-	assert.Panics(t, func() {
-		Just(1, 2, 3, 4).Tail(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
-			return nil, nil
+	runCheckedTest(t, func(t *testing.T) {
+		assert.Panics(t, func() {
+			Just(1, 2, 3, 4).Tail(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+				return nil, nil
+			})
 		})
 	})
 }
 
 func TestWalk(t *testing.T) {
-	var result int
-	Just(1, 2, 3, 4, 5).Walk(func(item interface{}, pipe chan<- interface{}) {
-		if item.(int)%2 != 0 {
-			pipe <- item
-		}
-	}, UnlimitedWorkers()).ForEach(func(item interface{}) {
-		result += item.(int)
+	runCheckedTest(t, func(t *testing.T) {
+		var result int
+		Just(1, 2, 3, 4, 5).Walk(func(item interface{}, pipe chan<- interface{}) {
+			if item.(int)%2 != 0 {
+				pipe <- item
+			}
+		}, UnlimitedWorkers()).ForEach(func(item interface{}) {
+			result += item.(int)
+		})
+		assert.Equal(t, 9, result)
+	})
+}
+
+func TestStream_AnyMach(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
+			return item.(int) == 4
+		}))
+		assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
+			return item.(int) == 0
+		}))
+		assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
+			return item.(int) == 2
+		}))
+		assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
+			return item.(int) == 2
+		}))
+	})
+}
+
+func TestStream_AllMach(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		assetEqual(
+			t, true, Just(1, 2, 3).AllMach(func(item interface{}) bool {
+				return true
+			}),
+		)
+		assetEqual(
+			t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
+				return false
+			}),
+		)
+		assetEqual(
+			t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
+				return item.(int) == 1
+			}),
+		)
+	})
+}
+
+func TestStream_NoneMatch(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		assetEqual(
+			t, true, Just(1, 2, 3).NoneMatch(func(item interface{}) bool {
+				return false
+			}),
+		)
+		assetEqual(
+			t, false, Just(1, 2, 3).NoneMatch(func(item interface{}) bool {
+				return true
+			}),
+		)
+		assetEqual(
+			t, true, Just(1, 2, 3).NoneMatch(func(item interface{}) bool {
+				return item.(int) == 4
+			}),
+		)
+	})
+}
+
+func TestConcat(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		a1 := []interface{}{1, 2, 3}
+		a2 := []interface{}{4, 5, 6}
+		s1 := Just(a1...)
+		s2 := Just(a2...)
+		stream := Concat(s1, s2)
+		var items []interface{}
+		for item := range stream.source {
+			items = append(items, item)
+		}
+		sort.Slice(items, func(i, j int) bool {
+			return items[i].(int) < items[j].(int)
+		})
+		ints := make([]interface{}, 0)
+		ints = append(ints, a1...)
+		ints = append(ints, a2...)
+		assetEqual(t, ints, items)
+	})
+}
+
+func TestStream_Skip(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		assetEqual(t, 3, Just(1, 2, 3, 4).Skip(1).Count())
+		assetEqual(t, 1, Just(1, 2, 3, 4).Skip(3).Count())
+		assetEqual(t, 4, Just(1, 2, 3, 4).Skip(0).Count())
+		equal(t, Just(1, 2, 3, 4).Skip(3), []interface{}{4})
+		assert.Panics(t, func() {
+			Just(1, 2, 3, 4).Skip(-1)
+		})
+	})
+}
+
+func TestStream_Concat(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		stream := Just(1).Concat(Just(2), Just(3))
+		var items []interface{}
+		for item := range stream.source {
+			items = append(items, item)
+		}
+		sort.Slice(items, func(i, j int) bool {
+			return items[i].(int) < items[j].(int)
+		})
+		assetEqual(t, []interface{}{1, 2, 3}, items)
+
+		just := Just(1)
+		equal(t, just.Concat(just), []interface{}{1})
 	})
-	assert.Equal(t, 9, result)
 }
 
 func BenchmarkParallelMapReduce(b *testing.B) {
@@ -377,6 +546,12 @@ func BenchmarkMapReduce(b *testing.B) {
 	}).Map(mapper).Reduce(reducer)
 }
 
+func assetEqual(t *testing.T, except, data interface{}) {
+	if !reflect.DeepEqual(except, data) {
+		t.Errorf(" %v, want %v", data, except)
+	}
+}
+
 func equal(t *testing.T, stream Stream, data []interface{}) {
 	items := make([]interface{}, 0)
 	for item := range stream.source {
@@ -387,85 +562,10 @@ func equal(t *testing.T, stream Stream, data []interface{}) {
 	}
 }
 
-func assetEqual(t *testing.T, except, data interface{}) {
-	if !reflect.DeepEqual(except, data) {
-		t.Errorf(" %v, want %v", data, except)
-	}
-}
-
-func TestStream_AnyMach(t *testing.T) {
-	assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
-		return item.(int) == 4
-	}))
-	assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
-		return item.(int) == 0
-	}))
-	assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
-		return item.(int) == 2
-	}))
-	assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
-		return item.(int) == 2
-	}))
-}
-
-func TestStream_AllMach(t *testing.T) {
-	assetEqual(
-		t, true, Just(1, 2, 3).AllMach(func(item interface{}) bool {
-			return true
-		}),
-	)
-	assetEqual(
-		t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
-			return false
-		}),
-	)
-	assetEqual(
-		t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
-			return item.(int) == 1
-		}),
-	)
-}
-
-func TestConcat(t *testing.T) {
-	a1 := []interface{}{1, 2, 3}
-	a2 := []interface{}{4, 5, 6}
-	s1 := Just(a1...)
-	s2 := Just(a2...)
-	stream := Concat(s1, s2)
-	var items []interface{}
-	for item := range stream.source {
-		items = append(items, item)
-	}
-	sort.Slice(items, func(i, j int) bool {
-		return items[i].(int) < items[j].(int)
-	})
-	ints := make([]interface{}, 0)
-	ints = append(ints, a1...)
-	ints = append(ints, a2...)
-	assetEqual(t, ints, items)
-}
-
-func TestStream_Skip(t *testing.T) {
-	assetEqual(t, 3, Just(1, 2, 3, 4).Skip(1).Count())
-	assetEqual(t, 1, Just(1, 2, 3, 4).Skip(3).Count())
-	assetEqual(t, 4, Just(1, 2, 3, 4).Skip(0).Count())
-	equal(t, Just(1, 2, 3, 4).Skip(3), []interface{}{4})
-	assert.Panics(t, func() {
-		Just(1, 2, 3, 4).Skip(-1)
-	})
-}
-
-func TestStream_Concat(t *testing.T) {
-	stream := Just(1).Concat(Just(2), Just(3))
-	var items []interface{}
-	for item := range stream.source {
-		items = append(items, item)
-	}
-	sort.Slice(items, func(i, j int) bool {
-		return items[i].(int) < items[j].(int)
-	})
-	assetEqual(t, []interface{}{1, 2, 3}, items)
-
-	just := Just(1)
-	equal(t, just.Concat(just), []interface{}{1})
+func runCheckedTest(t *testing.T, fn func(t *testing.T)) {
+	goroutines := runtime.NumGoroutine()
+	fn(t)
+	// let scheduler schedule first
+	time.Sleep(time.Millisecond)
+	assert.True(t, runtime.NumGoroutine() <= goroutines)
 }

core/lang/lang.go

@@ -4,8 +4,8 @@ package lang
 var Placeholder PlaceholderType
 
 type (
-	// GenericType can be used to hold any type.
-	GenericType = interface{}
+	// AnyType can be used to hold any type.
+	AnyType = interface{}
 	// PlaceholderType represents a placeholder type.
 	PlaceholderType = struct{}
 )

core/mr/mapreduce.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/tal-tech/go-zero/core/errorx"
 	"github.com/tal-tech/go-zero/core/lang"
-	"github.com/tal-tech/go-zero/core/syncx"
 	"github.com/tal-tech/go-zero/core/threading"
 )
 
@@ -95,9 +94,9 @@ func Map(generate GenerateFunc, mapper MapFunc, opts ...Option) chan interface{}
 	options := buildOptions(opts...)
 	source := buildSource(generate)
 	collector := make(chan interface{}, options.workers)
-	done := syncx.NewDoneChan()
+	done := make(chan lang.PlaceholderType)
 
-	go executeMappers(options.ctx, mapper, source, collector, done.Done(), options.workers)
+	go executeMappers(options.ctx, mapper, source, collector, done, options.workers)
 
 	return collector
 }
@@ -122,13 +121,13 @@ func MapReduceWithSource(source <-chan interface{}, mapper MapperFunc, reducer R
 	}()
 
 	collector := make(chan interface{}, options.workers)
-	done := syncx.NewDoneChan()
-	writer := newGuardedWriter(options.ctx, output, done.Done())
+	done := make(chan lang.PlaceholderType)
+	writer := newGuardedWriter(options.ctx, output, done)
 	var closeOnce sync.Once
 	var retErr errorx.AtomicError
 	finish := func() {
 		closeOnce.Do(func() {
-			done.Close()
+			close(done)
 			close(output)
 		})
 	}
@@ -159,7 +158,7 @@ func MapReduceWithSource(source <-chan interface{}, mapper MapperFunc, reducer R
 
 	go executeMappers(options.ctx, func(item interface{}, w Writer) {
 		mapper(item, w, cancel)
-	}, source, collector, done.Done(), options.workers)
+	}, source, collector, done, options.workers)
 
 	value, ok := <-output
 	if err := retErr.Load(); err != nil {

core/mr/readme-cn.md

@@ -0,0 +1,89 @@
+# mapreduce
+
+[English](readme.md) | 简体中文
+
+## 为什么需要 MapReduce
+
+在实际的业务场景中我们常常需要从不同的 rpc 服务中获取相应属性来组装成复杂对象。
+
+比如要查询商品详情：
+
+1. 商品服务-查询商品属性
+2. 库存服务-查询库存属性
+3. 价格服务-查询价格属性
+4. 营销服务-查询营销属性
+
+如果是串行调用的话响应时间会随着 rpc 调用次数呈线性增长，所以我们要优化性能一般会将串行改并行。
+
+简单的场景下使用 `WaitGroup` 也能够满足需求，但是如果我们需要对 rpc 调用返回的数据进行校验、数据加工转换、数据汇总呢？继续使用 `WaitGroup` 就有点力不从心了，go 的官方库中并没有这种工具（java 中提供了 CompleteFuture），我们依据 MapReduce 架构思想实现了进程内的数据批处理 MapReduce 并发工具类。
+
+## 设计思路
+
+我们尝试把自己代入到作者的角色梳理一下并发工具可能的业务场景：
+
+1. 查询商品详情：支持并发调用多个服务来组合产品属性，支持调用错误可以立即结束。
+2. 商品详情页自动推荐用户卡券：支持并发校验卡券，校验失败自动剔除，返回全部卡券。
+
+以上实际都是在进行对输入数据进行处理最后输出清洗后的数据，针对数据处理有个非常经典的异步模式：生产者消费者模式。于是我们可以抽象一下数据批处理的生命周期，大致可以分为三个阶段：
+
+<img src="https://raw.githubusercontent.com/zeromicro/zero-doc/main/doc/images/mapreduce-serial-cn.png" width="500">
+
+1. 数据生产 generate
+2. 数据加工 mapper
+3. 数据聚合 reducer
+
+其中数据生产是不可或缺的阶段，数据加工、数据聚合是可选阶段，数据生产与加工支持并发调用，数据聚合基本属于纯内存操作单协程即可。
+
+再来思考一下不同阶段之间数据应该如何流转，既然不同阶段的数据处理都是由不同 goroutine 执行的，那么很自然的可以考虑采用 channel 来实现 goroutine 之间的通信。
+
+<img src="https://raw.githubusercontent.com/zeromicro/zero-doc/main/doc/images/mapreduce-cn.png" width="500">
+
+
+如何实现随时终止流程呢？
+
+`goroutine` 中监听一个全局的结束 `channel` 和调用方提供的 `ctx` 就行。
+
+## 简单示例
+
+并行求平方和（不要嫌弃示例简单，只是模拟并发）
+
+```go
+package main
+
+import (
+    "fmt"
+    "log"
+
+    "github.com/tal-tech/go-zero/core/mr"
+)
+
+func main() {
+    val, err := mr.MapReduce(func(source chan<- interface{}) {
+        // generator
+        for i := 0; i < 10; i++ {
+            source <- i
+        }
+    }, func(item interface{}, writer mr.Writer, cancel func(error)) {
+        // mapper
+        i := item.(int)
+        writer.Write(i * i)
+    }, func(pipe <-chan interface{}, writer mr.Writer, cancel func(error)) {
+        // reducer
+        var sum int
+        for i := range pipe {
+            sum += i.(int)
+        }
+        writer.Write(sum)
+    })
+    if err != nil {
+        log.Fatal(err)
+    }
+    fmt.Println("result:", val)
+}
+```
+
+更多示例：[https://github.com/zeromicro/zero-examples/tree/main/mapreduce](https://github.com/zeromicro/zero-examples/tree/main/mapreduce)
+
+## 欢迎 star！⭐
+
+如果你正在使用或者觉得这个项目对你有帮助，请 **star** 支持，感谢！

core/mr/readme.md

@@ -0,0 +1,90 @@
+<img align="right" width="150px" src="https://raw.githubusercontent.com/zeromicro/zero-doc/main/doc/images/go-zero.png">
+
+# mapreduce
+
+English | [简体中文](readme-cn.md)
+
+## Why MapReduce is needed
+
+In practical business scenarios we often need to get the corresponding properties from different rpc services to assemble complex objects.
+
+For example, to query product details.
+
+1. product service - query product attributes
+2. inventory service - query inventory properties
+3. price service - query price attributes
+4. marketing service - query marketing properties
+
+If it is a serial call, the response time will increase linearly with the number of rpc calls, so we will generally change serial to parallel to optimize response time.
+
+Simple scenarios using `WaitGroup` can also meet the needs, but what if we need to check the data returned by the rpc call, data processing, data aggregation? The official go library does not have such a tool (CompleteFuture is provided in java), so we implemented an in-process data batching MapReduce concurrent tool based on the MapReduce architecture.
+
+## Design ideas
+
+Let's try to put ourselves in the author's shoes and sort out the possible business scenarios for the concurrency tool:
+
+1. querying product details: supporting concurrent calls to multiple services to combine product attributes, and supporting call errors that can be ended immediately.
+2. automatic recommendation of user card coupons on product details page: support concurrently verifying card coupons, automatically rejecting them if they fail, and returning all of them.
+
+The above is actually processing the input data and finally outputting the cleaned data. There is a very classic asynchronous pattern for data processing: the producer-consumer pattern. So we can abstract the life cycle of data batch processing, which can be roughly divided into three phases.
+
+<img src="https://raw.githubusercontent.com/zeromicro/zero-doc/main/doc/images/mapreduce-serial-en.png" width="500">
+
+1. data production generate
+2. data processing mapper
+3. data aggregation reducer
+
+Data producing is an indispensable stage, data processing and data aggregation are optional stages, data producing and processing support concurrent calls, data aggregation is basically a pure memory operation, so a single concurrent process can do it.
+
+Since different stages of data processing are performed by different goroutines, it is natural to consider the use of channel to achieve communication between goroutines.
+
+<img src="https://raw.githubusercontent.com/zeromicro/zero-doc/main/doc/images/mapreduce-en.png" width="500">
+
+How can I terminate the process at any time?
+
+It's simple, just receive from a  channel or the given context in the goroutine.
+
+## A simple example
+
+Calculate the sum of squares, simulating the concurrency.
+
+```go
+package main
+
+import (
+    "fmt"
+    "log"
+
+    "github.com/tal-tech/go-zero/core/mr"
+)
+
+func main() {
+    val, err := mr.MapReduce(func(source chan<- interface{}) {
+        // generator
+        for i := 0; i < 10; i++ {
+            source <- i
+        }
+    }, func(item interface{}, writer mr.Writer, cancel func(error)) {
+        // mapper
+        i := item.(int)
+        writer.Write(i * i)
+    }, func(pipe <-chan interface{}, writer mr.Writer, cancel func(error)) {
+        // reducer
+        var sum int
+        for i := range pipe {
+            sum += i.(int)
+        }
+        writer.Write(sum)
+    })
+    if err != nil {
+        log.Fatal(err)
+    }
+    fmt.Println("result:", val)
+}
+```
+
+More examples: [https://github.com/zeromicro/zero-examples/tree/main/mapreduce](https://github.com/zeromicro/zero-examples/tree/main/mapreduce)
+
+## Give a Star! ⭐
+
+If you like or are using this project to learn or start your solution, please give it a star. Thanks!

readme-cn.md

@@ -104,10 +104,10 @@ GO111MODULE=on GOPROXY=https://goproxy.cn/,direct go get -u github.com/tal-tech/
 
     ```shell
     # Go 1.15 及之前版本
-    GO111MODULE=on GOPROXY=https://goproxy.cn/,direct go get -u github.com/tal-tech/go-zero/tools/goctl@cli
+    GO111MODULE=on GOPROXY=https://goproxy.cn/,direct go get -u github.com/tal-tech/go-zero/tools/goctl@latest
     
     # Go 1.16 及以后版本
-    GOPROXY=https://goproxy.cn/,direct go install github.com/tal-tech/go-zero/tools/goctl@cli
+    GOPROXY=https://goproxy.cn/,direct go install github.com/tal-tech/go-zero/tools/goctl@latest
     ```
 
     确保 goctl 可执行

readme.md

@@ -107,10 +107,10 @@ go get -u github.com/tal-tech/go-zero
 
    ```shell
    # for Go 1.15 and earlier
-   GO111MODULE=on go get -u github.com/tal-tech/go-zero/tools/goctl@cli
+   GO111MODULE=on go get -u github.com/tal-tech/go-zero/tools/goctl@latest
    
    # for Go 1.16 and later
-   go install github.com/tal-tech/go-zero/tools/goctl@cli
+   go install github.com/tal-tech/go-zero/tools/goctl@latest
    ```
 
    make sure goctl is executable.

tools/goctl/api/parser/g4/gen/api/apiparser_parser.go

@@ -1,4 +1,5 @@
-package api // ApiParser
+package api
+
 import (
 	"fmt"
 	"reflect"

tools/goctl/api/spec/spec.go

@@ -1,5 +1,6 @@
 package spec
 
+// RoutePrefixKey is the prefix keyword for the routes.
 const RoutePrefixKey = "prefix"
 
 type (

tools/goctl/internal/version/version.go

@@ -6,7 +6,7 @@ import (
 )
 
 // BuildVersion is the version of goctl.
-const BuildVersion = "1.2.4"
+const BuildVersion = "1.2.5"
 
 var tag = map[string]int{"pre-alpha": 0, "alpha": 1, "pre-bata": 2, "beta": 3, "released": 4, "": 5}
 

tools/goctl/model/sql/parser/parser.go

@@ -371,6 +371,7 @@ func getTableFields(table *model.Table) (map[string]*Field, error) {
 	return fieldM, nil
 }
 
+// GetSafeTables escapes the golang keywords from sql tables.
 func GetSafeTables(tables []*parser.Table) []*parser.Table {
 	var list []*parser.Table
 	for _, t := range tables {
@@ -381,6 +382,7 @@ func GetSafeTables(tables []*parser.Table) []*parser.Table {
 	return list
 }
 
+// GetSafeTable escapes the golang keywords from sql table.
 func GetSafeTable(table *parser.Table) *parser.Table {
 	table.Name = su.EscapeGolangKeyword(table.Name)
 	for _, c := range table.Columns {

tools/goctl/upgrade/upgrade.go

@@ -9,11 +9,11 @@ import (
 )
 
 // Upgrade gets the latest goctl by
-// go get -u github.com/tal-tech/go-zero/tools/goctl
+// go install github.com/tal-tech/go-zero/tools/goctl@latest
 func Upgrade(_ *cli.Context) error {
-	cmd := `GO111MODULE=on GOPROXY=https://goproxy.cn/,direct go install github.com/tal-tech/go-zero/tools/goctl@cli`
+	cmd := `GO111MODULE=on GOPROXY=https://goproxy.cn/,direct go install github.com/tal-tech/go-zero/tools/goctl@latest`
 	if runtime.GOOS == "windows" {
-		cmd = `set GOPROXY=https://goproxy.cn,direct && go install github.com/tal-tech/go-zero/tools/goctl@cli`
+		cmd = `set GOPROXY=https://goproxy.cn,direct && go install github.com/tal-tech/go-zero/tools/goctl@latest`
 	}
 	info, err := execx.Run(cmd, "")
 	if err != nil {

tools/goctl/util/file.go

@@ -95,12 +95,30 @@ func GetGitHome() (string, error) {
 
 // GetTemplateDir returns the category path value in GoctlHome where could get it by GetGoctlHome
 func GetTemplateDir(category string) (string, error) {
-	goctlHome, err := GetGoctlHome()
+	home, err := GetGoctlHome()
 	if err != nil {
 		return "", err
 	}
+	if home == goctlHome {
+		// backward compatible, it will be removed in the feature
+		// backward compatible start
+		beforeTemplateDir := filepath.Join(home, version.GetGoctlVersion(), category)
+		fs, _ := ioutil.ReadDir(beforeTemplateDir)
+		var hasContent bool
+		for _, e := range fs {
+			if e.Size() > 0 {
+				hasContent = true
+			}
+		}
+		if hasContent {
+			return beforeTemplateDir, nil
+		}
+		// backward compatible end
 
-	return filepath.Join(goctlHome, version.GetGoctlVersion(), category), nil
+		return filepath.Join(home, category), nil
+	}
+
+	return filepath.Join(home, version.GetGoctlVersion(), category), nil
 }
 
 // InitTemplates creates template files GoctlHome where could get it by GetGoctlHome

tools/goctl/util/file_test.go

@@ -1,13 +1,66 @@
 package util
 
 import (
+	"io/ioutil"
 	"os"
 	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/tal-tech/go-zero/tools/goctl/internal/version"
 )
 
+func TestGetTemplateDir(t *testing.T) {
+	category := "foo"
+	t.Run("before_have_templates", func(t *testing.T) {
+		home := t.TempDir()
+		RegisterGoctlHome("")
+		RegisterGoctlHome(home)
+		v := version.GetGoctlVersion()
+		dir := filepath.Join(home, v, category)
+		err := MkdirIfNotExist(dir)
+		if err != nil {
+			return
+		}
+		tempFile := filepath.Join(dir, "bar.txt")
+		err = ioutil.WriteFile(tempFile, []byte("foo"), os.ModePerm)
+		if err != nil {
+			return
+		}
+		templateDir, err := GetTemplateDir(category)
+		if err != nil {
+			return
+		}
+		assert.Equal(t, dir, templateDir)
+		RegisterGoctlHome("")
+	})
+
+	t.Run("before_has_no_template", func(t *testing.T) {
+		home := t.TempDir()
+		RegisterGoctlHome("")
+		RegisterGoctlHome(home)
+		dir := filepath.Join(home, category)
+		err := MkdirIfNotExist(dir)
+		if err != nil {
+			return
+		}
+		templateDir, err := GetTemplateDir(category)
+		if err != nil {
+			return
+		}
+		assert.Equal(t, dir, templateDir)
+	})
+
+	t.Run("default", func(t *testing.T) {
+		RegisterGoctlHome("")
+		dir, err := GetTemplateDir(category)
+		if err != nil {
+			return
+		}
+		assert.Contains(t, dir, version.BuildVersion)
+	})
+}
+
 func TestGetGitHome(t *testing.T) {
 	homeDir, err := os.UserHomeDir()
 	if err != nil {

tools/goctl/util/string.go

@@ -96,6 +96,7 @@ func isNumber(r rune) bool {
 	return '0' <= r && r <= '9'
 }
 
+// EscapeGolangKeyword escapes the golang keywords.
 func EscapeGolangKeyword(s string) string {
 	if !isGolangKeyword(s) {
 		return s

zrpc/config.go

@@ -83,6 +83,12 @@ func (cc RpcClientConf) BuildTarget() (string, error) {
 	if cc.Etcd.HasAccount() {
 		discov.RegisterAccount(cc.Etcd.Hosts, cc.Etcd.User, cc.Etcd.Pass)
 	}
+	if cc.Etcd.HasTLS() {
+		if err := discov.RegisterTLS(cc.Etcd.Hosts, cc.Etcd.CertFile, cc.Etcd.CertKeyFile,
+			cc.Etcd.CACertFile, cc.Etcd.InsecureSkipVerify); err != nil {
+			return "", err
+		}
+	}
 
 	return resolver.BuildDiscovTarget(cc.Etcd.Hosts, cc.Etcd.Key), nil
 }

zrpc/internal/rpcpubserver.go

@@ -21,6 +21,10 @@ func NewRpcPubServer(etcd discov.EtcdConf, listenOn string, opts ...ServerOption
 		if etcd.HasAccount() {
 			pubOpts = append(pubOpts, discov.WithPubEtcdAccount(etcd.User, etcd.Pass))
 		}
+		if etcd.HasTLS() {
+			pubOpts = append(pubOpts, discov.WithPubEtcdTLS(etcd.CertFile, etcd.CertKeyFile,
+				etcd.CACertFile, etcd.InsecureSkipVerify))
+		}
 		pubClient := discov.NewPublisher(etcd.Hosts, etcd.Key, pubListenOn, pubOpts...)
 		return pubClient.KeepAlive()
 	}

zrpc/server_test.go

@@ -43,7 +43,7 @@ func TestServer(t *testing.T) {
 				Mode:        "console",
 			},
 		},
-		ListenOn:      ":8080",
+		ListenOn:      "localhost:8080",
 		Etcd:          discov.EtcdConf{},
 		Auth:          false,
 		Redis:         redis.RedisKeyConf{},
@@ -67,7 +67,7 @@ func TestServerError(t *testing.T) {
 				Mode:        "console",
 			},
 		},
-		ListenOn: ":8080",
+		ListenOn: "localhost:8080",
 		Etcd: discov.EtcdConf{
 			Hosts: []string{"localhost"},
 		},
@@ -86,7 +86,7 @@ func TestServer_HasEtcd(t *testing.T) {
 				Mode:        "console",
 			},
 		},
-		ListenOn: ":8080",
+		ListenOn: "localhost:8080",
 		Etcd: discov.EtcdConf{
 			Hosts: []string{"notexist"},
 			Key:   "any",