update stringx doc

* reactor sql generation

* reactor sql generation

* add console & example

* optimize unit test & add document

* modify default config

* remove test file

* Revert "remove test file"

This reverts commit 81041f9e

* fix stringx.go & optimize example

* remove unused code

.codecov.yml

@@ -0,0 +1,3 @@
+ignore:
+  - "example/*"
+  - "tools/*"

.github/workflows/go.yml

@@ -25,10 +25,11 @@ jobs:
     - name: Get dependencies
       run: |
         go get -v -t -d ./...
-        if [ -f Gopkg.toml ]; then
-            curl https://raw.githubusercontent.com/golang/dep/master/install.sh | sh
-            dep ensure
-        fi
 
     - name: Test
-      run: go test -v -race ./...
+      run: go test -race -coverprofile=coverage.txt -covermode=atomic ./...
+
+    - name: Codecov
+      uses: codecov/codecov-action@v1.0.6
+      with:
+        token: ${{secrets.CODECOV_TOKEN}}

.gitlab-ci.yml

@@ -1,18 +0,0 @@
-stages:
-- analysis
-
-variables:
-  GOPATH: '/runner-cache/zero'
-  GOCACHE: '/runner-cache/zero'
-  GOPROXY: 'https://goproxy.cn,direct'
-
-analysis:
-  stage: analysis
-  image: golang
-  script:
-    - go version && go env
-    - go test -short $(go list ./...) | grep -v "no test"
-  only:
-   - merge_requests
-  tags:
-    - common

core/breaker/breakers.go

@@ -60,17 +60,15 @@ func do(name string, execute func(b Breaker) error) error {
 	lock.RUnlock()
 	if ok {
 		return execute(b)
-	} else {
-		lock.Lock()
-		b, ok = breakers[name]
-		if ok {
-			lock.Unlock()
-			return execute(b)
-		} else {
-			b = NewBreaker(WithName(name))
-			breakers[name] = b
-			lock.Unlock()
-			return execute(b)
-		}
 	}
+
+	lock.Lock()
+	b, ok = breakers[name]
+	if !ok {
+		b = NewBreaker(WithName(name))
+		breakers[name] = b
+	}
+	lock.Unlock()
+
+	return execute(b)
 }

core/collection/cache_test.go

@@ -72,9 +72,9 @@ func TestCacheWithLruEvicts(t *testing.T) {
 	cache.Set("third", "third element")
 	cache.Set("fourth", "fourth element")
 
-	value, ok := cache.Get("first")
+	_, ok := cache.Get("first")
 	assert.False(t, ok)
-	value, ok = cache.Get("second")
+	value, ok := cache.Get("second")
 	assert.True(t, ok)
 	assert.Equal(t, "second element", value)
 	value, ok = cache.Get("third")
@@ -94,9 +94,9 @@ func TestCacheWithLruEvicted(t *testing.T) {
 	cache.Set("third", "third element")
 	cache.Set("fourth", "fourth element")
 
-	value, ok := cache.Get("first")
+	_, ok := cache.Get("first")
 	assert.False(t, ok)
-	value, ok = cache.Get("second")
+	value, ok := cache.Get("second")
 	assert.True(t, ok)
 	assert.Equal(t, "second element", value)
 	cache.Set("fifth", "fifth element")

core/contextx/valueonlycontext_test.go

@@ -34,7 +34,7 @@ func TestContextCancel(t *testing.T) {
 	assert.NotEqual(t, context.Canceled, c2.Err())
 }
 
-func TestConextDeadline(t *testing.T) {
+func TestContextDeadline(t *testing.T) {
 	c, _ := context.WithDeadline(context.Background(), time.Now().Add(10*time.Millisecond))
 	o := ValueOnlyFrom(c)
 	select {

core/discov/facade.go

@@ -2,7 +2,7 @@ package discov
 
 import (
 	"github.com/tal-tech/go-zero/core/discov/internal"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 )
 
 type (
@@ -26,7 +26,7 @@ func NewFacade(endpoints []string) Facade {
 
 func (f Facade) Client() internal.EtcdClient {
 	conn, err := f.registry.GetConn(f.endpoints)
-	lang.Must(err)
+	logx.Must(err)
 	return conn
 }
 

core/executors/bulkexecutor_test.go

@@ -12,14 +12,14 @@ func TestBulkExecutor(t *testing.T) {
 	var values []int
 	var lock sync.Mutex
 
-	exeutor := NewBulkExecutor(func(items []interface{}) {
+	executor := NewBulkExecutor(func(items []interface{}) {
 		lock.Lock()
 		values = append(values, len(items))
 		lock.Unlock()
 	}, WithBulkTasks(10), WithBulkInterval(time.Minute))
 
 	for i := 0; i < 50; i++ {
-		exeutor.Add(1)
+		executor.Add(1)
 		time.Sleep(time.Millisecond)
 	}
 
@@ -40,13 +40,13 @@ func TestBulkExecutorFlushInterval(t *testing.T) {
 	var wait sync.WaitGroup
 
 	wait.Add(1)
-	exeutor := NewBulkExecutor(func(items []interface{}) {
+	executor := NewBulkExecutor(func(items []interface{}) {
 		assert.Equal(t, size, len(items))
 		wait.Done()
 	}, WithBulkTasks(caches), WithBulkInterval(time.Millisecond*100))
 
 	for i := 0; i < size; i++ {
-		exeutor.Add(1)
+		executor.Add(1)
 	}
 
 	wait.Wait()

core/executors/chunkexecutor_test.go

@@ -12,14 +12,14 @@ func TestChunkExecutor(t *testing.T) {
 	var values []int
 	var lock sync.Mutex
 
-	exeutor := NewChunkExecutor(func(items []interface{}) {
+	executor := NewChunkExecutor(func(items []interface{}) {
 		lock.Lock()
 		values = append(values, len(items))
 		lock.Unlock()
 	}, WithChunkBytes(10), WithFlushInterval(time.Minute))
 
 	for i := 0; i < 50; i++ {
-		exeutor.Add(1, 1)
+		executor.Add(1, 1)
 		time.Sleep(time.Millisecond)
 	}
 
@@ -40,13 +40,13 @@ func TestChunkExecutorFlushInterval(t *testing.T) {
 	var wait sync.WaitGroup
 
 	wait.Add(1)
-	exeutor := NewChunkExecutor(func(items []interface{}) {
+	executor := NewChunkExecutor(func(items []interface{}) {
 		assert.Equal(t, size, len(items))
 		wait.Done()
 	}, WithChunkBytes(caches), WithFlushInterval(time.Millisecond*100))
 
 	for i := 0; i < size; i++ {
-		exeutor.Add(1, 1)
+		executor.Add(1, 1)
 	}
 
 	wait.Wait()

core/executors/periodicalexecutor.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/tal-tech/go-zero/core/lang"
 	"github.com/tal-tech/go-zero/core/proc"
 	"github.com/tal-tech/go-zero/core/syncx"
 	"github.com/tal-tech/go-zero/core/threading"
@@ -32,19 +33,21 @@ type (
 		container TaskContainer
 		waitGroup sync.WaitGroup
 		// avoid race condition on waitGroup when calling wg.Add/Done/Wait(...)
-		wgBarrier syncx.Barrier
-		guarded   bool
-		newTicker func(duration time.Duration) timex.Ticker
-		lock      sync.Mutex
+		wgBarrier   syncx.Barrier
+		confirmChan chan lang.PlaceholderType
+		guarded     bool
+		newTicker   func(duration time.Duration) timex.Ticker
+		lock        sync.Mutex
 	}
 )
 
 func NewPeriodicalExecutor(interval time.Duration, container TaskContainer) *PeriodicalExecutor {
 	executor := &PeriodicalExecutor{
 		// buffer 1 to let the caller go quickly
-		commander: make(chan interface{}, 1),
-		interval:  interval,
-		container: container,
+		commander:   make(chan interface{}, 1),
+		interval:    interval,
+		container:   container,
+		confirmChan: make(chan lang.PlaceholderType),
 		newTicker: func(d time.Duration) timex.Ticker {
 			return timex.NewTicker(interval)
 		},
@@ -59,10 +62,12 @@ func NewPeriodicalExecutor(interval time.Duration, container TaskContainer) *Per
 func (pe *PeriodicalExecutor) Add(task interface{}) {
 	if vals, ok := pe.addAndCheck(task); ok {
 		pe.commander <- vals
+		<-pe.confirmChan
 	}
 }
 
 func (pe *PeriodicalExecutor) Flush() bool {
+	pe.enterExecution()
 	return pe.executeTasks(func() interface{} {
 		pe.lock.Lock()
 		defer pe.lock.Unlock()
@@ -114,6 +119,8 @@ func (pe *PeriodicalExecutor) backgroundFlush() {
 			select {
 			case vals := <-pe.commander:
 				commanded = true
+				pe.enterExecution()
+				pe.confirmChan <- lang.Placeholder
 				pe.executeTasks(vals)
 				last = timex.Now()
 			case <-ticker.Chan():
@@ -135,13 +142,18 @@ func (pe *PeriodicalExecutor) backgroundFlush() {
 	})
 }
 
-func (pe *PeriodicalExecutor) executeTasks(tasks interface{}) bool {
+func (pe *PeriodicalExecutor) doneExecution() {
+	pe.waitGroup.Done()
+}
+
+func (pe *PeriodicalExecutor) enterExecution() {
 	pe.wgBarrier.Guard(func() {
 		pe.waitGroup.Add(1)
 	})
-	defer pe.wgBarrier.Guard(func() {
-		pe.waitGroup.Done()
-	})
+}
+
+func (pe *PeriodicalExecutor) executeTasks(tasks interface{}) bool {
+	defer pe.doneExecution()
 
 	ok := pe.hasTasks(tasks)
 	if ok {

core/executors/periodicalexecutor_test.go

@@ -106,6 +106,40 @@ func TestPeriodicalExecutor_Bulk(t *testing.T) {
 	lock.Unlock()
 }
 
+func TestPeriodicalExecutor_Wait(t *testing.T) {
+	var lock sync.Mutex
+	executer := NewBulkExecutor(func(tasks []interface{}) {
+		lock.Lock()
+		defer lock.Unlock()
+		time.Sleep(10 * time.Millisecond)
+	}, WithBulkTasks(1), WithBulkInterval(time.Second))
+	for i := 0; i < 10; i++ {
+		executer.Add(1)
+	}
+	executer.Flush()
+	executer.Wait()
+}
+
+func TestPeriodicalExecutor_WaitFast(t *testing.T) {
+	const total = 3
+	var cnt int
+	var lock sync.Mutex
+	executer := NewBulkExecutor(func(tasks []interface{}) {
+		defer func() {
+			cnt++
+		}()
+		lock.Lock()
+		defer lock.Unlock()
+		time.Sleep(10 * time.Millisecond)
+	}, WithBulkTasks(1), WithBulkInterval(10*time.Millisecond))
+	for i := 0; i < total; i++ {
+		executer.Add(2)
+	}
+	executer.Flush()
+	executer.Wait()
+	assert.Equal(t, total, cnt)
+}
+
 // go test -benchtime 10s -bench .
 func BenchmarkExecutor(b *testing.B) {
 	b.ReportAllocs()

core/filex/file_test.go

@@ -15,34 +15,34 @@ const (
 	text          = `first line
 Cum sociis natoque penatibus et magnis dis parturient. Phasellus laoreet lorem vel dolor tempus vehicula. Vivamus sagittis lacus vel augue laoreet rutrum faucibus. Integer legentibus erat a ante historiarum dapibus.
 Quisque ut dolor gravida, placerat libero vel, euismod. Quam temere in vitiis, legem sancimus haerentia. Qui ipsorum lingua Celtae, nostra Galli appellantur. Quis aute iure reprehenderit in voluptate velit esse. Fabio vel iudice vincam, sunt in culpa qui officia. Cras mattis iudicium purus sit amet fermentum.
-Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculus sed magna.
+Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculous sed magna.
 Magna pars studiorum, prodita quaerimus. Cum ceteris in veneratione tui montes, nascetur mus. Morbi odio eros, volutpat ut pharetra vitae, lobortis sed nibh. Plura mihi bona sunt, inclinet, amari petere vellent. Idque Caesaris facere voluntate liceret: sese habere. Tu quoque, Brute, fili mi, nihil timor populi, nihil!
 Tityre, tu patulae recubans sub tegmine fagi dolor. Inmensae subtilitatis, obscuris et malesuada fames. Quae vero auctorem tractata ab fiducia dicuntur.
 Cum sociis natoque penatibus et magnis dis parturient. Phasellus laoreet lorem vel dolor tempus vehicula. Vivamus sagittis lacus vel augue laoreet rutrum faucibus. Integer legentibus erat a ante historiarum dapibus.
 Quisque ut dolor gravida, placerat libero vel, euismod. Quam temere in vitiis, legem sancimus haerentia. Qui ipsorum lingua Celtae, nostra Galli appellantur. Quis aute iure reprehenderit in voluptate velit esse. Fabio vel iudice vincam, sunt in culpa qui officia. Cras mattis iudicium purus sit amet fermentum.
-Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculus sed magna.
+Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculous sed magna.
 Magna pars studiorum, prodita quaerimus. Cum ceteris in veneratione tui montes, nascetur mus. Morbi odio eros, volutpat ut pharetra vitae, lobortis sed nibh. Plura mihi bona sunt, inclinet, amari petere vellent. Idque Caesaris facere voluntate liceret: sese habere. Tu quoque, Brute, fili mi, nihil timor populi, nihil!
 Tityre, tu patulae recubans sub tegmine fagi dolor. Inmensae subtilitatis, obscuris et malesuada fames. Quae vero auctorem tractata ab fiducia dicuntur.
 Cum sociis natoque penatibus et magnis dis parturient. Phasellus laoreet lorem vel dolor tempus vehicula. Vivamus sagittis lacus vel augue laoreet rutrum faucibus. Integer legentibus erat a ante historiarum dapibus.
 Quisque ut dolor gravida, placerat libero vel, euismod. Quam temere in vitiis, legem sancimus haerentia. Qui ipsorum lingua Celtae, nostra Galli appellantur. Quis aute iure reprehenderit in voluptate velit esse. Fabio vel iudice vincam, sunt in culpa qui officia. Cras mattis iudicium purus sit amet fermentum.
-Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculus sed magna.
+Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculous sed magna.
 Magna pars studiorum, prodita quaerimus. Cum ceteris in veneratione tui montes, nascetur mus. Morbi odio eros, volutpat ut pharetra vitae, lobortis sed nibh. Plura mihi bona sunt, inclinet, amari petere vellent. Idque Caesaris facere voluntate liceret: sese habere. Tu quoque, Brute, fili mi, nihil timor populi, nihil!
 Tityre, tu patulae recubans sub tegmine fagi dolor. Inmensae subtilitatis, obscuris et malesuada fames. Quae vero auctorem tractata ab fiducia dicuntur.
 ` + longLine
 	textWithLastNewline = `first line
 Cum sociis natoque penatibus et magnis dis parturient. Phasellus laoreet lorem vel dolor tempus vehicula. Vivamus sagittis lacus vel augue laoreet rutrum faucibus. Integer legentibus erat a ante historiarum dapibus.
 Quisque ut dolor gravida, placerat libero vel, euismod. Quam temere in vitiis, legem sancimus haerentia. Qui ipsorum lingua Celtae, nostra Galli appellantur. Quis aute iure reprehenderit in voluptate velit esse. Fabio vel iudice vincam, sunt in culpa qui officia. Cras mattis iudicium purus sit amet fermentum.
-Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculus sed magna.
+Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculous sed magna.
 Magna pars studiorum, prodita quaerimus. Cum ceteris in veneratione tui montes, nascetur mus. Morbi odio eros, volutpat ut pharetra vitae, lobortis sed nibh. Plura mihi bona sunt, inclinet, amari petere vellent. Idque Caesaris facere voluntate liceret: sese habere. Tu quoque, Brute, fili mi, nihil timor populi, nihil!
 Tityre, tu patulae recubans sub tegmine fagi dolor. Inmensae subtilitatis, obscuris et malesuada fames. Quae vero auctorem tractata ab fiducia dicuntur.
 Cum sociis natoque penatibus et magnis dis parturient. Phasellus laoreet lorem vel dolor tempus vehicula. Vivamus sagittis lacus vel augue laoreet rutrum faucibus. Integer legentibus erat a ante historiarum dapibus.
 Quisque ut dolor gravida, placerat libero vel, euismod. Quam temere in vitiis, legem sancimus haerentia. Qui ipsorum lingua Celtae, nostra Galli appellantur. Quis aute iure reprehenderit in voluptate velit esse. Fabio vel iudice vincam, sunt in culpa qui officia. Cras mattis iudicium purus sit amet fermentum.
-Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculus sed magna.
+Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculous sed magna.
 Magna pars studiorum, prodita quaerimus. Cum ceteris in veneratione tui montes, nascetur mus. Morbi odio eros, volutpat ut pharetra vitae, lobortis sed nibh. Plura mihi bona sunt, inclinet, amari petere vellent. Idque Caesaris facere voluntate liceret: sese habere. Tu quoque, Brute, fili mi, nihil timor populi, nihil!
 Tityre, tu patulae recubans sub tegmine fagi dolor. Inmensae subtilitatis, obscuris et malesuada fames. Quae vero auctorem tractata ab fiducia dicuntur.
 Cum sociis natoque penatibus et magnis dis parturient. Phasellus laoreet lorem vel dolor tempus vehicula. Vivamus sagittis lacus vel augue laoreet rutrum faucibus. Integer legentibus erat a ante historiarum dapibus.
 Quisque ut dolor gravida, placerat libero vel, euismod. Quam temere in vitiis, legem sancimus haerentia. Qui ipsorum lingua Celtae, nostra Galli appellantur. Quis aute iure reprehenderit in voluptate velit esse. Fabio vel iudice vincam, sunt in culpa qui officia. Cras mattis iudicium purus sit amet fermentum.
-Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculus sed magna.
+Quo usque tandem abutere, Catilina, patientia nostra? Gallia est omnis divisa in partes tres, quarum. Quam diu etiam furor iste tuus nos eludet? Quid securi etiam tamquam eu fugiat nulla pariatur. Curabitur blandit tempus ardua ridiculous sed magna.
 Magna pars studiorum, prodita quaerimus. Cum ceteris in veneratione tui montes, nascetur mus. Morbi odio eros, volutpat ut pharetra vitae, lobortis sed nibh. Plura mihi bona sunt, inclinet, amari petere vellent. Idque Caesaris facere voluntate liceret: sese habere. Tu quoque, Brute, fili mi, nihil timor populi, nihil!
 Tityre, tu patulae recubans sub tegmine fagi dolor. Inmensae subtilitatis, obscuris et malesuada fames. Quae vero auctorem tractata ab fiducia dicuntur.
 ` + longLine + "\n"

core/fx/fn.go

@@ -49,7 +49,7 @@ func From(generate GenerateFunc) Stream {
 	return Range(source)
 }
 
-// Just converts the given arbitary items to a Stream.
+// Just converts the given arbitrary items to a Stream.
 func Just(items ...interface{}) Stream {
 	source := make(chan interface{}, len(items))
 	for _, item := range items {
@@ -195,7 +195,7 @@ func (p Stream) Merge() Stream {
 	return Range(source)
 }
 
-// Parallel applies the given ParallenFunc to each item concurrently with given number of workers.
+// Parallel applies the given ParallelFunc to each item concurrently with given number of workers.
 func (p Stream) Parallel(fn ParallelFunc, opts ...Option) {
 	p.Walk(func(item interface{}, pipe chan<- interface{}) {
 		fn(item)

core/lang/lang.go

@@ -1,16 +1,8 @@
 package lang
 
-import "log"
-
 var Placeholder PlaceholderType
 
 type (
 	GenericType     = interface{}
 	PlaceholderType = struct{}
 )
-
-func Must(err error) {
-	if err != nil {
-		log.Fatal(err)
-	}
-}

core/lang/lang_test.go

@@ -1,7 +0,0 @@
-package lang
-
-import "testing"
-
-func TestMust(t *testing.T) {
-	Must(nil)
-}

core/logx/logs.go

@@ -17,7 +17,6 @@ import (
 	"sync/atomic"
 
 	"github.com/tal-tech/go-zero/core/iox"
-	"github.com/tal-tech/go-zero/core/lang"
 	"github.com/tal-tech/go-zero/core/sysx"
 	"github.com/tal-tech/go-zero/core/timex"
 )
@@ -46,6 +45,7 @@ const (
 	levelInfo   = "info"
 	levelError  = "error"
 	levelSevere = "severe"
+	levelFatal  = "fatal"
 	levelSlow   = "slow"
 	levelStat   = "stat"
 
@@ -100,7 +100,7 @@ type (
 )
 
 func MustSetup(c LogConf) {
-	lang.Must(SetUp(c))
+	Must(SetUp(c))
 }
 
 // SetUp sets up the logx. If already set up, just return nil.
@@ -210,6 +210,14 @@ func Infof(format string, v ...interface{}) {
 	infoSync(fmt.Sprintf(format, v...))
 }
 
+func Must(err error) {
+	if err != nil {
+		msg := formatWithCaller(err.Error(), 3)
+		output(severeLog, levelFatal, msg)
+		os.Exit(1)
+	}
+}
+
 func SetLevel(level uint32) {
 	atomic.StoreUint32(&logLevel, level)
 }

core/logx/logs_test.go

@@ -131,6 +131,10 @@ func TestSetLevelWithDuration(t *testing.T) {
 	assert.Equal(t, 0, writer.builder.Len())
 }
 
+func TestMustNil(t *testing.T) {
+	Must(nil)
+}
+
 func BenchmarkCopyByteSliceAppend(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		var buf []byte

core/queue/balancedpusher.go

@@ -0,0 +1,44 @@
+package queue
+
+import (
+	"errors"
+	"sync/atomic"
+
+	"github.com/tal-tech/go-zero/core/logx"
+)
+
+var ErrNoAvailablePusher = errors.New("no available pusher")
+
+type BalancedPusher struct {
+	name    string
+	pushers []Pusher
+	index   uint64
+}
+
+func NewBalancedPusher(pushers []Pusher) Pusher {
+	return &BalancedPusher{
+		name:    generateName(pushers),
+		pushers: pushers,
+	}
+}
+
+func (pusher *BalancedPusher) Name() string {
+	return pusher.name
+}
+
+func (pusher *BalancedPusher) Push(message string) error {
+	size := len(pusher.pushers)
+
+	for i := 0; i < size; i++ {
+		index := atomic.AddUint64(&pusher.index, 1) % uint64(size)
+		target := pusher.pushers[index]
+
+		if err := target.Push(message); err != nil {
+			logx.Error(err)
+		} else {
+			return nil
+		}
+	}
+
+	return ErrNoAvailablePusher
+}

core/queue/balancedpusher_test.go

@@ -0,0 +1,43 @@
+package queue
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBalancedQueuePusher(t *testing.T) {
+	const numPushers = 100
+	var pushers []Pusher
+	var mockedPushers []*mockedPusher
+	for i := 0; i < numPushers; i++ {
+		p := &mockedPusher{
+			name: "pusher:" + strconv.Itoa(i),
+		}
+		pushers = append(pushers, p)
+		mockedPushers = append(mockedPushers, p)
+	}
+
+	pusher := NewBalancedPusher(pushers)
+	assert.True(t, len(pusher.Name()) > 0)
+
+	for i := 0; i < numPushers*1000; i++ {
+		assert.Nil(t, pusher.Push("item"))
+	}
+
+	var counts []int
+	for _, p := range mockedPushers {
+		counts = append(counts, p.count)
+	}
+	mean := calcMean(counts)
+	variance := calcVariance(mean, counts)
+	assert.True(t, variance < 100, fmt.Sprintf("too big variance - %.2f", variance))
+}
+
+func TestBalancedQueuePusher_NoAvailable(t *testing.T) {
+	pusher := NewBalancedPusher(nil)
+	assert.True(t, len(pusher.Name()) == 0)
+	assert.Equal(t, ErrNoAvailablePusher, pusher.Push("item"))
+}

core/queue/consumer.go

@@ -0,0 +1,10 @@
+package queue
+
+type (
+	Consumer interface {
+		Consume(string) error
+		OnEvent(event interface{})
+	}
+
+	ConsumerFactory func() (Consumer, error)
+)

core/queue/messagequeue.go

@@ -0,0 +1,6 @@
+package queue
+
+type MessageQueue interface {
+	Start()
+	Stop()
+}

core/queue/multipusher.go

@@ -0,0 +1,31 @@
+package queue
+
+import "github.com/tal-tech/go-zero/core/errorx"
+
+type MultiPusher struct {
+	name    string
+	pushers []Pusher
+}
+
+func NewMultiPusher(pushers []Pusher) Pusher {
+	return &MultiPusher{
+		name:    generateName(pushers),
+		pushers: pushers,
+	}
+}
+
+func (pusher *MultiPusher) Name() string {
+	return pusher.name
+}
+
+func (pusher *MultiPusher) Push(message string) error {
+	var batchError errorx.BatchError
+
+	for _, each := range pusher.pushers {
+		if err := each.Push(message); err != nil {
+			batchError.Add(err)
+		}
+	}
+
+	return batchError.Err()
+}

core/queue/multipusher_test.go

@@ -0,0 +1,39 @@
+package queue
+
+import (
+	"fmt"
+	"math"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMultiQueuePusher(t *testing.T) {
+	const numPushers = 100
+	var pushers []Pusher
+	var mockedPushers []*mockedPusher
+	for i := 0; i < numPushers; i++ {
+		p := &mockedPusher{
+			name: "pusher:" + strconv.Itoa(i),
+		}
+		pushers = append(pushers, p)
+		mockedPushers = append(mockedPushers, p)
+	}
+
+	pusher := NewMultiPusher(pushers)
+	assert.True(t, len(pusher.Name()) > 0)
+
+	for i := 0; i < 1000; i++ {
+		_ = pusher.Push("item")
+	}
+
+	var counts []int
+	for _, p := range mockedPushers {
+		counts = append(counts, p.count)
+	}
+	mean := calcMean(counts)
+	variance := calcVariance(mean, counts)
+	assert.True(t, math.Abs(mean-1000*(1-failProba)) < 10)
+	assert.True(t, variance < 100, fmt.Sprintf("too big variance - %.2f", variance))
+}

core/queue/producer.go

@@ -0,0 +1,15 @@
+package queue
+
+type (
+	Producer interface {
+		AddListener(listener ProduceListener)
+		Produce() (string, bool)
+	}
+
+	ProduceListener interface {
+		OnProducerPause()
+		OnProducerResume()
+	}
+
+	ProducerFactory func() (Producer, error)
+)

core/queue/queue.go

@@ -0,0 +1,239 @@
+package queue
+
+import (
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/tal-tech/go-zero/core/logx"
+	"github.com/tal-tech/go-zero/core/rescue"
+	"github.com/tal-tech/go-zero/core/stat"
+	"github.com/tal-tech/go-zero/core/threading"
+	"github.com/tal-tech/go-zero/core/timex"
+)
+
+const queueName = "queue"
+
+type (
+	Queue struct {
+		name                 string
+		metrics              *stat.Metrics
+		producerFactory      ProducerFactory
+		producerRoutineGroup *threading.RoutineGroup
+		consumerFactory      ConsumerFactory
+		consumerRoutineGroup *threading.RoutineGroup
+		producerCount        int
+		consumerCount        int
+		active               int32
+		channel              chan string
+		quit                 chan struct{}
+		listeners            []Listener
+		eventLock            sync.Mutex
+		eventChannels        []chan interface{}
+	}
+
+	Listener interface {
+		OnPause()
+		OnResume()
+	}
+
+	Poller interface {
+		Name() string
+		Poll() string
+	}
+
+	Pusher interface {
+		Name() string
+		Push(string) error
+	}
+)
+
+func NewQueue(producerFactory ProducerFactory, consumerFactory ConsumerFactory) *Queue {
+	queue := &Queue{
+		metrics:              stat.NewMetrics(queueName),
+		producerFactory:      producerFactory,
+		producerRoutineGroup: threading.NewRoutineGroup(),
+		consumerFactory:      consumerFactory,
+		consumerRoutineGroup: threading.NewRoutineGroup(),
+		producerCount:        runtime.NumCPU(),
+		consumerCount:        runtime.NumCPU() << 1,
+		channel:              make(chan string),
+		quit:                 make(chan struct{}),
+	}
+	queue.SetName(queueName)
+
+	return queue
+}
+
+func (queue *Queue) AddListener(listener Listener) {
+	queue.listeners = append(queue.listeners, listener)
+}
+
+func (queue *Queue) Broadcast(message interface{}) {
+	go func() {
+		queue.eventLock.Lock()
+		defer queue.eventLock.Unlock()
+
+		for _, channel := range queue.eventChannels {
+			channel <- message
+		}
+	}()
+}
+
+func (queue *Queue) SetName(name string) {
+	queue.name = name
+	queue.metrics.SetName(name)
+}
+
+func (queue *Queue) SetNumConsumer(count int) {
+	queue.consumerCount = count
+}
+
+func (queue *Queue) SetNumProducer(count int) {
+	queue.producerCount = count
+}
+
+func (queue *Queue) Start() {
+	queue.startProducers(queue.producerCount)
+	queue.startConsumers(queue.consumerCount)
+
+	queue.producerRoutineGroup.Wait()
+	close(queue.channel)
+	queue.consumerRoutineGroup.Wait()
+}
+
+func (queue *Queue) Stop() {
+	close(queue.quit)
+}
+
+func (queue *Queue) consume(eventChan chan interface{}) {
+	var consumer Consumer
+
+	for {
+		var err error
+		if consumer, err = queue.consumerFactory(); err != nil {
+			logx.Errorf("Error on creating consumer: %v", err)
+			time.Sleep(time.Second)
+		} else {
+			break
+		}
+	}
+
+	for {
+		select {
+		case message, ok := <-queue.channel:
+			if ok {
+				queue.consumeOne(consumer, message)
+			} else {
+				logx.Info("Task channel was closed, quitting consumer...")
+				return
+			}
+		case event := <-eventChan:
+			consumer.OnEvent(event)
+		}
+	}
+}
+
+func (queue *Queue) consumeOne(consumer Consumer, message string) {
+	threading.RunSafe(func() {
+		startTime := timex.Now()
+		defer func() {
+			duration := timex.Since(startTime)
+			queue.metrics.Add(stat.Task{
+				Duration: duration,
+			})
+			logx.WithDuration(duration).Infof("%s", message)
+		}()
+
+		if err := consumer.Consume(message); err != nil {
+			logx.Errorf("Error occurred while consuming %v: %v", message, err)
+		}
+	})
+}
+
+func (queue *Queue) pause() {
+	for _, listener := range queue.listeners {
+		listener.OnPause()
+	}
+}
+
+func (queue *Queue) produce() {
+	var producer Producer
+
+	for {
+		var err error
+		if producer, err = queue.producerFactory(); err != nil {
+			logx.Errorf("Error on creating producer: %v", err)
+			time.Sleep(time.Second)
+		} else {
+			break
+		}
+	}
+
+	atomic.AddInt32(&queue.active, 1)
+	producer.AddListener(routineListener{
+		queue: queue,
+	})
+
+	for {
+		select {
+		case <-queue.quit:
+			logx.Info("Quitting producer")
+			return
+		default:
+			if v, ok := queue.produceOne(producer); ok {
+				queue.channel <- v
+			}
+		}
+	}
+}
+
+func (queue *Queue) produceOne(producer Producer) (string, bool) {
+	// avoid panic quit the producer, just log it and continue
+	defer rescue.Recover()
+
+	return producer.Produce()
+}
+
+func (queue *Queue) resume() {
+	for _, listener := range queue.listeners {
+		listener.OnResume()
+	}
+}
+
+func (queue *Queue) startConsumers(number int) {
+	for i := 0; i < number; i++ {
+		eventChan := make(chan interface{})
+		queue.eventLock.Lock()
+		queue.eventChannels = append(queue.eventChannels, eventChan)
+		queue.eventLock.Unlock()
+		queue.consumerRoutineGroup.Run(func() {
+			queue.consume(eventChan)
+		})
+	}
+}
+
+func (queue *Queue) startProducers(number int) {
+	for i := 0; i < number; i++ {
+		queue.producerRoutineGroup.Run(func() {
+			queue.produce()
+		})
+	}
+}
+
+type routineListener struct {
+	queue *Queue
+}
+
+func (rl routineListener) OnProducerPause() {
+	if atomic.AddInt32(&rl.queue.active, -1) <= 0 {
+		rl.queue.pause()
+	}
+}
+
+func (rl routineListener) OnProducerResume() {
+	if atomic.AddInt32(&rl.queue.active, 1) == 1 {
+		rl.queue.resume()
+	}
+}

core/queue/queue_test.go

@@ -0,0 +1,94 @@
+package queue
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	consumers = 4
+	rounds    = 100
+)
+
+func TestQueue(t *testing.T) {
+	producer := newMockedProducer(rounds)
+	consumer := newMockedConsumer()
+	consumer.wait.Add(consumers)
+	q := NewQueue(func() (Producer, error) {
+		return producer, nil
+	}, func() (Consumer, error) {
+		return consumer, nil
+	})
+	q.AddListener(new(mockedListener))
+	q.SetName("mockqueue")
+	q.SetNumConsumer(consumers)
+	q.SetNumProducer(1)
+	q.pause()
+	q.resume()
+	go func() {
+		producer.wait.Wait()
+		q.Stop()
+	}()
+	q.Start()
+	assert.Equal(t, int32(rounds), atomic.LoadInt32(&consumer.count))
+}
+
+type mockedConsumer struct {
+	count  int32
+	events int32
+	wait   sync.WaitGroup
+}
+
+func newMockedConsumer() *mockedConsumer {
+	return new(mockedConsumer)
+}
+
+func (c *mockedConsumer) Consume(string) error {
+	atomic.AddInt32(&c.count, 1)
+	return nil
+}
+
+func (c *mockedConsumer) OnEvent(interface{}) {
+	if atomic.AddInt32(&c.events, 1) <= consumers {
+		c.wait.Done()
+	}
+}
+
+type mockedProducer struct {
+	total int32
+	count int32
+	wait  sync.WaitGroup
+}
+
+func newMockedProducer(total int32) *mockedProducer {
+	p := new(mockedProducer)
+	p.total = total
+	p.wait.Add(int(total))
+	return p
+}
+
+func (p *mockedProducer) AddListener(listener ProduceListener) {
+}
+
+func (p *mockedProducer) Produce() (string, bool) {
+	if atomic.AddInt32(&p.count, 1) <= p.total {
+		p.wait.Done()
+		return "item", true
+	} else {
+		time.Sleep(time.Second)
+		return "", false
+	}
+}
+
+type mockedListener struct {
+}
+
+func (l *mockedListener) OnPause() {
+}
+
+func (l *mockedListener) OnResume() {
+}

core/queue/util.go

@@ -0,0 +1,12 @@
+package queue
+
+import "strings"
+
+func generateName(pushers []Pusher) string {
+	names := make([]string, len(pushers))
+	for i, pusher := range pushers {
+		names[i] = pusher.Name()
+	}
+
+	return strings.Join(names, ",")
+}

core/queue/util_test.go

@@ -0,0 +1,77 @@
+package queue
+
+import (
+	"errors"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/tal-tech/go-zero/core/logx"
+	"github.com/tal-tech/go-zero/core/mathx"
+)
+
+var (
+	proba     = mathx.NewProba()
+	failProba = 0.01
+)
+
+func init() {
+	logx.Disable()
+}
+
+func TestGenerateName(t *testing.T) {
+	pushers := []Pusher{
+		&mockedPusher{name: "first"},
+		&mockedPusher{name: "second"},
+		&mockedPusher{name: "third"},
+	}
+
+	assert.Equal(t, "first,second,third", generateName(pushers))
+}
+
+func TestGenerateNameNil(t *testing.T) {
+	var pushers []Pusher
+	assert.Equal(t, "", generateName(pushers))
+}
+
+func calcMean(vals []int) float64 {
+	if len(vals) == 0 {
+		return 0
+	}
+
+	var result float64
+	for _, val := range vals {
+		result += float64(val)
+	}
+	return result / float64(len(vals))
+}
+
+func calcVariance(mean float64, vals []int) float64 {
+	if len(vals) == 0 {
+		return 0
+	}
+
+	var result float64
+	for _, val := range vals {
+		result += math.Pow(float64(val)-mean, 2)
+	}
+	return result / float64(len(vals))
+}
+
+type mockedPusher struct {
+	name  string
+	count int
+}
+
+func (p *mockedPusher) Name() string {
+	return p.name
+}
+
+func (p *mockedPusher) Push(s string) error {
+	if proba.TrueOnProba(failProba) {
+		return errors.New("dummy")
+	}
+
+	p.count++
+	return nil
+}

core/stat/alert.go

@@ -14,7 +14,6 @@ import (
 	"github.com/tal-tech/go-zero/core/proc"
 	"github.com/tal-tech/go-zero/core/sysx"
 	"github.com/tal-tech/go-zero/core/timex"
-	"github.com/tal-tech/go-zero/core/utils"
 )
 
 const (
@@ -24,7 +23,7 @@ const (
 )
 
 var (
-	reporter     = utils.Report
+	reporter     func(string)
 	lock         sync.RWMutex
 	lessExecutor = executors.NewLessExecutor(time.Minute * 5)
 	dropped      int32

core/stat/internal/cpu_linux.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/tal-tech/go-zero/core/iox"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 )
 
 const (
@@ -24,17 +24,17 @@ var (
 
 func init() {
 	cpus, err := perCpuUsage()
-	lang.Must(err)
+	logx.Must(err)
 	cores = uint64(len(cpus))
 
 	sets, err := cpuSets()
-	lang.Must(err)
+	logx.Must(err)
 	quota = float64(len(sets))
 	cq, err := cpuQuota()
 	if err == nil {
 		if cq != -1 {
 			period, err := cpuPeriod()
-			lang.Must(err)
+			logx.Must(err)
 
 			limit := float64(cq) / float64(period)
 			if limit < quota {
@@ -44,10 +44,10 @@ func init() {
 	}
 
 	preSystem, err = systemCpuUsage()
-	lang.Must(err)
+	logx.Must(err)
 
 	preTotal, err = totalCpuUsage()
-	lang.Must(err)
+	logx.Must(err)
 }
 
 func RefreshCpu() uint64 {

core/stores/internal/cleaner.go

@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/tal-tech/go-zero/core/collection"
-	"github.com/tal-tech/go-zero/core/lang"
 	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/proc"
 	"github.com/tal-tech/go-zero/core/stat"
@@ -33,7 +32,7 @@ type delayTask struct {
 func init() {
 	var err error
 	timingWheel, err = collection.NewTimingWheel(time.Second, timingWheelSlots, clean)
-	lang.Must(err)
+	logx.Must(err)
 
 	proc.AddShutdownListener(func() {
 		timingWheel.Drain(clean)

core/stores/kv/store_test.go

@@ -212,10 +212,12 @@ func TestRedis_Persist(t *testing.T) {
 		assert.Nil(t, err)
 		assert.False(t, ok)
 		err = client.Expire("key", 5)
+		assert.Nil(t, err)
 		ok, err = client.Persist("key")
 		assert.Nil(t, err)
 		assert.True(t, ok)
 		err = client.Expireat("key", time.Now().Unix()+5)
+		assert.Nil(t, err)
 		ok, err = client.Persist("key")
 		assert.Nil(t, err)
 		assert.True(t, ok)
@@ -379,7 +381,7 @@ func TestRedis_SortedSet(t *testing.T) {
 		rank, err := client.Zrank("key", "value2")
 		assert.Nil(t, err)
 		assert.Equal(t, int64(1), rank)
-		rank, err = client.Zrank("key", "value4")
+		_, err = client.Zrank("key", "value4")
 		assert.Equal(t, redis.Nil, err)
 		num, err := client.Zrem("key", "value2", "value3")
 		assert.Nil(t, err)

core/stores/redis/redis_test.go

@@ -249,10 +249,12 @@ func TestRedis_Persist(t *testing.T) {
 		assert.Nil(t, err)
 		assert.False(t, ok)
 		err = client.Expire("key", 5)
+		assert.Nil(t, err)
 		ok, err = client.Persist("key")
 		assert.Nil(t, err)
 		assert.True(t, ok)
 		err = client.Expireat("key", time.Now().Unix()+5)
+		assert.Nil(t, err)
 		ok, err = client.Persist("key")
 		assert.Nil(t, err)
 		assert.True(t, ok)
@@ -447,7 +449,7 @@ func TestRedis_SortedSet(t *testing.T) {
 		rank, err := client.Zrank("key", "value2")
 		assert.Nil(t, err)
 		assert.Equal(t, int64(1), rank)
-		rank, err = client.Zrank("key", "value4")
+		_, err = client.Zrank("key", "value4")
 		assert.Equal(t, Nil, err)
 		num, err := client.Zrem("key", "value2", "value3")
 		assert.Nil(t, err)
@@ -558,6 +560,7 @@ func TestRedis_Pipelined(t *testing.T) {
 		assert.Nil(t, err)
 		assert.Equal(t, "1", value)
 		score, err := client.Zscore("zadd", "zadd")
+		assert.Nil(t, err)
 		assert.Equal(t, int64(12), score)
 	})
 }

core/stringx/replacer.go

@@ -63,11 +63,6 @@ func (r *replacer) Replace(text string) string {
 			i = j - 1
 			builder.WriteString(r.mapping[string(chars[start:end])])
 		} else {
-			if j < size {
-				end = j + 1
-			} else {
-				end = size
-			}
 			builder.WriteRune(chars[i])
 		}
 		start = -1

core/stringx/trie.go

@@ -2,7 +2,11 @@ package stringx
 
 import "github.com/tal-tech/go-zero/core/lang"
 
+const defaultMask = '*'
+
 type (
+	TrieOption func(trie *trieNode)
+
 	Trie interface {
 		Filter(text string) (string, []string, bool)
 		FindKeywords(text string) []string
@@ -10,6 +14,7 @@ type (
 
 	trieNode struct {
 		node
+		mask rune
 	}
 
 	scope struct {
@@ -18,8 +23,15 @@ type (
 	}
 )
 
-func NewTrie(words []string) Trie {
+func NewTrie(words []string, opts ...TrieOption) Trie {
 	n := new(trieNode)
+
+	for _, opt := range opts {
+		opt(n)
+	}
+	if n.mask == 0 {
+		n.mask = defaultMask
+	}
 	for _, word := range words {
 		n.add(word)
 	}
@@ -114,6 +126,12 @@ func (n *trieNode) findKeywordScopes(chars []rune) []scope {
 
 func (n *trieNode) replaceWithAsterisk(chars []rune, start, stop int) {
 	for i := start; i < stop; i++ {
-		chars[i] = '*'
+		chars[i] = n.mask
+	}
+}
+
+func WithMask(mask rune) TrieOption {
+	return func(n *trieNode) {
+		n.mask = mask
 	}
 }

core/stringx/trie_test.go

@@ -109,25 +109,25 @@ func TestTrie(t *testing.T) {
 func TestTrieSingleWord(t *testing.T) {
 	trie := NewTrie([]string{
 		"闹",
-	})
+	}, WithMask('#'))
 	output, keywords, ok := trie.Filter("今晚真热闹")
 	assert.ElementsMatch(t, []string{"闹"}, keywords)
 	assert.True(t, ok)
-	assert.Equal(t, "今晚真热*", output)
+	assert.Equal(t, "今晚真热#", output)
 }
 
 func TestTrieOverlap(t *testing.T) {
 	trie := NewTrie([]string{
 		"一二三四五",
 		"二三四五六七八",
-	})
+	}, WithMask('#'))
 	output, keywords, ok := trie.Filter("零一二三四五六七八九十")
 	assert.ElementsMatch(t, []string{
 		"一二三四五",
 		"二三四五六七八",
 	}, keywords)
 	assert.True(t, ok)
-	assert.Equal(t, "零********九十", output)
+	assert.Equal(t, "零########九十", output)
 }
 
 func TestTrieNested(t *testing.T) {
@@ -135,7 +135,7 @@ func TestTrieNested(t *testing.T) {
 		"一二三",
 		"一二三四五",
 		"一二三四五六七八",
-	})
+	}, WithMask('#'))
 	output, keywords, ok := trie.Filter("零一二三四五六七八九十")
 	assert.ElementsMatch(t, []string{
 		"一二三",
@@ -143,7 +143,7 @@ func TestTrieNested(t *testing.T) {
 		"一二三四五六七八",
 	}, keywords)
 	assert.True(t, ok)
-	assert.Equal(t, "零********九十", output)
+	assert.Equal(t, "零########九十", output)
 }
 
 func BenchmarkTrie(b *testing.B) {

core/sysx/host.go

@@ -3,7 +3,7 @@ package sysx
 import (
 	"os"
 
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/stringx"
 )
 
 var hostname string
@@ -11,7 +11,9 @@ var hostname string
 func init() {
 	var err error
 	hostname, err = os.Hostname()
-	lang.Must(err)
+	if err != nil {
+		hostname = stringx.RandId()
+	}
 }
 
 func Hostname() string {

core/utils/report.go

@@ -1,5 +0,0 @@
-package utils
-
-func Report(content string) {
-	// TODO: implement the report method
-}

doc/goctl.md

@@ -3,8 +3,8 @@
 ## goctl用途
 * 定义api请求
 * 根据定义的api自动生成golang(后端), java(iOS & Android), typescript(web & 晓程序)，dart(flutter)
-* 生成MySQL CURD (https://goctl.xiaoheiban.cn)
-*  生成MongoDB CURD (https://goctl.xiaoheiban.cn)
+* 生成MySQL CURD+Cache
+*  生成MongoDB CURD+Cache
 
 ## goctl使用说明
 #### goctl参数说明
@@ -179,23 +179,31 @@ service user-api {
   * 在定义的get/post/put/delete等请求的handler和logic里增加处理业务逻辑的代码
 
 #### 根据定义好的api文件生成java代码
-	`goctl api java -api user/user.api -dir ./src`
+```shell
+goctl api java -api user/user.api -dir ./src
+```
 
 #### 根据定义好的api文件生成typescript代码
-	`goctl api ts -api user/user.api -dir ./src -webapi ***`
-	
-	ts需要指定webapi所在目录
+```shell
+goctl api ts -api user/user.api -dir ./src -webapi ***
+
+ts需要指定webapi所在目录
+```
 
 #### 根据定义好的api文件生成Dart代码
-	`goctl api dart -api user/user.api -dir ./src`
+```shell
+goctl api dart -api user/user.api -dir ./src
+```
 
 ## 根据定义好的简单go文件生成mongo代码文件(仅限golang使用)  
-    `goctl model mongo -src {{yourDir}}/xiao/service/xhb/user/model/usermodel.go -cache yes`
-    
-    -src需要提供简单的usermodel.go文件，里面只需要提供一个结构体即可
-    -cache 控制是否需要缓存 yes=需要 no=不需要
-    src 示例代码如下
-  ```
+```shell
+goctl model mongo -src {{yourDir}}/xiao/service/xhb/user/model/usermodel.go -cache yes
+
+-src需要提供简单的usermodel.go文件，里面只需要提供一个结构体即可
+-cache 控制是否需要缓存 yes=需要 no=不需要
+src 示例代码如下
+```
+  ```go
 package model
     
 type User struct {
@@ -261,5 +269,4 @@ type User struct {
     │   └── test.go [强制覆盖更新]
     └── test.proto
   ```
-  - 注意 ：目前rpc目录生成的proto文件暂不支持import外部proto文件   
-* 如有不理解的地方，随时问Kim/Kevin
+  - 注意 ：目前rpc目录生成的proto文件暂不支持import外部proto文件   

doc/keywords.md

@@ -0,0 +1,79 @@
+# 高效的关键词替换和敏感词过滤工具
+
+## 1. 算法介绍
+
+利用高效的Trie树建立关键词树，如下图所示，然后依次查找字符串中的相连字符是否形成树的一条路径
+
+<img src="images/trie.png" alt="trie" width="350" />
+
+发现掘金上[这篇文章](https://juejin.im/post/6844903750490914829)写的比较详细，可以一读，具体原理在此不详述。
+
+## 2. 关键词替换
+
+```go
+replacer := stringx.NewReplacer(map[string]string{
+  "PHP": "PPT",
+  "世界上": "吹牛",
+})
+fmt.Println(replacer.Replace("PHP是世界上最好的语言！"))
+```
+
+可以得到：
+```
+PPT是吹牛最好的语言！
+```
+
+示例代码见`example/stringx/replace/replace.go`
+
+## 3. 查找敏感词
+
+```go
+filter := stringx.NewTrie([]string{
+  "AV演员",
+  "苍井空",
+  "AV",
+  "日本AV女优",
+  "AV演员色情",
+})
+keywords := filter.FindKeywords("日本AV演员兼电视、电影演员。苍井空AV女优是xx出道, 日本AV女优们最精彩的表演是AV演员色情表演")
+fmt.Println(keywords)
+```
+
+可以得到：
+
+```
+[苍井空 日本AV女优 AV演员色情 AV AV演员]
+```
+
+## 4. 敏感词过滤
+
+```go
+filter := stringx.NewTrie([]string{
+  "AV演员",
+  "苍井空",
+  "AV",
+  "日本AV女优",
+  "AV演员色情",
+}, stringx.WithMask('?')) // 默认替换为*
+safe, keywords, found := filter.Filter("日本AV演员兼电视、电影演员。苍井空AV女优是xx出道, 日本AV女优们最精彩的表演是AV演员色情表演")
+fmt.Println(safe)
+fmt.Println(keywords)
+fmt.Println(found)
+```
+
+可以得到：
+
+```
+日本????兼电视、电影演员。?????女优是xx出道, ??????们最精彩的表演是??????表演
+[苍井空 日本AV女优 AV演员色情 AV AV演员]
+true
+```
+
+示例代码见`example/stringx/filter/filter.go`
+
+## 5. Benchmark
+
+| Sentences | Keywords | Regex    | go-zero |
+| --------- | -------- | -------- | ------- |
+| 10000     | 10000    | 16min10s | 27.2ms  |
+

example/breaker/main.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/tal-tech/go-zero/core/breaker"
 	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	"gopkg.in/cheggaaa/pb.v1"
 )
 
@@ -99,7 +100,7 @@ func main() {
 
 	gb := breaker.NewBreaker()
 	fp, err := os.Create("result.csv")
-	lang.Must(err)
+	logx.Must(err)
 	defer fp.Close()
 	fmt.Fprintln(fp, "seconds,state,googleCalls,netflixCalls")
 

example/etcd/sub/sub.go

@@ -5,12 +5,12 @@ import (
 	"time"
 
 	"github.com/tal-tech/go-zero/core/discov"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 )
 
 func main() {
 	sub, err := discov.NewSubscriber([]string{"etcd.discovery:2379"}, "028F2C35852D", discov.Exclusive())
-	lang.Must(err)
+	logx.Must(err)
 
 	ticker := time.NewTicker(time.Second * 3)
 	defer ticker.Stop()

example/http/breaker/client/client.go

@@ -9,12 +9,13 @@ import (
 	"time"
 
 	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/threading"
 	"gopkg.in/cheggaaa/pb.v1"
 )
 
 var (
-	freq     = flag.Int("freq", 100, "frequence")
+	freq     = flag.Int("freq", 100, "frequency")
 	duration = flag.String("duration", "10s", "duration")
 )
 
@@ -83,8 +84,8 @@ func (m *metric) reset() counting {
 	return result
 }
 
-func runRequests(url string, frequence int, metrics *metric, done <-chan lang.PlaceholderType) {
-	ticker := time.NewTicker(time.Second / time.Duration(frequence))
+func runRequests(url string, frequency int, metrics *metric, done <-chan lang.PlaceholderType) {
+	ticker := time.NewTicker(time.Second / time.Duration(frequency))
 	defer ticker.Stop()
 
 	for {
@@ -119,14 +120,14 @@ func main() {
 	flag.Parse()
 
 	fp, err := os.Create("result.csv")
-	lang.Must(err)
+	logx.Must(err)
 	defer fp.Close()
 	fmt.Fprintln(fp, "seconds,goodOk,goodFail,goodReject,goodErrs,goodUnknowns,goodDropRatio,"+
 		"heavyOk,heavyFail,heavyReject,heavyErrs,heavyUnknowns,heavyDropRatio")
 
 	var gm, hm metric
 	dur, err := time.ParseDuration(*duration)
-	lang.Must(err)
+	logx.Must(err)
 	done := make(chan lang.PlaceholderType)
 	group := threading.NewRoutineGroup()
 	group.RunSafe(func() {

example/load/main.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/tal-tech/go-zero/core/collection"
 	"github.com/tal-tech/go-zero/core/executors"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/syncx"
 	"gopkg.in/cheggaaa/pb.v1"
 )
@@ -47,7 +47,7 @@ func main() {
 	lessWriter = executors.NewLessExecutor(interval * total / 100)
 
 	fp, err := os.Create("result.csv")
-	lang.Must(err)
+	logx.Must(err)
 	defer fp.Close()
 	fmt.Fprintln(fp, "second,maxFlight,flying,agressiveAvgFlying,lazyAvgFlying,bothAvgFlying")
 

example/load/simulate/client/main.go

@@ -11,7 +11,7 @@ import (
 	"time"
 
 	"github.com/tal-tech/go-zero/core/fx"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 )
 
 var (
@@ -27,7 +27,7 @@ func main() {
 	flag.Parse()
 
 	fp, err := os.Create("result.csv")
-	lang.Must(err)
+	logx.Must(err)
 	defer fp.Close()
 	fmt.Fprintln(fp, "seconds,total,pass,fail,drop")
 

example/periodicalexecutor/pe.go

@@ -8,11 +8,11 @@ import (
 )
 
 func main() {
-	exeutor := executors.NewBulkExecutor(func(items []interface{}) {
+	executor := executors.NewBulkExecutor(func(items []interface{}) {
 		fmt.Println(len(items))
 	}, executors.WithBulkTasks(10))
 	for {
-		exeutor.Add(1)
+		executor.Add(1)
 		time.Sleep(time.Millisecond * 90)
 	}
 }

example/rpc/client/lb/main.go

@@ -2,7 +2,9 @@ package main
 
 import (
 	"context"
+	"flag"
 	"fmt"
+	"log"
 	"time"
 
 	"github.com/tal-tech/go-zero/core/discov"
@@ -10,13 +12,31 @@ import (
 	"github.com/tal-tech/go-zero/rpcx"
 )
 
+var lb = flag.String("t", "direct", "the load balancer type")
+
 func main() {
-	cli := rpcx.MustNewClient(rpcx.RpcClientConf{
-		Etcd: discov.EtcdConf{
-			Hosts: []string{"localhost:2379"},
-			Key:   "rpcx",
-		},
-	})
+	flag.Parse()
+
+	var cli rpcx.Client
+	switch *lb {
+	case "direct":
+		cli = rpcx.MustNewClient(rpcx.RpcClientConf{
+			Endpoints: []string{
+				"localhost:3456",
+				"localhost:3457",
+			},
+		})
+	case "discov":
+		cli = rpcx.MustNewClient(rpcx.RpcClientConf{
+			Etcd: discov.EtcdConf{
+				Hosts: []string{"localhost:2379"},
+				Key:   "rpcx",
+			},
+		})
+	default:
+		log.Fatal("bad load balancing type")
+	}
+
 	greet := unary.NewGreeterClient(cli.Conn())
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()

example/stringx/filter/filter.go

@@ -0,0 +1,21 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/tal-tech/go-zero/core/stringx"
+)
+
+func main() {
+	filter := stringx.NewTrie([]string{
+		"AV演员",
+		"苍井空",
+		"AV",
+		"日本AV女优",
+		"AV演员色情",
+	}, stringx.WithMask('?'))
+	safe, keywords, found := filter.Filter("日本AV演员兼电视、电影演员。苍井空AV女优是xx出道, 日本AV女优们最精彩的表演是AV演员色情表演")
+	fmt.Println(safe)
+	fmt.Println(keywords)
+	fmt.Println(found)
+}

example/stringx/replace/replace.go

@@ -0,0 +1,15 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/tal-tech/go-zero/core/stringx"
+)
+
+func main() {
+	replacer := stringx.NewReplacer(map[string]string{
+		"PHP": "PPT",
+		"世界上": "吹牛",
+	})
+	fmt.Println(replacer.Replace("PHP是世界上最好的语言！"))
+}

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/globalsign/mgo v0.0.0-20181015135952-eeefdecb41b8
 	github.com/go-redis/redis v6.15.7+incompatible
 	github.com/go-sql-driver/mysql v1.5.0
+	github.com/go-xorm/builder v0.3.4
 	github.com/gogo/protobuf v1.3.1 // indirect
 	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
 	github.com/golang/mock v1.4.3
@@ -22,6 +23,7 @@ require (
 	github.com/google/uuid v1.1.1
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.14.3 // indirect
+	github.com/iancoleman/strcase v0.0.0-20191112232945-16388991a334
 	github.com/justinas/alice v1.2.0
 	github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect
 	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
@@ -41,6 +43,7 @@ require (
 	github.com/stretchr/testify v1.5.1
 	github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6 // indirect
 	github.com/urfave/cli v1.22.4
+	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
 	github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb // indirect
 	go.etcd.io/etcd v0.0.0-20200402134248-51bdeb39e698
 	go.uber.org/automaxprocs v1.3.0

go.sum

@@ -76,6 +76,10 @@ github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gG
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/go-xorm/builder v0.3.4 h1:FxkeGB4Cggdw3tPwutLCpfjng2jugfkg6LDMrd/KsoY=
+github.com/go-xorm/builder v0.3.4/go.mod h1:KxkQkNN1DpPKTedxXyTQcmH+rXfvk4LZ9SOOBoZBAxw=
+github.com/go-xorm/sqlfiddle v0.0.0-20180821085327-62ce714f951a h1:9wScpmSP5A3Bk8V3XHWUcJmYTh+ZnlHVyc+A4oZYS3Y=
+github.com/go-xorm/sqlfiddle v0.0.0-20180821085327-62ce714f951a/go.mod h1:56xuuqnHyryaerycW3BfssRdxQstACi0Epw/yC5E2xM=
 github.com/godbus/dbus/v5 v5.0.3/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
@@ -134,6 +138,8 @@ github.com/grpc-ecosystem/grpc-gateway v1.14.3 h1:OCJlWkOUoTnl0neNGlf4fUm3TmbEtg
 github.com/grpc-ecosystem/grpc-gateway v1.14.3/go.mod h1:6CwZWGDSPRJidgKAtJVvND6soZe6fT7iteq8wDPdhb0=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/iancoleman/strcase v0.0.0-20191112232945-16388991a334 h1:VHgatEHNcBFEB7inlalqfNqw65aNkM1lGX2yt3NmbS8=
+github.com/iancoleman/strcase v0.0.0-20191112232945-16388991a334/go.mod h1:SK73tn/9oHe+/Y0h39VT4UCxmurVJkR5NA7kMEAOgSE=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jonboulle/clockwork v0.1.0 h1:VKV+ZcuP6l3yW9doeqz6ziZGgcynBVQO+obU0+0hcPo=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
@@ -266,6 +272,8 @@ github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6 h1:YdYsPAZ2pC6Tow/nPZOPQ96O3hm/ToAkGsPLzedXERk=
 github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg=
+github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2 h1:zzrxE1FKn5ryBNl9eKOeqQ58Y/Qpo3Q9QNxKHX5uzzQ=
+github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2/go.mod h1:hzfGeIUDq/j97IG+FhNqkowIyEcD88LrW6fyU3K3WqY=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb h1:ZkM6LRnq40pR1Ox0hTHlnpkcOTuFIDQpZ1IN8rKKhX0=
 github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=

readme.md

@@ -1,6 +1,10 @@
-# go-zero项目介绍
+# go-zero
 
-![Go](https://github.com/tal-tech/go-zero/workflows/Go/badge.svg?branch=master)
+[![Go](https://github.com/tal-tech/go-zero/workflows/Go/badge.svg?branch=master)](https://github.com/tal-tech/go-zero/actions)
+[![codecov](https://codecov.io/gh/tal-tech/go-zero/branch/master/graph/badge.svg)](https://codecov.io/gh/tal-tech/go-zero)
+[![Go Report Card](https://goreportcard.com/badge/github.com/tal-tech/go-zero)](https://goreportcard.com/report/github.com/tal-tech/go-zero)
+[![Release](https://img.shields.io/github/v/release/tal-tech/go-zero.svg?style=flat-square)](https://github.com/tal-tech/go-zero)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
 
 ## 1. go-zero框架背景
 
@@ -123,7 +127,6 @@ go-zero是一个集成了各种工程实践的包含web和rpc框架，有如下
    生成的文件结构如下：
 
    ```
-   .
    ├── greet
    │   ├── etc
    │   │   └── greet-api.json        // 配置文件
@@ -141,30 +144,27 @@ go-zero是一个集成了各种工程实践的包含web和rpc框架，有如下
    │       └── types
    │           └── types.go          // 请求、返回等类型定义
    └── greet.api                     // api描述文件
-   
-   8 directories, 9 files
    ```
-
    生成的代码可以直接运行：
-
-   ```shell
+   
+```shell
    cd greet
    go run greet.go -f etc/greet-api.json
-   ```
+```
 
-   默认侦听在8888端口（可以在配置文件里修改），可以通过curl请求：
+默认侦听在8888端口（可以在配置文件里修改），可以通过curl请求：
 
-   ```shell
+```shell
    ➜  go-zero git:(master) curl -w "\ncode: %{http_code}\n" http://localhost:8888/greet/from/kevin
    {"code":0}
    code: 200
-   ```
+```
 
-   编写业务代码：
+编写业务代码：
 
-   * 可以在servicecontext.go里面传递依赖给logic，比如mysql, redis等
+* 可以在servicecontext.go里面传递依赖给logic，比如mysql, redis等
    * 在api定义的get/post/put/delete等请求对应的logic里增加业务处理逻辑
-
+   
 4. 可以根据api文件生成前端需要的Java, TypeScript, Dart, JavaScript代码
 
    ```shell
@@ -173,6 +173,17 @@ go-zero是一个集成了各种工程实践的包含web和rpc框架，有如下
    ...
    ```
 
-### 微信交流群
+## 8. Benchmark
+
+![benchmark](doc/images/benchmark.png)
+
+[测试代码见这里](https://github.com/smallnest/go-web-framework-benchmark)
+
+## 9. 文档
+
+* [goctl使用帮助](doc/goctl.md)
+* [关键字替换和敏感词过滤工具](doc/keywords.md)
+
+## 10. 微信交流群
 
 添加我的微信：kevwan，请注明go-zero，我拉进go-zero社区群🤝

rest/engine.go

@@ -0,0 +1,214 @@
+package rest
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/justinas/alice"
+	"github.com/tal-tech/go-zero/core/codec"
+	"github.com/tal-tech/go-zero/core/load"
+	"github.com/tal-tech/go-zero/core/stat"
+	"github.com/tal-tech/go-zero/rest/handler"
+	"github.com/tal-tech/go-zero/rest/httpx"
+	"github.com/tal-tech/go-zero/rest/internal"
+	"github.com/tal-tech/go-zero/rest/router"
+)
+
+// use 1000m to represent 100%
+const topCpuUsage = 1000
+
+var ErrSignatureConfig = errors.New("bad config for Signature")
+
+type engine struct {
+	conf                 RestConf
+	routes               []featuredRoutes
+	unauthorizedCallback handler.UnauthorizedCallback
+	unsignedCallback     handler.UnsignedCallback
+	middlewares          []Middleware
+	shedder              load.Shedder
+	priorityShedder      load.Shedder
+}
+
+func newEngine(c RestConf) *engine {
+	srv := &engine{
+		conf: c,
+	}
+	if c.CpuThreshold > 0 {
+		srv.shedder = load.NewAdaptiveShedder(load.WithCpuThreshold(c.CpuThreshold))
+		srv.priorityShedder = load.NewAdaptiveShedder(load.WithCpuThreshold(
+			(c.CpuThreshold + topCpuUsage) >> 1))
+	}
+
+	return srv
+}
+
+func (s *engine) AddRoutes(r featuredRoutes) {
+	s.routes = append(s.routes, r)
+}
+
+func (s *engine) SetUnauthorizedCallback(callback handler.UnauthorizedCallback) {
+	s.unauthorizedCallback = callback
+}
+
+func (s *engine) SetUnsignedCallback(callback handler.UnsignedCallback) {
+	s.unsignedCallback = callback
+}
+
+func (s *engine) Start() error {
+	return s.StartWithRouter(router.NewPatRouter())
+}
+
+func (s *engine) StartWithRouter(router httpx.Router) error {
+	if err := s.bindRoutes(router); err != nil {
+		return err
+	}
+
+	return internal.StartHttp(s.conf.Host, s.conf.Port, router)
+}
+
+func (s *engine) appendAuthHandler(fr featuredRoutes, chain alice.Chain,
+	verifier func(alice.Chain) alice.Chain) alice.Chain {
+	if fr.jwt.enabled {
+		if len(fr.jwt.prevSecret) == 0 {
+			chain = chain.Append(handler.Authorize(fr.jwt.secret,
+				handler.WithUnauthorizedCallback(s.unauthorizedCallback)))
+		} else {
+			chain = chain.Append(handler.Authorize(fr.jwt.secret,
+				handler.WithPrevSecret(fr.jwt.prevSecret),
+				handler.WithUnauthorizedCallback(s.unauthorizedCallback)))
+		}
+	}
+
+	return verifier(chain)
+}
+
+func (s *engine) bindFeaturedRoutes(router httpx.Router, fr featuredRoutes, metrics *stat.Metrics) error {
+	verifier, err := s.signatureVerifier(fr.signature)
+	if err != nil {
+		return err
+	}
+
+	for _, route := range fr.routes {
+		if err := s.bindRoute(fr, router, metrics, route, verifier); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *engine) bindRoute(fr featuredRoutes, router httpx.Router, metrics *stat.Metrics,
+	route Route, verifier func(chain alice.Chain) alice.Chain) error {
+	chain := alice.New(
+		handler.TracingHandler,
+		s.getLogHandler(),
+		handler.MaxConns(s.conf.MaxConns),
+		handler.BreakerHandler(route.Method, route.Path, metrics),
+		handler.SheddingHandler(s.getShedder(fr.priority), metrics),
+		handler.TimeoutHandler(time.Duration(s.conf.Timeout)*time.Millisecond),
+		handler.RecoverHandler,
+		handler.MetricHandler(metrics),
+		handler.PromMetricHandler(route.Path),
+		handler.MaxBytesHandler(s.conf.MaxBytes),
+		handler.GunzipHandler,
+	)
+	chain = s.appendAuthHandler(fr, chain, verifier)
+
+	for _, middleware := range s.middlewares {
+		chain = chain.Append(convertMiddleware(middleware))
+	}
+	handle := chain.ThenFunc(route.Handler)
+
+	return router.Handle(route.Method, route.Path, handle)
+}
+
+func (s *engine) bindRoutes(router httpx.Router) error {
+	metrics := s.createMetrics()
+
+	for _, fr := range s.routes {
+		if err := s.bindFeaturedRoutes(router, fr, metrics); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *engine) createMetrics() *stat.Metrics {
+	var metrics *stat.Metrics
+
+	if len(s.conf.Name) > 0 {
+		metrics = stat.NewMetrics(s.conf.Name)
+	} else {
+		metrics = stat.NewMetrics(fmt.Sprintf("%s:%d", s.conf.Host, s.conf.Port))
+	}
+
+	return metrics
+}
+
+func (s *engine) getLogHandler() func(http.Handler) http.Handler {
+	if s.conf.Verbose {
+		return handler.DetailedLogHandler
+	} else {
+		return handler.LogHandler
+	}
+}
+
+func (s *engine) getShedder(priority bool) load.Shedder {
+	if priority && s.priorityShedder != nil {
+		return s.priorityShedder
+	}
+	return s.shedder
+}
+
+func (s *engine) signatureVerifier(signature signatureSetting) (func(chain alice.Chain) alice.Chain, error) {
+	if !signature.enabled {
+		return func(chain alice.Chain) alice.Chain {
+			return chain
+		}, nil
+	}
+
+	if len(signature.PrivateKeys) == 0 {
+		if signature.Strict {
+			return nil, ErrSignatureConfig
+		} else {
+			return func(chain alice.Chain) alice.Chain {
+				return chain
+			}, nil
+		}
+	}
+
+	decrypters := make(map[string]codec.RsaDecrypter)
+	for _, key := range signature.PrivateKeys {
+		fingerprint := key.Fingerprint
+		file := key.KeyFile
+		decrypter, err := codec.NewRsaDecrypter(file)
+		if err != nil {
+			return nil, err
+		}
+
+		decrypters[fingerprint] = decrypter
+	}
+
+	return func(chain alice.Chain) alice.Chain {
+		if s.unsignedCallback != nil {
+			return chain.Append(handler.ContentSecurityHandler(
+				decrypters, signature.Expiry, signature.Strict, s.unsignedCallback))
+		} else {
+			return chain.Append(handler.ContentSecurityHandler(
+				decrypters, signature.Expiry, signature.Strict))
+		}
+	}, nil
+}
+
+func (s *engine) use(middleware Middleware) {
+	s.middlewares = append(s.middlewares, middleware)
+}
+
+func convertMiddleware(ware Middleware) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(ware(next.ServeHTTP))
+	}
+}

rest/handler/authhandler.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/dgrijalva/jwt-go"
 	"github.com/tal-tech/go-zero/core/logx"
-	"github.com/tal-tech/go-zero/rest/internal"
+	"github.com/tal-tech/go-zero/rest/token"
 )
 
 const (
@@ -43,7 +43,7 @@ func Authorize(secret string, opts ...AuthorizeOption) func(http.Handler) http.H
 		opt(&authOpts)
 	}
 
-	parser := internal.NewTokenParser()
+	parser := token.NewTokenParser()
 	return func(next http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			token, err := parser.ParseToken(r, secret, authOpts.PrevSecret)

rest/handler/breakerhandler.go

@@ -8,7 +8,7 @@ import (
 	"github.com/tal-tech/go-zero/core/breaker"
 	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/stat"
-	"github.com/tal-tech/go-zero/rest/internal"
+	"github.com/tal-tech/go-zero/rest/httpx"
 	"github.com/tal-tech/go-zero/rest/internal/security"
 )
 
@@ -22,7 +22,7 @@ func BreakerHandler(method, path string, metrics *stat.Metrics) func(http.Handle
 			if err != nil {
 				metrics.AddDrop()
 				logx.Errorf("[http] dropped, %s - %s - %s",
-					r.RequestURI, internal.GetRemoteAddr(r), r.UserAgent())
+					r.RequestURI, httpx.GetRemoteAddr(r), r.UserAgent())
 				w.WriteHeader(http.StatusServiceUnavailable)
 				return
 			}

rest/handler/contentsecurityhandler_test.go

@@ -217,6 +217,7 @@ func TestContentSecurityHandler(t *testing.T) {
 					signature:   test.signature,
 				}
 				req, err := buildRequest(setting)
+				assert.Nil(t, err)
 				resp := httptest.NewRecorder()
 				handler.ServeHTTP(resp, req)
 				assert.Equal(t, test.statusCode, resp.Code)
@@ -249,6 +250,7 @@ func TestContentSecurityHandler_UnsignedCallback(t *testing.T) {
 		signature: "badone",
 	}
 	req, err := buildRequest(setting)
+	assert.Nil(t, err)
 	resp := httptest.NewRecorder()
 	handler.ServeHTTP(resp, req)
 	assert.Equal(t, http.StatusOK, resp.Code)
@@ -285,6 +287,7 @@ func TestContentSecurityHandler_UnsignedCallback_WrongTime(t *testing.T) {
 		fingerprint: fingerprint,
 	}
 	req, err := buildRequest(setting)
+	assert.Nil(t, err)
 	resp := httptest.NewRecorder()
 	handler.ServeHTTP(resp, req)
 	assert.Equal(t, http.StatusOK, resp.Code)

rest/handler/loghandler.go

@@ -13,6 +13,7 @@ import (
 	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/timex"
 	"github.com/tal-tech/go-zero/core/utils"
+	"github.com/tal-tech/go-zero/rest/httpx"
 	"github.com/tal-tech/go-zero/rest/internal"
 )
 
@@ -112,10 +113,10 @@ func logBrief(r *http.Request, code int, timer *utils.ElapsedTimer, logs *intern
 	var buf bytes.Buffer
 	duration := timer.Duration()
 	buf.WriteString(fmt.Sprintf("%d - %s - %s - %s - %s",
-		code, r.RequestURI, internal.GetRemoteAddr(r), r.UserAgent(), timex.ReprOfDuration(duration)))
+		code, r.RequestURI, httpx.GetRemoteAddr(r), r.UserAgent(), timex.ReprOfDuration(duration)))
 	if duration > slowThreshold {
 		logx.Slowf("[HTTP] %d - %s - %s - %s - slowcall(%s)",
-			code, r.RequestURI, internal.GetRemoteAddr(r), r.UserAgent(), timex.ReprOfDuration(duration))
+			code, r.RequestURI, httpx.GetRemoteAddr(r), r.UserAgent(), timex.ReprOfDuration(duration))
 	}
 
 	ok := isOkResponse(code)

rest/handler/sheddinghandler.go

@@ -7,7 +7,7 @@ import (
 	"github.com/tal-tech/go-zero/core/load"
 	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/stat"
-	"github.com/tal-tech/go-zero/rest/internal"
+	"github.com/tal-tech/go-zero/rest/httpx"
 	"github.com/tal-tech/go-zero/rest/internal/security"
 )
 
@@ -35,7 +35,7 @@ func SheddingHandler(shedder load.Shedder, metrics *stat.Metrics) func(http.Hand
 				metrics.AddDrop()
 				sheddingStat.IncrementDrop()
 				logx.Errorf("[http] dropped, %s - %s - %s",
-					r.RequestURI, internal.GetRemoteAddr(r), r.UserAgent())
+					r.RequestURI, httpx.GetRemoteAddr(r), r.UserAgent())
 				w.WriteHeader(http.StatusServiceUnavailable)
 				return
 			}

rest/httpx/requests.go

@@ -84,7 +84,6 @@ func ParseHeader(headerValue string) map[string]string {
 // Parses the post request which contains json in body.
 func ParseJsonBody(r *http.Request, v interface{}) error {
 	var reader io.Reader
-
 	if withJsonBody(r) {
 		reader = io.LimitReader(r.Body, maxBodyLen)
 	} else {

rest/httpx/util.go

@@ -1,4 +1,4 @@
-package internal
+package httpx
 
 import "net/http"
 

rest/httpx/util_test.go

@@ -1,4 +1,4 @@
-package internal
+package httpx
 
 import (
 	"net/http"
@@ -16,4 +16,3 @@ func TestGetRemoteAddr(t *testing.T) {
 	r.Header.Set(xForwardFor, host)
 	assert.Equal(t, host, GetRemoteAddr(r))
 }
-

rest/internal/log.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"github.com/tal-tech/go-zero/core/logx"
+	"github.com/tal-tech/go-zero/rest/httpx"
 )
 
 const LogContext = "request_logs"
@@ -79,5 +80,5 @@ func formatf(r *http.Request, format string, v ...interface{}) string {
 }
 
 func formatWithReq(r *http.Request, v string) string {
-	return fmt.Sprintf("(%s - %s) %s", r.RequestURI, GetRemoteAddr(r), v)
+	return fmt.Sprintf("(%s - %s) %s", r.RequestURI, httpx.GetRemoteAddr(r), v)
 }

rest/ngin.go

@@ -1,170 +0,0 @@
-package rest
-
-import (
-	"log"
-	"net/http"
-
-	"github.com/tal-tech/go-zero/core/logx"
-	"github.com/tal-tech/go-zero/rest/handler"
-	"github.com/tal-tech/go-zero/rest/httpx"
-)
-
-type (
-	runOptions struct {
-		start func(*engine) error
-	}
-
-	RunOption func(*Server)
-
-	Server struct {
-		ngin *engine
-		opts runOptions
-	}
-)
-
-func MustNewServer(c RestConf, opts ...RunOption) *Server {
-	engine, err := NewServer(c, opts...)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return engine
-}
-
-func NewServer(c RestConf, opts ...RunOption) (*Server, error) {
-	if err := c.SetUp(); err != nil {
-		return nil, err
-	}
-
-	server := &Server{
-		ngin: newEngine(c),
-		opts: runOptions{
-			start: func(srv *engine) error {
-				return srv.Start()
-			},
-		},
-	}
-
-	for _, opt := range opts {
-		opt(server)
-	}
-
-	return server, nil
-}
-
-func (e *Server) AddRoutes(rs []Route, opts ...RouteOption) {
-	r := featuredRoutes{
-		routes: rs,
-	}
-	for _, opt := range opts {
-		opt(&r)
-	}
-	e.ngin.AddRoutes(r)
-}
-
-func (e *Server) AddRoute(r Route, opts ...RouteOption) {
-	e.AddRoutes([]Route{r}, opts...)
-}
-
-func (e *Server) Start() {
-	handleError(e.opts.start(e.ngin))
-}
-
-func (e *Server) Stop() {
-	logx.Close()
-}
-
-func (e *Server) Use(middleware Middleware) {
-	e.ngin.use(middleware)
-}
-
-func ToMiddleware(handler func(next http.Handler) http.Handler) Middleware {
-	return func(handle http.HandlerFunc) http.HandlerFunc {
-		return handler(handle).ServeHTTP
-	}
-}
-
-func WithJwt(secret string) RouteOption {
-	return func(r *featuredRoutes) {
-		validateSecret(secret)
-		r.jwt.enabled = true
-		r.jwt.secret = secret
-	}
-}
-
-func WithJwtTransition(secret, prevSecret string) RouteOption {
-	return func(r *featuredRoutes) {
-		// why not validate prevSecret, because prevSecret is an already used one,
-		// even it not meet our requirement, we still need to allow the transition.
-		validateSecret(secret)
-		r.jwt.enabled = true
-		r.jwt.secret = secret
-		r.jwt.prevSecret = prevSecret
-	}
-}
-
-func WithMiddleware(middleware Middleware, rs ...Route) []Route {
-	routes := make([]Route, len(rs))
-
-	for i := range rs {
-		route := rs[i]
-		routes[i] = Route{
-			Method:  route.Method,
-			Path:    route.Path,
-			Handler: middleware(route.Handler),
-		}
-	}
-
-	return routes
-}
-
-func WithPriority() RouteOption {
-	return func(r *featuredRoutes) {
-		r.priority = true
-	}
-}
-
-func WithRouter(router httpx.Router) RunOption {
-	return func(server *Server) {
-		server.opts.start = func(srv *engine) error {
-			return srv.StartWithRouter(router)
-		}
-	}
-}
-
-func WithSignature(signature SignatureConf) RouteOption {
-	return func(r *featuredRoutes) {
-		r.signature.enabled = true
-		r.signature.Strict = signature.Strict
-		r.signature.Expiry = signature.Expiry
-		r.signature.PrivateKeys = signature.PrivateKeys
-	}
-}
-
-func WithUnauthorizedCallback(callback handler.UnauthorizedCallback) RunOption {
-	return func(engine *Server) {
-		engine.ngin.SetUnauthorizedCallback(callback)
-	}
-}
-
-func WithUnsignedCallback(callback handler.UnsignedCallback) RunOption {
-	return func(engine *Server) {
-		engine.ngin.SetUnsignedCallback(callback)
-	}
-}
-
-func handleError(err error) {
-	// ErrServerClosed means the server is closed manually
-	if err == nil || err == http.ErrServerClosed {
-		return
-	}
-
-	logx.Error(err)
-	panic(err)
-}
-
-func validateSecret(secret string) {
-	if len(secret) < 8 {
-		panic("secret's length can't be less than 8")
-	}
-}

rest/server.go

@@ -1,214 +1,170 @@
 package rest
 
 import (
-	"errors"
-	"fmt"
+	"log"
 	"net/http"
-	"time"
 
-	"github.com/justinas/alice"
-	"github.com/tal-tech/go-zero/core/codec"
-	"github.com/tal-tech/go-zero/core/load"
-	"github.com/tal-tech/go-zero/core/stat"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/rest/handler"
 	"github.com/tal-tech/go-zero/rest/httpx"
-	"github.com/tal-tech/go-zero/rest/internal"
-	"github.com/tal-tech/go-zero/rest/internal/router"
 )
 
-// use 1000m to represent 100%
-const topCpuUsage = 1000
-
-var ErrSignatureConfig = errors.New("bad config for Signature")
-
-type engine struct {
-	conf                 RestConf
-	routes               []featuredRoutes
-	unauthorizedCallback handler.UnauthorizedCallback
-	unsignedCallback     handler.UnsignedCallback
-	middlewares          []Middleware
-	shedder              load.Shedder
-	priorityShedder      load.Shedder
-}
-
-func newEngine(c RestConf) *engine {
-	srv := &engine{
-		conf: c,
-	}
-	if c.CpuThreshold > 0 {
-		srv.shedder = load.NewAdaptiveShedder(load.WithCpuThreshold(c.CpuThreshold))
-		srv.priorityShedder = load.NewAdaptiveShedder(load.WithCpuThreshold(
-			(c.CpuThreshold + topCpuUsage) >> 1))
+type (
+	runOptions struct {
+		start func(*engine) error
 	}
 
-	return srv
-}
+	RunOption func(*Server)
 
-func (s *engine) AddRoutes(r featuredRoutes) {
-	s.routes = append(s.routes, r)
-}
-
-func (s *engine) SetUnauthorizedCallback(callback handler.UnauthorizedCallback) {
-	s.unauthorizedCallback = callback
-}
-
-func (s *engine) SetUnsignedCallback(callback handler.UnsignedCallback) {
-	s.unsignedCallback = callback
-}
-
-func (s *engine) Start() error {
-	return s.StartWithRouter(router.NewPatRouter())
-}
-
-func (s *engine) StartWithRouter(router httpx.Router) error {
-	if err := s.bindRoutes(router); err != nil {
-		return err
+	Server struct {
+		ngin *engine
+		opts runOptions
 	}
+)
 
-	return internal.StartHttp(s.conf.Host, s.conf.Port, router)
-}
-
-func (s *engine) appendAuthHandler(fr featuredRoutes, chain alice.Chain,
-	verifier func(alice.Chain) alice.Chain) alice.Chain {
-	if fr.jwt.enabled {
-		if len(fr.jwt.prevSecret) == 0 {
-			chain = chain.Append(handler.Authorize(fr.jwt.secret,
-				handler.WithUnauthorizedCallback(s.unauthorizedCallback)))
-		} else {
-			chain = chain.Append(handler.Authorize(fr.jwt.secret,
-				handler.WithPrevSecret(fr.jwt.prevSecret),
-				handler.WithUnauthorizedCallback(s.unauthorizedCallback)))
-		}
-	}
-
-	return verifier(chain)
-}
-
-func (s *engine) bindFeaturedRoutes(router httpx.Router, fr featuredRoutes, metrics *stat.Metrics) error {
-	verifier, err := s.signatureVerifier(fr.signature)
+func MustNewServer(c RestConf, opts ...RunOption) *Server {
+	engine, err := NewServer(c, opts...)
 	if err != nil {
-		return err
+		log.Fatal(err)
 	}
 
-	for _, route := range fr.routes {
-		if err := s.bindRoute(fr, router, metrics, route, verifier); err != nil {
-			return err
+	return engine
+}
+
+func NewServer(c RestConf, opts ...RunOption) (*Server, error) {
+	if err := c.SetUp(); err != nil {
+		return nil, err
+	}
+
+	server := &Server{
+		ngin: newEngine(c),
+		opts: runOptions{
+			start: func(srv *engine) error {
+				return srv.Start()
+			},
+		},
+	}
+
+	for _, opt := range opts {
+		opt(server)
+	}
+
+	return server, nil
+}
+
+func (e *Server) AddRoutes(rs []Route, opts ...RouteOption) {
+	r := featuredRoutes{
+		routes: rs,
+	}
+	for _, opt := range opts {
+		opt(&r)
+	}
+	e.ngin.AddRoutes(r)
+}
+
+func (e *Server) AddRoute(r Route, opts ...RouteOption) {
+	e.AddRoutes([]Route{r}, opts...)
+}
+
+func (e *Server) Start() {
+	handleError(e.opts.start(e.ngin))
+}
+
+func (e *Server) Stop() {
+	logx.Close()
+}
+
+func (e *Server) Use(middleware Middleware) {
+	e.ngin.use(middleware)
+}
+
+func ToMiddleware(handler func(next http.Handler) http.Handler) Middleware {
+	return func(handle http.HandlerFunc) http.HandlerFunc {
+		return handler(handle).ServeHTTP
+	}
+}
+
+func WithJwt(secret string) RouteOption {
+	return func(r *featuredRoutes) {
+		validateSecret(secret)
+		r.jwt.enabled = true
+		r.jwt.secret = secret
+	}
+}
+
+func WithJwtTransition(secret, prevSecret string) RouteOption {
+	return func(r *featuredRoutes) {
+		// why not validate prevSecret, because prevSecret is an already used one,
+		// even it not meet our requirement, we still need to allow the transition.
+		validateSecret(secret)
+		r.jwt.enabled = true
+		r.jwt.secret = secret
+		r.jwt.prevSecret = prevSecret
+	}
+}
+
+func WithMiddleware(middleware Middleware, rs ...Route) []Route {
+	routes := make([]Route, len(rs))
+
+	for i := range rs {
+		route := rs[i]
+		routes[i] = Route{
+			Method:  route.Method,
+			Path:    route.Path,
+			Handler: middleware(route.Handler),
 		}
 	}
 
-	return nil
+	return routes
 }
 
-func (s *engine) bindRoute(fr featuredRoutes, router httpx.Router, metrics *stat.Metrics,
-	route Route, verifier func(chain alice.Chain) alice.Chain) error {
-	chain := alice.New(
-		handler.TracingHandler,
-		s.getLogHandler(),
-		handler.MaxConns(s.conf.MaxConns),
-		handler.BreakerHandler(route.Method, route.Path, metrics),
-		handler.SheddingHandler(s.getShedder(fr.priority), metrics),
-		handler.TimeoutHandler(time.Duration(s.conf.Timeout)*time.Millisecond),
-		handler.RecoverHandler,
-		handler.MetricHandler(metrics),
-		handler.PromMetricHandler(route.Path),
-		handler.MaxBytesHandler(s.conf.MaxBytes),
-		handler.GunzipHandler,
-	)
-	chain = s.appendAuthHandler(fr, chain, verifier)
-
-	for _, middleware := range s.middlewares {
-		chain = chain.Append(convertMiddleware(middleware))
+func WithPriority() RouteOption {
+	return func(r *featuredRoutes) {
+		r.priority = true
 	}
-	handle := chain.ThenFunc(route.Handler)
-
-	return router.Handle(route.Method, route.Path, handle)
 }
 
-func (s *engine) bindRoutes(router httpx.Router) error {
-	metrics := s.createMetrics()
-
-	for _, fr := range s.routes {
-		if err := s.bindFeaturedRoutes(router, fr, metrics); err != nil {
-			return err
+func WithRouter(router httpx.Router) RunOption {
+	return func(server *Server) {
+		server.opts.start = func(srv *engine) error {
+			return srv.StartWithRouter(router)
 		}
 	}
-
-	return nil
 }
 
-func (s *engine) createMetrics() *stat.Metrics {
-	var metrics *stat.Metrics
-
-	if len(s.conf.Name) > 0 {
-		metrics = stat.NewMetrics(s.conf.Name)
-	} else {
-		metrics = stat.NewMetrics(fmt.Sprintf("%s:%d", s.conf.Host, s.conf.Port))
-	}
-
-	return metrics
-}
-
-func (s *engine) getLogHandler() func(http.Handler) http.Handler {
-	if s.conf.Verbose {
-		return handler.DetailedLogHandler
-	} else {
-		return handler.LogHandler
+func WithSignature(signature SignatureConf) RouteOption {
+	return func(r *featuredRoutes) {
+		r.signature.enabled = true
+		r.signature.Strict = signature.Strict
+		r.signature.Expiry = signature.Expiry
+		r.signature.PrivateKeys = signature.PrivateKeys
 	}
 }
 
-func (s *engine) getShedder(priority bool) load.Shedder {
-	if priority && s.priorityShedder != nil {
-		return s.priorityShedder
-	}
-	return s.shedder
-}
-
-func (s *engine) signatureVerifier(signature signatureSetting) (func(chain alice.Chain) alice.Chain, error) {
-	if !signature.enabled {
-		return func(chain alice.Chain) alice.Chain {
-			return chain
-		}, nil
-	}
-
-	if len(signature.PrivateKeys) == 0 {
-		if signature.Strict {
-			return nil, ErrSignatureConfig
-		} else {
-			return func(chain alice.Chain) alice.Chain {
-				return chain
-			}, nil
-		}
-	}
-
-	decrypters := make(map[string]codec.RsaDecrypter)
-	for _, key := range signature.PrivateKeys {
-		fingerprint := key.Fingerprint
-		file := key.KeyFile
-		decrypter, err := codec.NewRsaDecrypter(file)
-		if err != nil {
-			return nil, err
-		}
-
-		decrypters[fingerprint] = decrypter
-	}
-
-	return func(chain alice.Chain) alice.Chain {
-		if s.unsignedCallback != nil {
-			return chain.Append(handler.ContentSecurityHandler(
-				decrypters, signature.Expiry, signature.Strict, s.unsignedCallback))
-		} else {
-			return chain.Append(handler.ContentSecurityHandler(
-				decrypters, signature.Expiry, signature.Strict))
-		}
-	}, nil
-}
-
-func (s *engine) use(middleware Middleware) {
-	s.middlewares = append(s.middlewares, middleware)
-}
-
-func convertMiddleware(ware Middleware) func(http.Handler) http.Handler {
-	return func(next http.Handler) http.Handler {
-		return http.HandlerFunc(ware(next.ServeHTTP))
+func WithUnauthorizedCallback(callback handler.UnauthorizedCallback) RunOption {
+	return func(engine *Server) {
+		engine.ngin.SetUnauthorizedCallback(callback)
+	}
+}
+
+func WithUnsignedCallback(callback handler.UnsignedCallback) RunOption {
+	return func(engine *Server) {
+		engine.ngin.SetUnsignedCallback(callback)
+	}
+}
+
+func handleError(err error) {
+	// ErrServerClosed means the server is closed manually
+	if err == nil || err == http.ErrServerClosed {
+		return
+	}
+
+	logx.Error(err)
+	panic(err)
+}
+
+func validateSecret(secret string) {
+	if len(secret) < 8 {
+		panic("secret's length can't be less than 8")
 	}
 }

rest/server_test.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/tal-tech/go-zero/rest/httpx"
-	"github.com/tal-tech/go-zero/rest/internal/router"
+	"github.com/tal-tech/go-zero/rest/router"
 )
 
 func TestWithMiddleware(t *testing.T) {

rest/token/tokenparser.go

@@ -1,4 +1,4 @@
-package internal
+package token
 
 import (
 	"net/http"

rest/token/tokenparser_test.go

@@ -1,4 +1,4 @@
-package internal
+package token
 
 import (
 	"net/http"

rpcx/client.go

@@ -49,10 +49,10 @@ func NewClient(c RpcClientConf, options ...internal.ClientOption) (Client, error
 
 	var client Client
 	var err error
-	if len(c.Server) > 0 {
-		client, err = internal.NewDirectClient(c.Server, opts...)
+	if len(c.Endpoints) > 0 {
+		client, err = internal.NewClient(internal.BuildDirectTarget(c.Endpoints), opts...)
 	} else if err = c.Etcd.Validate(); err == nil {
-		client, err = internal.NewDiscovClient(c.Etcd.Hosts, c.Etcd.Key, opts...)
+		client, err = internal.NewClient(internal.BuildDiscovTarget(c.Etcd.Hosts, c.Etcd.Key), opts...)
 	}
 	if err != nil {
 		return nil, err
@@ -64,7 +64,7 @@ func NewClient(c RpcClientConf, options ...internal.ClientOption) (Client, error
 }
 
 func NewClientNoAuth(c discov.EtcdConf) (Client, error) {
-	client, err := internal.NewDiscovClient(c.Hosts, c.Key)
+	client, err := internal.NewClient(internal.BuildDiscovTarget(c.Hosts, c.Key))
 	if err != nil {
 		return nil, err
 	}
@@ -74,6 +74,10 @@ func NewClientNoAuth(c discov.EtcdConf) (Client, error) {
 	}, nil
 }
 
+func NewClientWithTarget(target string, opts ...internal.ClientOption) (Client, error) {
+	return internal.NewClient(target, opts...)
+}
+
 func (rc *RpcClient) Conn() *grpc.ClientConn {
 	return rc.client.Conn()
 }

rpcx/config.go

@@ -21,19 +21,19 @@ type (
 	}
 
 	RpcClientConf struct {
-		Etcd    discov.EtcdConf `json:",optional"`
-		Server  string          `json:",optional=!Etcd"`
-		App     string          `json:",optional"`
-		Token   string          `json:",optional"`
-		Timeout int64           `json:",optional"`
+		Etcd      discov.EtcdConf `json:",optional"`
+		Endpoints []string        `json:",optional=!Etcd"`
+		App       string          `json:",optional"`
+		Token     string          `json:",optional"`
+		Timeout   int64           `json:",optional"`
 	}
 )
 
-func NewDirectClientConf(server, app, token string) RpcClientConf {
+func NewDirectClientConf(endpoints []string, app, token string) RpcClientConf {
 	return RpcClientConf{
-		Server: server,
-		App:    app,
-		Token:  token,
+		Endpoints: endpoints,
+		App:       app,
+		Token:     token,
 	}
 }
 

rpcx/internal/balancer/p2c/p2c.go

@@ -21,7 +21,7 @@ import (
 
 const (
 	Name            = "p2c_ewma"
-	decayTime       = int64(time.Millisecond * 600)
+	decayTime       = int64(time.Second * 10) // default value from finagle
 	forcePick       = int64(time.Second)
 	initSuccess     = 1000
 	throttleSuccess = initSuccess / 2

rpcx/internal/client.go

@@ -5,12 +5,18 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/tal-tech/go-zero/rpcx/internal/balancer/p2c"
 	"github.com/tal-tech/go-zero/rpcx/internal/clientinterceptors"
+	"github.com/tal-tech/go-zero/rpcx/internal/resolver"
 	"google.golang.org/grpc"
 )
 
 const dialTimeout = time.Second * 3
 
+func init() {
+	resolver.RegisterResolver()
+}
+
 type (
 	ClientOptions struct {
 		Timeout     time.Duration
@@ -18,8 +24,26 @@ type (
 	}
 
 	ClientOption func(options *ClientOptions)
+
+	client struct {
+		conn *grpc.ClientConn
+	}
 )
 
+func NewClient(target string, opts ...ClientOption) (*client, error) {
+	opts = append(opts, WithDialOption(grpc.WithBalancerName(p2c.Name)))
+	conn, err := dial(target, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &client{conn: conn}, nil
+}
+
+func (c *client) Conn() *grpc.ClientConn {
+	return c.conn
+}
+
 func WithDialOption(opt grpc.DialOption) ClientOption {
 	return func(options *ClientOptions) {
 		options.DialOptions = append(options.DialOptions, opt)

rpcx/internal/directclient.go

@@ -1,26 +0,0 @@
-package internal
-
-import (
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/balancer/roundrobin"
-)
-
-type DirectClient struct {
-	conn *grpc.ClientConn
-}
-
-func NewDirectClient(server string, opts ...ClientOption) (*DirectClient, error) {
-	opts = append(opts, WithDialOption(grpc.WithBalancerName(roundrobin.Name)))
-	conn, err := dial(server, opts...)
-	if err != nil {
-		return nil, err
-	}
-
-	return &DirectClient{
-		conn: conn,
-	}, nil
-}
-
-func (c *DirectClient) Conn() *grpc.ClientConn {
-	return c.conn
-}

rpcx/internal/discovclient.go

@@ -1,34 +0,0 @@
-package internal
-
-import (
-	"fmt"
-	"strings"
-
-	"github.com/tal-tech/go-zero/rpcx/internal/balancer/p2c"
-	"github.com/tal-tech/go-zero/rpcx/internal/resolver"
-	"google.golang.org/grpc"
-)
-
-func init() {
-	resolver.RegisterResolver()
-}
-
-type DiscovClient struct {
-	conn *grpc.ClientConn
-}
-
-func NewDiscovClient(endpoints []string, key string, opts ...ClientOption) (*DiscovClient, error) {
-	opts = append(opts, WithDialOption(grpc.WithBalancerName(p2c.Name)))
-	target := fmt.Sprintf("%s://%s/%s", resolver.DiscovScheme,
-		strings.Join(endpoints, resolver.EndpointSep), key)
-	conn, err := dial(target, opts...)
-	if err != nil {
-		return nil, err
-	}
-
-	return &DiscovClient{conn: conn}, nil
-}
-
-func (c *DiscovClient) Conn() *grpc.ClientConn {
-	return c.conn
-}

rpcx/internal/resolver/directbuilder.go

@@ -0,0 +1,30 @@
+package resolver
+
+import (
+	"strings"
+
+	"google.golang.org/grpc/resolver"
+)
+
+type directBuilder struct{}
+
+func (d *directBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (
+	resolver.Resolver, error) {
+	var addrs []resolver.Address
+	endpoints := strings.Split(target.Endpoint, EndpointSep)
+
+	for _, val := range subset(endpoints, subsetSize) {
+		addrs = append(addrs, resolver.Address{
+			Addr: val,
+		})
+	}
+	cc.UpdateState(resolver.State{
+		Addresses: addrs,
+	})
+
+	return &nopResolver{cc: cc}, nil
+}
+
+func (d *directBuilder) Scheme() string {
+	return DirectScheme
+}

rpcx/internal/resolver/discovbuilder.go

@@ -0,0 +1,39 @@
+package resolver
+
+import (
+	"strings"
+
+	"github.com/tal-tech/go-zero/core/discov"
+	"google.golang.org/grpc/resolver"
+)
+
+type discovBuilder struct{}
+
+func (d *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (
+	resolver.Resolver, error) {
+	hosts := strings.Split(target.Authority, EndpointSep)
+	sub, err := discov.NewSubscriber(hosts, target.Endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	update := func() {
+		var addrs []resolver.Address
+		for _, val := range subset(sub.Values(), subsetSize) {
+			addrs = append(addrs, resolver.Address{
+				Addr: val,
+			})
+		}
+		cc.UpdateState(resolver.State{
+			Addresses: addrs,
+		})
+	}
+	sub.AddListener(update)
+	update()
+
+	return &nopResolver{cc: cc}, nil
+}
+
+func (d *discovBuilder) Scheme() string {
+	return DiscovScheme
+}

rpcx/internal/resolver/resolver.go

@@ -1,68 +1,30 @@
 package resolver
 
-import (
-	"fmt"
-	"strings"
-
-	"github.com/tal-tech/go-zero/core/discov"
-	"google.golang.org/grpc/resolver"
-)
+import "google.golang.org/grpc/resolver"
 
 const (
+	DirectScheme = "direct"
 	DiscovScheme = "discov"
 	EndpointSep  = ","
 	subsetSize   = 32
 )
 
-var builder discovBuilder
+var (
+	dirBuilder directBuilder
+	disBuilder discovBuilder
+)
 
-type discovBuilder struct{}
-
-func (b *discovBuilder) Scheme() string {
-	return DiscovScheme
+func RegisterResolver() {
+	resolver.Register(&dirBuilder)
+	resolver.Register(&disBuilder)
 }
 
-func (b *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (
-	resolver.Resolver, error) {
-	if target.Scheme != DiscovScheme {
-		return nil, fmt.Errorf("bad scheme: %s", target.Scheme)
-	}
-
-	hosts := strings.Split(target.Authority, EndpointSep)
-	sub, err := discov.NewSubscriber(hosts, target.Endpoint)
-	if err != nil {
-		return nil, err
-	}
-
-	update := func() {
-		var addrs []resolver.Address
-		for _, val := range subset(sub.Values(), subsetSize) {
-			addrs = append(addrs, resolver.Address{
-				Addr: val,
-			})
-		}
-		cc.UpdateState(resolver.State{
-			Addresses: addrs,
-		})
-	}
-	sub.AddListener(update)
-	update()
-
-	return &discovResolver{
-		cc: cc,
-	}, nil
-}
-
-type discovResolver struct {
+type nopResolver struct {
 	cc resolver.ClientConn
 }
 
-func (r *discovResolver) Close() {
+func (r *nopResolver) Close() {
 }
 
-func (r *discovResolver) ResolveNow(options resolver.ResolveNowOptions) {
-}
-
-func RegisterResolver() {
-	resolver.Register(&builder)
+func (r *nopResolver) ResolveNow(options resolver.ResolveNowOptions) {
 }

rpcx/internal/target.go

@@ -0,0 +1,17 @@
+package internal
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/tal-tech/go-zero/rpcx/internal/resolver"
+)
+
+func BuildDirectTarget(endpoints []string) string {
+	return fmt.Sprintf("%s:///%s", resolver.DirectScheme, strings.Join(endpoints, resolver.EndpointSep))
+}
+
+func BuildDiscovTarget(endpoints []string, key string) string {
+	return fmt.Sprintf("%s://%s/%s", resolver.DiscovScheme,
+		strings.Join(endpoints, resolver.EndpointSep), key)
+}

rpcx/proxy.go

@@ -38,11 +38,11 @@ func (p *RpcProxy) TakeConn(ctx context.Context) (*grpc.ClientConn, error) {
 			return client, nil
 		}
 
-		client, err := NewClient(RpcClientConf{
-			Server: p.backend,
-			App:    cred.App,
-			Token:  cred.Token,
-		}, p.options...)
+		opts := append(p.options, WithDialOption(grpc.WithPerRPCCredentials(&auth.Credential{
+			App:   cred.App,
+			Token: cred.Token,
+		})))
+		client, err := NewClientWithTarget(p.backend, opts...)
 		if err != nil {
 			return nil, err
 		}

tools/goctl/api/dartgen/gen.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 	"strings"
 
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/tools/goctl/api/parser"
 	"github.com/urfave/cli"
 )
@@ -32,8 +32,8 @@ func DartCommand(c *cli.Context) error {
 		dir = dir + "/"
 	}
 	api.Info.Title = strings.Replace(apiFile, ".api", "", -1)
-	lang.Must(genData(dir+"data/", api))
-	lang.Must(genApi(dir+"api/", api))
-	lang.Must(genVars(dir + "vars/"))
+	logx.Must(genData(dir+"data/", api))
+	logx.Must(genApi(dir+"api/", api))
+	logx.Must(genVars(dir + "vars/"))
 	return nil
 }

tools/goctl/api/gogen/gen.go

@@ -1,6 +1,7 @@
 package gogen
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"os"
@@ -13,7 +14,7 @@ import (
 	"time"
 
 	"github.com/logrusorgru/aurora"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	apiformat "github.com/tal-tech/go-zero/tools/goctl/api/format"
 	"github.com/tal-tech/go-zero/tools/goctl/api/parser"
 	apiutil "github.com/tal-tech/go-zero/tools/goctl/api/util"
@@ -44,17 +45,18 @@ func GoCommand(c *cli.Context) error {
 		return err
 	}
 
-	lang.Must(util.MkdirIfNotExist(dir))
-	lang.Must(genEtc(dir, api))
-	lang.Must(genConfig(dir))
-	lang.Must(genMain(dir, api))
-	lang.Must(genServiceContext(dir, api))
-	lang.Must(genTypes(dir, api))
-	lang.Must(genHandlers(dir, api))
-	lang.Must(genRoutes(dir, api))
-	lang.Must(genLogic(dir, api))
+	logx.Must(util.MkdirIfNotExist(dir))
+	logx.Must(genEtc(dir, api))
+	logx.Must(genConfig(dir))
+	logx.Must(genMain(dir, api))
+	logx.Must(genServiceContext(dir, api))
+	logx.Must(genTypes(dir, api))
+	logx.Must(genHandlers(dir, api))
+	logx.Must(genRoutes(dir, api))
+	logx.Must(genLogic(dir, api))
 	// it does not work
 	format(dir)
+	createGoModFileIfNeed(dir)
 
 	if err := backupAndSweep(apiFile); err != nil {
 		return err
@@ -98,7 +100,7 @@ func format(dir string) {
 	cmd := exec.Command("go", "fmt", "./"+dir+"...")
 	_, err := cmd.CombinedOutput()
 	if err != nil {
-		print(err.Error())
+		fmt.Println(err.Error())
 	}
 }
 
@@ -131,3 +133,42 @@ func sweep() error {
 		return nil
 	})
 }
+
+func createGoModFileIfNeed(dir string) {
+	absDir, err := filepath.Abs(dir)
+	if err != nil {
+		panic(err)
+	}
+
+	var tempPath = absDir
+	var hasGoMod = false
+	for {
+		if tempPath == filepath.Dir(tempPath) {
+			break
+		}
+		tempPath = filepath.Dir(tempPath)
+		if util.FileExists(filepath.Join(tempPath, goModeIdentifier)) {
+			hasGoMod = true
+			break
+		}
+	}
+	if !hasGoMod {
+		gopath := os.Getenv("GOPATH")
+		parent := path.Join(gopath, "src")
+		pos := strings.Index(absDir, parent)
+		if pos < 0 {
+			moduleName := absDir[len(filepath.Dir(absDir))+1:]
+			cmd := exec.Command("go", "mod", "init", moduleName)
+			cmd.Dir = dir
+			var stdout, stderr bytes.Buffer
+			cmd.Stdout = &stdout
+			cmd.Stderr = &stderr
+			err := cmd.Run()
+			if err != nil {
+				fmt.Println(err.Error())
+			}
+			outStr, errStr := string(stdout.Bytes()), string(stderr.Bytes())
+			fmt.Printf(outStr + "\n" + errStr)
+		}
+	}
+}

tools/goctl/api/gogen/genhandlers.go

@@ -72,7 +72,7 @@ func genHandler(dir string, group spec.Group, route spec.Route) error {
 		req = ""
 	}
 	var logicResponse string
-	var writeResponse = "nil, nil"
+	var writeResponse string
 	var respWriter = `httpx.WriteJson(w, http.StatusOK, resp)`
 	if len(route.ResponseType.Name) > 0 {
 		logicResponse = "resp, err :="

tools/goctl/api/gogen/genroutes.go

@@ -43,6 +43,7 @@ var mapping = map[string]string{
 	"head":   "http.MethodHead",
 	"post":   "http.MethodPost",
 	"put":    "http.MethodPut",
+	"patch":  "http.MethodPatch",
 }
 
 type (

tools/goctl/api/gogen/util.go

@@ -28,6 +28,9 @@ func getParentPackage(dir string) (string, error) {
 	var tempPath = absDir
 	var hasGoMod = false
 	for {
+		if tempPath == filepath.Dir(tempPath) {
+			break
+		}
 		tempPath = filepath.Dir(tempPath)
 		if goctlutil.FileExists(filepath.Join(tempPath, goModeIdentifier)) {
 			tempPath = filepath.Dir(tempPath)

tools/goctl/api/javagen/gen.go

@@ -6,7 +6,7 @@ import (
 	"strings"
 
 	"github.com/logrusorgru/aurora"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/tools/goctl/api/parser"
 	"github.com/tal-tech/go-zero/tools/goctl/util"
 	"github.com/urfave/cli"
@@ -36,9 +36,9 @@ func JavaCommand(c *cli.Context) error {
 		packetName = packetName[:len(packetName)-4]
 	}
 
-	lang.Must(util.MkdirIfNotExist(dir))
-	lang.Must(genPacket(dir, packetName, api))
-	lang.Must(genComponents(dir, packetName, api))
+	logx.Must(util.MkdirIfNotExist(dir))
+	logx.Must(genPacket(dir, packetName, api))
+	logx.Must(genComponents(dir, packetName, api))
 
 	fmt.Println(aurora.Green("Done."))
 	return nil

tools/goctl/api/ktgen/cmd.go

@@ -0,0 +1,42 @@
+package ktgen
+
+import (
+	"errors"
+
+	"github.com/tal-tech/go-zero/tools/goctl/api/parser"
+	"github.com/urfave/cli"
+)
+
+func KtCommand(c *cli.Context) error {
+	apiFile := c.String("api")
+	if apiFile == "" {
+		return errors.New("missing -api")
+	}
+	dir := c.String("dir")
+	if dir == "" {
+		return errors.New("missing -dir")
+	}
+	pkg := c.String("pkg")
+	if pkg == "" {
+		return errors.New("missing -pkg")
+	}
+
+	p, e := parser.NewParser(apiFile)
+	if e != nil {
+		return e
+	}
+	api, e := p.Parse()
+	if e != nil {
+		return e
+	}
+
+	e = genBase(dir, pkg, api)
+	if e != nil {
+		return e
+	}
+	e = genApi(dir, pkg, api)
+	if e != nil {
+		return e
+	}
+	return nil
+}

tools/goctl/api/ktgen/funcs.go

@@ -0,0 +1,77 @@
+package ktgen
+
+import (
+	"log"
+	"strings"
+	"text/template"
+
+	"github.com/iancoleman/strcase"
+	"github.com/tal-tech/go-zero/tools/goctl/api/util"
+)
+
+var funcsMap = template.FuncMap{
+	"lowCamelCase":    lowCamelCase,
+	"routeToFuncName": routeToFuncName,
+	"parseType":       parseType,
+	"add":             add,
+	"upperCase":       upperCase,
+}
+
+func lowCamelCase(s string) string {
+	if len(s) < 1 {
+		return ""
+	}
+	s = util.ToCamelCase(util.ToSnakeCase(s))
+	return util.ToLower(s[:1]) + s[1:]
+}
+
+func routeToFuncName(method, path string) string {
+	if !strings.HasPrefix(path, "/") {
+		path = "/" + path
+	}
+
+	path = strings.ReplaceAll(path, "/", "_")
+	path = strings.ReplaceAll(path, "-", "_")
+	path = strings.ReplaceAll(path, ":", "With_")
+
+	return strings.ToLower(method) + strcase.ToCamel(path)
+}
+
+func parseType(t string) string {
+	t = strings.Replace(t, "*", "", -1)
+	if strings.HasPrefix(t, "[]") {
+		return "List<" + parseType(t[2:]) + ">"
+	}
+
+	if strings.HasPrefix(t, "map") {
+		tys, e := util.DecomposeType(t)
+		if e != nil {
+			log.Fatal(e)
+		}
+		if len(tys) != 2 {
+			log.Fatal("Map type number !=2")
+		}
+		return "Map<String," + parseType(tys[1]) + ">"
+	}
+
+	switch t {
+	case "string":
+		return "String"
+	case "int", "int32", "int64":
+		return "Int"
+	case "float", "float32", "float64":
+		return "Double"
+	case "bool":
+		return "Boolean"
+	default:
+		return t
+	}
+}
+
+func add(a, i int) int {
+	return a + i
+}
+
+func upperCase(s string) string {
+	return strings.ToUpper(s)
+}

tools/goctl/api/ktgen/gen.go

@@ -0,0 +1,150 @@
+package ktgen
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"text/template"
+
+	"github.com/iancoleman/strcase"
+	"github.com/tal-tech/go-zero/tools/goctl/api/spec"
+)
+
+const (
+	apiBaseTemplate = `package {{.}}
+
+import com.google.gson.Gson
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+import java.io.BufferedReader
+import java.io.InputStreamReader
+import java.io.OutputStreamWriter
+import java.net.HttpURLConnection
+import java.net.URL
+
+const val SERVER = "http://localhost:8080"
+
+suspend fun apiRequest(
+    method: String,
+    uri: String,
+    body: Any = "",
+    onOk: ((String) -> Unit)? = null,
+    onFail: ((String) -> Unit)? = null,
+    eventually: (() -> Unit)? = null
+) = withContext(Dispatchers.IO) {
+    val url = URL(SERVER + uri)
+    with(url.openConnection() as HttpURLConnection) {
+        connectTimeout = 3000
+        requestMethod = method
+        doInput = true
+        if (method == "POST" || method == "PUT" || method == "PATCH") {
+            setRequestProperty("Content-Type", "application/json")
+            doOutput = true
+            val data = when (body) {
+                is String -> {
+                    body
+                }
+                else -> {
+                    Gson().toJson(body)
+                }
+            }
+            val wr = OutputStreamWriter(outputStream)
+            wr.write(data)
+            wr.flush()
+        }
+
+         try {
+            if (responseCode >= 400) {
+                BufferedReader(InputStreamReader(errorStream)).use {
+                    val response = it.readText()
+                    onFail?.invoke(response)
+                }
+                return@with
+            }
+            //response
+            BufferedReader(InputStreamReader(inputStream)).use {
+                val response = it.readText()
+                onOk?.invoke(response)
+            }
+        } catch (e: Exception) {
+            e.message?.let { onFail?.invoke(it) }
+        }
+    }
+    eventually?.invoke()
+}
+`
+	apiTemplate = `package {{with .Info}}{{.Desc}}{{end}}
+
+import com.google.gson.Gson
+
+object {{with .Info}}{{.Title}}{{end}}{
+	{{range .Types}}
+	data class {{.Name}}({{$length := (len .Members)}}{{range $i,$item := .Members}}
+		val {{with $item}}{{lowCamelCase .Name}}: {{parseType .Type}}{{end}}{{if ne $i (add $length -1)}},{{end}}{{end}}
+	){{end}}
+	{{with .Service}}
+	{{range .Routes}}suspend fun {{routeToFuncName .Method .Path}}({{with .RequestType}}{{if ne .Name ""}}
+		req:{{.Name}},{{end}}{{end}}
+		onOk: (({{with .ResponseType}}{{.Name}}{{end}}) -> Unit)? = null,
+        onFail: ((String) -> Unit)? = null,
+        eventually: (() -> Unit)? = null
+    ){
+        apiRequest("{{upperCase .Method}}","{{.Path}}",{{with .RequestType}}{{if ne .Name ""}}body=req,{{end}}{{end}} onOk = { {{with .ResponseType}}
+            onOk?.invoke({{if ne .Name ""}}Gson().fromJson(it,{{.Name}}::class.java){{end}}){{end}}
+        }, onFail = onFail, eventually =eventually)
+    }
+	{{end}}{{end}}
+}`
+)
+
+func genBase(dir, pkg string, api *spec.ApiSpec) error {
+	e := os.MkdirAll(dir, 0755)
+	if e != nil {
+		return e
+	}
+	path := filepath.Join(dir, "BaseApi.kt")
+	if _, e := os.Stat(path); e == nil {
+		fmt.Println("BaseApi.kt already exists, skipped it.")
+		return nil
+	}
+
+	file, e := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644)
+	if e != nil {
+		return e
+	}
+	defer file.Close()
+
+	t, e := template.New("n").Parse(apiBaseTemplate)
+	if e != nil {
+		return e
+	}
+	e = t.Execute(file, pkg)
+	if e != nil {
+		return e
+	}
+	return nil
+}
+
+func genApi(dir, pkg string, api *spec.ApiSpec) error {
+	name := strcase.ToCamel(api.Info.Title + "Api")
+	path := filepath.Join(dir, name+".kt")
+	api.Info.Title = name
+	api.Info.Desc = pkg
+
+	e := os.MkdirAll(dir, 0755)
+	if e != nil {
+		return e
+	}
+
+	file, e := os.OpenFile(path, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
+	if e != nil {
+		return e
+	}
+	defer file.Close()
+
+	t, e := template.New("api").Funcs(funcsMap).Parse(apiTemplate)
+	if e != nil {
+		return e
+	}
+	return t.Execute(file, api)
+}

tools/goctl/api/main.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/tools/goctl/api/parser"
 )
 
@@ -14,8 +14,8 @@ func main() {
 	}
 
 	p, err := parser.NewParser(os.Args[1])
-	lang.Must(err)
+	logx.Must(err)
 	api, err := p.Parse()
-	lang.Must(err)
+	logx.Must(err)
 	fmt.Println(api)
 }

tools/goctl/api/parser/rootstate.go

@@ -99,8 +99,6 @@ func (s rootState) processToken(token string, annos []spec.Annotation) (state, e
 	switch token {
 	case infoDirective:
 		return newInfoState(s.baseState), nil
-	//case typeDirective:
-	//return newTypeState(s.baseState, annos), nil
 	case serviceDirective:
 		return newServiceState(s.baseState, annos), nil
 	default:

tools/goctl/api/tsgen/gen.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 
 	"github.com/logrusorgru/aurora"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/tools/goctl/api/parser"
 	"github.com/tal-tech/go-zero/tools/goctl/util"
 	"github.com/urfave/cli"
@@ -34,9 +34,9 @@ func TsCommand(c *cli.Context) error {
 		return err
 	}
 
-	lang.Must(util.MkdirIfNotExist(dir))
-	lang.Must(genHandler(dir, webApi, caller, api, unwrapApi))
-	lang.Must(genComponents(dir, api))
+	logx.Must(util.MkdirIfNotExist(dir))
+	logx.Must(genHandler(dir, webApi, caller, api, unwrapApi))
+	logx.Must(genComponents(dir, api))
 
 	fmt.Println(aurora.Green("Done."))
 	return nil

tools/goctl/api/tsgen/util.go

@@ -13,6 +13,10 @@ import (
 func writeProperty(writer io.Writer, member spec.Member, indent int, prefixForType func(string) string) error {
 	writeIndent(writer, indent)
 	ty, err := goTypeToTs(member.Type, prefixForType)
+	if err != nil {
+		return err
+	}
+
 	optionalTag := ""
 	if member.IsOptional() || member.IsOmitempty() {
 		optionalTag = "?"
@@ -21,13 +25,14 @@ func writeProperty(writer io.Writer, member spec.Member, indent int, prefixForTy
 	if err != nil {
 		return err
 	}
+
 	comment := member.GetComment()
 	if len(comment) > 0 {
 		comment = strings.TrimPrefix(comment, "//")
 		comment = " // " + strings.TrimSpace(comment)
 	}
 	if len(member.Docs) > 0 {
-		_, err = fmt.Fprintf(writer, "%s\n", strings.Join(member.Docs, ""))
+		fmt.Fprintf(writer, "%s\n", strings.Join(member.Docs, ""))
 		writeIndent(writer, 1)
 	}
 	_, err = fmt.Fprintf(writer, "%s%s: %s%s\n", name, optionalTag, ty, comment)