Revert "🐞 fix(gen): pg gen of insert (#1591)" (#1598)

This reverts commit cc4c4928e0.

.codecov.yml

@@ -1,4 +1,3 @@
+comment: false
 ignore:
-  - "doc"
-  - "example"
-  - "tools"
+  - "tools"

.github/FUNDING.yml

@@ -0,0 +1,12 @@
+# These are supported funding model platforms
+
+github: # Replace with up to 4 GitHub Sponsors-enabled usernames e.g., [user1, user2]
+patreon: # Replace with a single Patreon username
+open_collective: # Replace with a single Open Collective username
+ko_fi: # Replace with a single Ko-fi username
+tidelift: # Replace with a single Tidelift platform-name/package-name e.g., npm/babel
+community_bridge: # Replace with a single Community Bridge project-name e.g., cloud-foundry
+liberapay: # Replace with a single Liberapay username
+issuehunt: # Replace with a single IssueHunt username
+otechie: # Replace with a single Otechie username
+custom: https://gitee.com/kevwan/static/raw/master/images/sponsor.jpg

.github/ISSUE_TEMPLATE/bug_report.md

@@ -0,0 +1,40 @@
+---
+name: Bug report
+about: Create a report to help us improve
+title: ''
+labels: ''
+assignees: ''
+
+---
+
+**Describe the bug**
+A clear and concise description of what the bug is.
+
+**To Reproduce**
+Steps to reproduce the behavior, if applicable:
+
+1. The code is
+
+   ```go
+   
+   ```
+
+2. The error is
+
+   ```
+   
+   ```
+
+**Expected behavior**
+A clear and concise description of what you expected to happen.
+
+**Screenshots**
+If applicable, add screenshots to help explain your problem.
+
+**Environments (please complete the following information):**
+ - OS: [e.g. Linux]
+ - go-zero version [e.g. 1.2.1]
+ - goctl version [e.g. 1.2.1, optional]
+
+**More description**
+Add any other context about the problem here.

.github/ISSUE_TEMPLATE/feature_request.md

@@ -0,0 +1,20 @@
+---
+name: Feature request
+about: Suggest an idea for this project
+title: ''
+labels: ''
+assignees: ''
+
+---
+
+**Is your feature request related to a problem? Please describe.**
+A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
+
+**Describe the solution you'd like**
+A clear and concise description of what you want to happen.
+
+**Describe alternatives you've considered**
+A clear and concise description of any alternative solutions or features you've considered.
+
+**Additional context**
+Add any other context or screenshots about the feature request here.

.github/ISSUE_TEMPLATE/question.md

@@ -0,0 +1,10 @@
+---
+name: Question
+about: Ask a question on using go-zero or goctl
+title: ''
+labels: ''
+assignees: ''
+
+---
+
+

.github/workflows/go.yml

@@ -7,28 +7,50 @@ on:
     branches: [ master ]
 
 jobs:
-  build:
-    name: Build
+  test-linux:
+    name: Linux
     runs-on: ubuntu-latest
     steps:
+      - name: Set up Go 1.x
+        uses: actions/setup-go@v2
+        with:
+          go-version: ^1.15
+        id: go
 
-    - name: Set up Go 1.x
-      uses: actions/setup-go@v2
-      with:
-        go-version: ^1.13
-      id: go
+      - name: Check out code into the Go module directory
+        uses: actions/checkout@v2
 
-    - name: Check out code into the Go module directory
-      uses: actions/checkout@v2
+      - name: Get dependencies
+        run: |
+          go get -v -t -d ./...
 
-    - name: Get dependencies
-      run: |
-        go get -v -t -d ./...
+      - name: Lint
+        run: |
+          go vet -stdmethods=false $(go list ./...)
+          go install mvdan.cc/gofumpt@latest
+          test -z "$(gofumpt -s -l -extra .)" || echo "Please run 'gofumpt -l -w -extra .'"
 
-    - name: Test
-      run: go test -race -coverprofile=coverage.txt -covermode=atomic ./...
+      - name: Test
+        run: go test -race -coverprofile=coverage.txt -covermode=atomic ./...
 
-    - name: Codecov
-      uses: codecov/codecov-action@v1.0.6
-      with:
-        token: ${{secrets.CODECOV_TOKEN}}
+      - name: Codecov
+        uses: codecov/codecov-action@v2
+
+  test-win:
+    name: Windows
+    runs-on: windows-latest
+    steps:
+      - name: Set up Go 1.x
+        uses: actions/setup-go@v2
+        with:
+          go-version: ^1.15
+
+      - name: Checkout codebase
+        uses: actions/checkout@v2
+
+      - name: Test
+        run: |
+          go mod verify
+          go mod download
+          go test -v -race ./...
+          cd tools/goctl && go build -v goctl.go

.github/workflows/issue-translator.yml

@@ -0,0 +1,18 @@
+name: 'issue-translator'
+on: 
+  issue_comment: 
+    types: [created]
+  issues: 
+    types: [opened]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: tomsun28/issues-translate-action@v2.6
+        with:
+          IS_MODIFY_TITLE: true
+          # not require, default false, . Decide whether to modify the issue title
+          # if true, the robot account @Issues-translate-bot must have modification permissions, invite @Issues-translate-bot to your project or use your custom bot.
+          CUSTOM_BOT_NOTE: Bot detected the issue body's language is not English, translate it automatically. 👯👭🏻🧑‍🤝‍🧑👫🧑🏿‍🤝‍🧑🏻👩🏾‍🤝‍👨🏿👬🏿
+          # not require. Customize the translation robot prefix message.

.github/workflows/issues.yml

@@ -0,0 +1,19 @@
+name: Close inactive issues
+on:
+  schedule:
+    - cron: "30 1 * * *"
+
+jobs:
+  close-issues:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/stale@v3
+        with:
+          days-before-issue-stale: 30
+          days-before-issue-close: 14
+          stale-issue-label: "stale"
+          stale-issue-message: "This issue is stale because it has been open for 30 days with no activity."
+          close-issue-message: "This issue was closed because it has been inactive for 14 days since being marked as stale."
+          days-before-pr-stale: -1
+          days-before-pr-close: -1
+          repo-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/reviewdog.yml

@@ -0,0 +1,19 @@
+name: reviewdog
+on: [pull_request]
+jobs:
+  staticcheck:
+    name: runner / staticcheck
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: reviewdog/action-staticcheck@v1
+        with:
+          github_token: ${{ secrets.github_token }}
+          # Change reviewdog reporter if you need [github-pr-check,github-check,github-pr-review].
+          reporter: github-pr-review
+          # Report all results.
+          filter_mode: nofilter
+          # Exit with 1 when it find at least one finding.
+          fail_on_error: true
+          # Set staticcheck flags
+          staticcheck_flags: -checks=inherit,-SA1019,-SA1029,-SA5008

.gitignore

@@ -10,10 +10,15 @@
 !*/
 !api
 
+# ignore
 .idea
 **/.DS_Store
 **/logs
 
+# for test purpose
+**/adhoc
+**/testdata
+
 # gitlab ci
 .cache
 

CONTRIBUTING.md

@@ -0,0 +1,102 @@
+# Contributing
+
+Welcome to go-zero!
+
+-   [Before you get started](#before-you-get-started)
+    -   [Code of Conduct](#code-of-conduct)
+    -   [Community Expectations](#community-expectations)
+-   [Getting started](#getting-started)
+-   [Your First Contribution](#your-first-contribution)
+    -   [Find something to work on](#find-something-to-work-on)
+        -   [Find a good first topic](#find-a-good-first-topic)
+        -   [Work on an Issue](#work-on-an-issue)
+        -   [File an Issue](#file-an-issue)
+-   [Contributor Workflow](#contributor-workflow)
+    -   [Creating Pull Requests](#creating-pull-requests)
+    -   [Code Review](#code-review)
+    -   [Testing](#testing)
+
+# Before you get started
+
+## Code of Conduct
+
+Please make sure to read and observe our [Code of Conduct](/code-of-conduct.md).
+
+## Community Expectations
+
+go-zero is a community project driven by its community which strives to promote a healthy, friendly and productive environment.
+go-zero is a web and rpc framework written in Go. It's born to ensure the stability of the busy sites with resilient design. Builtin goctl greatly improves the development productivity.
+
+# Getting started
+
+- Fork the repository on GitHub.
+- Make your changes on your fork repository.
+- Submit a PR.
+
+
+# Your First Contribution
+
+We will help you to contribute in different areas like filing issues, developing features, fixing critical bugs and
+getting your work reviewed and merged.
+
+If you have questions about the development process,
+feel free to [file an issue](https://github.com/zeromicro/go-zero/issues/new/choose).
+
+## Find something to work on
+
+We are always in need of help, be it fixing documentation, reporting bugs or writing some code.
+Look at places where you feel best coding practices aren't followed, code refactoring is needed or tests are missing.
+Here is how you get started.
+
+### Find a good first topic
+
+[go-zero](https://github.com/zeromicro/go-zero) has beginner-friendly issues that provide a good first issue.
+For example, [go-zero](https://github.com/zeromicro/go-zero) has
+[help wanted](https://github.com/zeromicro/go-zero/issues?q=is%3Aopen+is%3Aissue+label%3A%22help+wanted%22) and
+[good first issue](https://github.com/zeromicro/go-zero/issues?q=is%3Aopen+is%3Aissue+label%3A%22good+first+issue%22)
+labels for issues that should not need deep knowledge of the system.
+We can help new contributors who wish to work on such issues.
+
+Another good way to contribute is to find a documentation improvement, such as a missing/broken link.
+Please see [Contributing](#contributing) below for the workflow.
+
+#### Work on an issue
+
+When you are willing to take on an issue, just reply on the issue. The maintainer will assign it to you.
+
+### File an Issue
+
+While we encourage everyone to contribute code, it is also appreciated when someone reports an issue.
+
+Please follow the prompted submission guidelines while opening an issue.
+
+# Contributor Workflow
+
+Please do not ever hesitate to ask a question or send a pull request.
+
+This is a rough outline of what a contributor's workflow looks like:
+
+- Create a topic branch from where to base the contribution. This is usually master.
+- Make commits of logical units.
+- Push changes in a topic branch to a personal fork of the repository.
+- Submit a pull request to [go-zero](https://github.com/zeromicro/go-zero).
+
+## Creating Pull Requests
+
+Pull requests are often called simply "PR".
+go-zero generally follows the standard [github pull request](https://help.github.com/articles/about-pull-requests/) process.
+To submit a proposed change, please develop the code/fix and add new test cases.
+After that, run these local verifications before submitting pull request to predict the pass or
+fail of continuous integration.
+
+* Format the code with `gofmt`
+* Run the test with data race enabled `go test -race ./...`
+
+## Code Review
+
+To make it easier for your PR to receive reviews, consider the reviewers will need you to:
+
+* follow [good coding guidelines](https://github.com/golang/go/wiki/CodeReviewComments).
+* write [good commit messages](https://chris.beams.io/posts/git-commit/).
+* break large changes into a logical series of smaller patches which individually make easily understandable changes, and in aggregate solve a broader issue.
+

ROADMAP.md

@@ -0,0 +1,28 @@
+# go-zero Roadmap
+
+This document defines a high level roadmap for go-zero development and upcoming releases.
+Community and contributor involvement is vital for successfully implementing all desired items for each release.
+We hope that the items listed below will inspire further engagement from the community to keep go-zero progressing and shipping exciting and valuable features.
+
+## 2021 Q2
+- [x] Support service discovery through K8S client api
+- [x] Log full sql statements for easier sql problem solving
+
+## 2021 Q3
+- [x] Support `goctl model pg` to support PostgreSQL code generation
+- [x] Adapt builtin tracing mechanism to opentracing solutions
+
+## 2021 Q4
+- [x] Support `username/password` authentication in ETCD
+- [x] Support `SSL/TLS` in ETCD
+- [x] Support `SSL/TLS` in `zRPC`
+- [x] Support `TLS` in redis connections
+- [x] Support `goctl bug` to report bugs conveniently
+
+## 2022
+- [x] Support `context` in redis related methods for timeout and tracing
+- [x] Support `context` in sql related methods for timeout and tracing
+- [ ] Support `context` in mongodb related methods for timeout and tracing
+- [ ] Add `httpx.Client` with governance, like circuit breaker etc.
+- [ ] Support `goctl doctor` command to report potential issues for given service
+- [ ] Support `goctl mock` command to start a mocking server with given `.api` file

code-of-conduct.md

@@ -0,0 +1,76 @@
+# Contributor Covenant Code of Conduct
+
+## Our Pledge
+
+In the interest of fostering an open and welcoming environment, we as
+contributors and maintainers pledge to make participation in our project and
+our community a harassment-free experience for everyone, regardless of age, body
+size, disability, ethnicity, sex characteristics, gender identity and expression,
+level of experience, education, socio-economic status, nationality, personal
+appearance, race, religion, or sexual identity and orientation.
+
+## Our Standards
+
+Examples of behavior that contributes to creating a positive environment
+include:
+
+* Using welcoming and inclusive language
+* Being respectful of differing viewpoints and experiences
+* Gracefully accepting constructive criticism
+* Focusing on what is best for the community
+* Showing empathy towards other community members
+
+Examples of unacceptable behavior by participants include:
+
+* The use of sexualized language or imagery and unwelcome sexual attention or
+  advances
+* Trolling, insulting/derogatory comments, and personal or political attacks
+* Public or private harassment
+* Publishing others' private information, such as a physical or electronic
+  address, without explicit permission
+* Other conduct which could reasonably be considered inappropriate in a
+  professional setting
+
+## Our Responsibilities
+
+Project maintainers are responsible for clarifying the standards of acceptable
+behavior and are expected to take appropriate and fair corrective action in
+response to any instances of unacceptable behavior.
+
+Project maintainers have the right and responsibility to remove, edit, or
+reject comments, commits, code, wiki edits, issues, and other contributions
+that are not aligned to this Code of Conduct, or to ban temporarily or
+permanently any contributor for other behaviors that they deem inappropriate,
+threatening, offensive, or harmful.
+
+## Scope
+
+This Code of Conduct applies within all project spaces, and it also applies when
+an individual is representing the project or its community in public spaces.
+Examples of representing a project or community include using an official
+project e-mail address, posting via an official social media account, or acting
+as an appointed representative at an online or offline event. Representation of
+a project may be further defined and clarified by project maintainers.
+
+## Enforcement
+
+Instances of abusive, harassing, or otherwise unacceptable behavior may be
+reported by contacting the project team at [INSERT EMAIL ADDRESS]. All
+complaints will be reviewed and investigated and will result in a response that
+is deemed necessary and appropriate to the circumstances. The project team is
+obligated to maintain confidentiality with regard to the reporter of an incident.
+Further details of specific enforcement policies may be posted separately.
+
+Project maintainers who do not follow or enforce the Code of Conduct in good
+faith may face temporary or permanent repercussions as determined by other
+members of the project's leadership.
+
+## Attribution
+
+This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4,
+available at https://www.contributor-covenant.org/version/1/4/code-of-conduct.html
+
+[homepage]: https://www.contributor-covenant.org
+
+For answers to common questions about this code of conduct, see
+https://www.contributor-covenant.org/faq

core/bloom/bloom.go

@@ -4,8 +4,8 @@ import (
 	"errors"
 	"strconv"
 
-	"github.com/tal-tech/go-zero/core/hash"
-	"github.com/tal-tech/go-zero/core/stores/redis"
+	"github.com/zeromicro/go-zero/core/hash"
+	"github.com/zeromicro/go-zero/core/stores/redis"
 )
 
 const (
@@ -27,43 +27,43 @@ return true
 `
 )
 
+// ErrTooLargeOffset indicates the offset is too large in bitset.
 var ErrTooLargeOffset = errors.New("too large offset")
 
 type (
-	BitSetProvider interface {
+	// A Filter is a bloom filter.
+	Filter struct {
+		bits   uint
+		bitSet bitSetProvider
+	}
+
+	bitSetProvider interface {
 		check([]uint) (bool, error)
 		set([]uint) error
 	}
-
-	BloomFilter struct {
-		bits   uint
-		bitSet BitSetProvider
-	}
 )
 
-// New create a BloomFilter, store is the backed redis, key is the key for the bloom filter,
+// New create a Filter, store is the backed redis, key is the key for the bloom filter,
 // bits is how many bits will be used, maps is how many hashes for each addition.
 // best practices:
 // elements - means how many actual elements
 // when maps = 14, formula: 0.7*(bits/maps), bits = 20*elements, the error rate is 0.000067 < 1e-4
 // for detailed error rate table, see http://pages.cs.wisc.edu/~cao/papers/summary-cache/node8.html
-func New(store *redis.Redis, key string, bits uint) *BloomFilter {
-	return &BloomFilter{
+func New(store *redis.Redis, key string, bits uint) *Filter {
+	return &Filter{
 		bits:   bits,
 		bitSet: newRedisBitSet(store, key, bits),
 	}
 }
 
-func (f *BloomFilter) Add(data []byte) error {
+// Add adds data into f.
+func (f *Filter) Add(data []byte) error {
 	locations := f.getLocations(data)
-	err := f.bitSet.set(locations)
-	if err != nil {
-		return err
-	}
-	return nil
+	return f.bitSet.set(locations)
 }
 
-func (f *BloomFilter) Exists(data []byte) (bool, error) {
+// Exists checks if data is in f.
+func (f *Filter) Exists(data []byte) (bool, error) {
 	locations := f.getLocations(data)
 	isSet, err := f.bitSet.check(locations)
 	if err != nil {
@@ -76,7 +76,7 @@ func (f *BloomFilter) Exists(data []byte) (bool, error) {
 	return true, nil
 }
 
-func (f *BloomFilter) getLocations(data []byte) []uint {
+func (f *Filter) getLocations(data []byte) []uint {
 	locations := make([]uint, maps)
 	for i := uint(0); i < maps; i++ {
 		hashValue := hash.Hash(append(data, byte(i)))
@@ -127,11 +127,12 @@ func (r *redisBitSet) check(offsets []uint) (bool, error) {
 		return false, err
 	}
 
-	if exists, ok := resp.(int64); !ok {
+	exists, ok := resp.(int64)
+	if !ok {
 		return false, nil
-	} else {
-		return exists == 1, nil
 	}
+
+	return exists == 1, nil
 }
 
 func (r *redisBitSet) del() error {
@@ -152,7 +153,7 @@ func (r *redisBitSet) set(offsets []uint) error {
 	_, err = r.store.Eval(setScript, []string{r.key}, args)
 	if err == redis.Nil {
 		return nil
-	} else {
-		return err
 	}
+
+	return err
 }

core/bloom/bloom_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/stores/redis/redistest"
+	"github.com/zeromicro/go-zero/core/stores/redis/redistest"
 )
 
 func TestRedisBitSet_New_Set_Test(t *testing.T) {

core/breaker/breaker.go

@@ -6,11 +6,11 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/tal-tech/go-zero/core/mathx"
-	"github.com/tal-tech/go-zero/core/proc"
-	"github.com/tal-tech/go-zero/core/stat"
-	"github.com/tal-tech/go-zero/core/stringx"
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/mathx"
+	"github.com/zeromicro/go-zero/core/proc"
+	"github.com/zeromicro/go-zero/core/stat"
+	"github.com/zeromicro/go-zero/core/stringx"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 const (
@@ -18,12 +18,14 @@ const (
 	timeFormat        = "15:04:05"
 )
 
-// ErrServiceUnavailable is returned when the CB state is open
+// ErrServiceUnavailable is returned when the Breaker state is open.
 var ErrServiceUnavailable = errors.New("circuit breaker is open")
 
 type (
+	// Acceptable is the func to check if the error can be accepted.
 	Acceptable func(err error) bool
 
+	// A Breaker represents a circuit breaker.
 	Breaker interface {
 		// Name returns the name of the Breaker.
 		Name() string
@@ -61,10 +63,14 @@ type (
 		DoWithFallbackAcceptable(req func() error, fallback func(err error) error, acceptable Acceptable) error
 	}
 
+	// Option defines the method to customize a Breaker.
 	Option func(breaker *circuitBreaker)
 
+	// Promise interface defines the callbacks that returned by Breaker.Allow.
 	Promise interface {
+		// Accept tells the Breaker that the call is successful.
 		Accept()
+		// Reject tells the Breaker that the call is failed.
 		Reject(reason string)
 	}
 
@@ -89,6 +95,8 @@ type (
 	}
 )
 
+// NewBreaker returns a Breaker object.
+// opts can be used to customize the Breaker.
 func NewBreaker(opts ...Option) Breaker {
 	var b circuitBreaker
 	for _, opt := range opts {
@@ -127,6 +135,7 @@ func (cb *circuitBreaker) Name() string {
 	return cb.name
 }
 
+// WithName returns a function to set the name of a Breaker.
 func WithName(name string) Option {
 	return func(b *circuitBreaker) {
 		b.name = name

core/breaker/breaker_test.go

@@ -8,7 +8,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/stat"
+	"github.com/zeromicro/go-zero/core/stat"
 )
 
 func init() {
@@ -122,8 +122,7 @@ func BenchmarkGoogleBreaker(b *testing.B) {
 	}
 }
 
-type mockedPromise struct {
-}
+type mockedPromise struct{}
 
 func (m *mockedPromise) Accept() {
 }

core/breaker/breakers.go

@@ -7,24 +7,28 @@ var (
 	breakers = make(map[string]Breaker)
 )
 
+// Do calls Breaker.Do on the Breaker with given name.
 func Do(name string, req func() error) error {
 	return do(name, func(b Breaker) error {
 		return b.Do(req)
 	})
 }
 
+// DoWithAcceptable calls Breaker.DoWithAcceptable on the Breaker with given name.
 func DoWithAcceptable(name string, req func() error, acceptable Acceptable) error {
 	return do(name, func(b Breaker) error {
 		return b.DoWithAcceptable(req, acceptable)
 	})
 }
 
+// DoWithFallback calls Breaker.DoWithFallback on the Breaker with given name.
 func DoWithFallback(name string, req func() error, fallback func(err error) error) error {
 	return do(name, func(b Breaker) error {
 		return b.DoWithFallback(req, fallback)
 	})
 }
 
+// DoWithFallbackAcceptable calls Breaker.DoWithFallbackAcceptable on the Breaker with given name.
 func DoWithFallbackAcceptable(name string, req func() error, fallback func(err error) error,
 	acceptable Acceptable) error {
 	return do(name, func(b Breaker) error {
@@ -32,6 +36,7 @@ func DoWithFallbackAcceptable(name string, req func() error, fallback func(err e
 	})
 }
 
+// GetBreaker returns the Breaker with the given name.
 func GetBreaker(name string) Breaker {
 	lock.RLock()
 	b, ok := breakers[name]
@@ -51,7 +56,8 @@ func GetBreaker(name string) Breaker {
 	return b
 }
 
-func NoBreakFor(name string) {
+// NoBreakerFor disables the circuit breaker for the given name.
+func NoBreakerFor(name string) {
 	lock.Lock()
 	breakers[name] = newNoOpBreaker()
 	lock.Unlock()

core/breaker/breakers_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/stat"
+	"github.com/zeromicro/go-zero/core/stat"
 )
 
 func init() {
@@ -55,7 +55,7 @@ func TestBreakersDoWithAcceptable(t *testing.T) {
 }
 
 func TestBreakersNoBreakerFor(t *testing.T) {
-	NoBreakFor("any")
+	NoBreakerFor("any")
 	errDummy := errors.New("any")
 	for i := 0; i < 10000; i++ {
 		assert.Equal(t, errDummy, GetBreaker("any").Do(func() error {

core/breaker/googlebreaker.go

@@ -4,8 +4,8 @@ import (
 	"math"
 	"time"
 
-	"github.com/tal-tech/go-zero/core/collection"
-	"github.com/tal-tech/go-zero/core/mathx"
+	"github.com/zeromicro/go-zero/core/collection"
+	"github.com/zeromicro/go-zero/core/mathx"
 )
 
 const (
@@ -64,9 +64,9 @@ func (b *googleBreaker) doReq(req func() error, fallback func(err error) error,
 	if err := b.accept(); err != nil {
 		if fallback != nil {
 			return fallback(err)
-		} else {
-			return err
 		}
+
+		return err
 	}
 
 	defer func() {
@@ -94,7 +94,7 @@ func (b *googleBreaker) markFailure() {
 	b.stat.Add(0)
 }
 
-func (b *googleBreaker) history() (accepts int64, total int64) {
+func (b *googleBreaker) history() (accepts, total int64) {
 	b.stat.Reduce(func(b *collection.Bucket) {
 		accepts += int64(b.Sum)
 		total += b.Count

core/breaker/googlebreaker_test.go

@@ -7,9 +7,9 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/collection"
-	"github.com/tal-tech/go-zero/core/mathx"
-	"github.com/tal-tech/go-zero/core/stat"
+	"github.com/zeromicro/go-zero/core/collection"
+	"github.com/zeromicro/go-zero/core/mathx"
+	"github.com/zeromicro/go-zero/core/stat"
 )
 
 const (

core/cmdline/input.go

@@ -7,11 +7,13 @@ import (
 	"strings"
 )
 
+// EnterToContinue let stdin waiting for an enter key to continue.
 func EnterToContinue() {
 	fmt.Print("Press 'Enter' to continue...")
 	bufio.NewReader(os.Stdin).ReadBytes('\n')
 }
 
+// ReadLine shows prompt to stdout and read a line from stdin.
 func ReadLine(prompt string) string {
 	fmt.Print(prompt)
 	input, _ := bufio.NewReader(os.Stdin).ReadString('\n')

core/cmdline/input_test.go

@@ -8,8 +8,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/iox"
-	"github.com/tal-tech/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/iox"
+	"github.com/zeromicro/go-zero/core/lang"
 )
 
 func TestEnterToContinue(t *testing.T) {

core/codec/aesecb.go

@@ -7,9 +7,10 @@ import (
 	"encoding/base64"
 	"errors"
 
-	"github.com/tal-tech/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/logx"
 )
 
+// ErrPaddingSize indicates bad padding size.
 var ErrPaddingSize = errors.New("padding size error")
 
 type ecb struct {
@@ -26,6 +27,7 @@ func newECB(b cipher.Block) *ecb {
 
 type ecbEncrypter ecb
 
+// NewECBEncrypter returns an ECB encrypter.
 func NewECBEncrypter(b cipher.Block) cipher.BlockMode {
 	return (*ecbEncrypter)(newECB(b))
 }
@@ -52,6 +54,7 @@ func (x *ecbEncrypter) CryptBlocks(dst, src []byte) {
 
 type ecbDecrypter ecb
 
+// NewECBDecrypter returns an ECB decrypter.
 func NewECBDecrypter(b cipher.Block) cipher.BlockMode {
 	return (*ecbDecrypter)(newECB(b))
 }
@@ -78,6 +81,7 @@ func (x *ecbDecrypter) CryptBlocks(dst, src []byte) {
 	}
 }
 
+// EcbDecrypt decrypts src with the given key.
 func EcbDecrypt(key, src []byte) ([]byte, error) {
 	block, err := aes.NewCipher(key)
 	if err != nil {
@@ -92,6 +96,8 @@ func EcbDecrypt(key, src []byte) ([]byte, error) {
 	return pkcs5Unpadding(decrypted, decrypter.BlockSize())
 }
 
+// EcbDecryptBase64 decrypts base64 encoded src with the given base64 encoded key.
+// The returned string is also base64 encoded.
 func EcbDecryptBase64(key, src string) (string, error) {
 	keyBytes, err := getKeyBytes(key)
 	if err != nil {
@@ -111,6 +117,7 @@ func EcbDecryptBase64(key, src string) (string, error) {
 	return base64.StdEncoding.EncodeToString(decryptedBytes), nil
 }
 
+// EcbEncrypt encrypts src with the given key.
 func EcbEncrypt(key, src []byte) ([]byte, error) {
 	block, err := aes.NewCipher(key)
 	if err != nil {
@@ -126,6 +133,8 @@ func EcbEncrypt(key, src []byte) ([]byte, error) {
 	return crypted, nil
 }
 
+// EcbEncryptBase64 encrypts base64 encoded src with the given base64 encoded key.
+// The returned string is also base64 encoded.
 func EcbEncryptBase64(key, src string) (string, error) {
 	keyBytes, err := getKeyBytes(key)
 	if err != nil {
@@ -146,15 +155,16 @@ func EcbEncryptBase64(key, src string) (string, error) {
 }
 
 func getKeyBytes(key string) ([]byte, error) {
-	if len(key) > 32 {
-		if keyBytes, err := base64.StdEncoding.DecodeString(key); err != nil {
-			return nil, err
-		} else {
-			return keyBytes, nil
-		}
+	if len(key) <= 32 {
+		return []byte(key), nil
 	}
 
-	return []byte(key), nil
+	keyBytes, err := base64.StdEncoding.DecodeString(key)
+	if err != nil {
+		return nil, err
+	}
+
+	return keyBytes, nil
 }
 
 func pkcs5Padding(ciphertext []byte, blockSize int) []byte {

core/codec/aesecb_test.go

@@ -0,0 +1,65 @@
+package codec
+
+import (
+	"encoding/base64"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAesEcb(t *testing.T) {
+	var (
+		key     = []byte("q4t7w!z%C*F-JaNdRgUjXn2r5u8x/A?D")
+		val     = []byte("hello")
+		badKey1 = []byte("aaaaaaaaa")
+		// more than 32 chars
+		badKey2 = []byte("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
+	)
+	_, err := EcbEncrypt(badKey1, val)
+	assert.NotNil(t, err)
+	_, err = EcbEncrypt(badKey2, val)
+	assert.NotNil(t, err)
+	dst, err := EcbEncrypt(key, val)
+	assert.Nil(t, err)
+	_, err = EcbDecrypt(badKey1, dst)
+	assert.NotNil(t, err)
+	_, err = EcbDecrypt(badKey2, dst)
+	assert.NotNil(t, err)
+	_, err = EcbDecrypt(key, val)
+	// not enough block, just nil
+	assert.Nil(t, err)
+	src, err := EcbDecrypt(key, dst)
+	assert.Nil(t, err)
+	assert.Equal(t, val, src)
+}
+
+func TestAesEcbBase64(t *testing.T) {
+	const (
+		val     = "hello"
+		badKey1 = "aaaaaaaaa"
+		// more than 32 chars
+		badKey2 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	)
+	key := []byte("q4t7w!z%C*F-JaNdRgUjXn2r5u8x/A?D")
+	b64Key := base64.StdEncoding.EncodeToString(key)
+	b64Val := base64.StdEncoding.EncodeToString([]byte(val))
+	_, err := EcbEncryptBase64(badKey1, val)
+	assert.NotNil(t, err)
+	_, err = EcbEncryptBase64(badKey2, val)
+	assert.NotNil(t, err)
+	_, err = EcbEncryptBase64(b64Key, val)
+	assert.NotNil(t, err)
+	dst, err := EcbEncryptBase64(b64Key, b64Val)
+	assert.Nil(t, err)
+	_, err = EcbDecryptBase64(badKey1, dst)
+	assert.NotNil(t, err)
+	_, err = EcbDecryptBase64(badKey2, dst)
+	assert.NotNil(t, err)
+	_, err = EcbDecryptBase64(b64Key, val)
+	assert.NotNil(t, err)
+	src, err := EcbDecryptBase64(b64Key, dst)
+	assert.Nil(t, err)
+	b, err := base64.StdEncoding.DecodeString(src)
+	assert.Nil(t, err)
+	assert.Equal(t, val, string(b))
+}

core/codec/dh.go

@@ -11,8 +11,11 @@ import (
 // 2048-bit MODP Group
 
 var (
-	ErrInvalidPriKey    = errors.New("invalid private key")
-	ErrInvalidPubKey    = errors.New("invalid public key")
+	// ErrInvalidPriKey indicates the invalid private key.
+	ErrInvalidPriKey = errors.New("invalid private key")
+	// ErrInvalidPubKey indicates the invalid public key.
+	ErrInvalidPubKey = errors.New("invalid public key")
+	// ErrPubKeyOutOfBound indicates the public key is out of bound.
 	ErrPubKeyOutOfBound = errors.New("public key out of bound")
 
 	p, _ = new(big.Int).SetString("FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3DC2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F83655D23DCA3AD961C62F356208552BB9ED529077096966D670C354E4ABC9804F1746C08CA18217C32905E462E36CE3BE39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9DE2BCBF6955817183995497CEA956AE515D2261898FA051015728E5A8AACAA68FFFFFFFFFFFFFFFF", 16)
@@ -20,11 +23,13 @@ var (
 	zero = big.NewInt(0)
 )
 
+// DhKey defines the Diffie Hellman key.
 type DhKey struct {
 	PriKey *big.Int
 	PubKey *big.Int
 }
 
+// ComputeKey returns a key from public key and private key.
 func ComputeKey(pubKey, priKey *big.Int) (*big.Int, error) {
 	if pubKey == nil {
 		return nil, ErrInvalidPubKey
@@ -41,6 +46,7 @@ func ComputeKey(pubKey, priKey *big.Int) (*big.Int, error) {
 	return new(big.Int).Exp(pubKey, priKey, p), nil
 }
 
+// GenerateKey returns a Diffie Hellman key.
 func GenerateKey() (*DhKey, error) {
 	var err error
 	var x *big.Int
@@ -63,10 +69,12 @@ func GenerateKey() (*DhKey, error) {
 	return key, nil
 }
 
+// NewPublicKey returns a public key from the given bytes.
 func NewPublicKey(bs []byte) *big.Int {
 	return new(big.Int).SetBytes(bs)
 }
 
+// Bytes returns public key bytes.
 func (k *DhKey) Bytes() []byte {
 	if k.PubKey == nil {
 		return nil

core/codec/gzip.go

@@ -8,6 +8,7 @@ import (
 
 const unzipLimit = 100 * 1024 * 1024 // 100MB
 
+// Gzip compresses bs.
 func Gzip(bs []byte) []byte {
 	var b bytes.Buffer
 
@@ -18,6 +19,7 @@ func Gzip(bs []byte) []byte {
 	return b.Bytes()
 }
 
+// Gunzip uncompresses bs.
 func Gunzip(bs []byte) ([]byte, error) {
 	r, err := gzip.NewReader(bytes.NewBuffer(bs))
 	if err != nil {

core/codec/hmac.go

@@ -7,12 +7,14 @@ import (
 	"io"
 )
 
+// Hmac returns HMAC bytes for body with the given key.
 func Hmac(key []byte, body string) []byte {
 	h := hmac.New(sha256.New, key)
 	io.WriteString(h, body)
 	return h.Sum(nil)
 }
 
+// HmacBase64 returns the base64 encoded string of HMAC for body with the given key.
 func HmacBase64(key []byte, body string) string {
 	return base64.StdEncoding.EncodeToString(Hmac(key, body))
 }

core/codec/rsa.go

@@ -11,17 +11,22 @@ import (
 )
 
 var (
+	// ErrPrivateKey indicates the invalid private key.
 	ErrPrivateKey = errors.New("private key error")
-	ErrPublicKey  = errors.New("failed to parse PEM block containing the public key")
-	ErrNotRsaKey  = errors.New("key type is not RSA")
+	// ErrPublicKey indicates the invalid public key.
+	ErrPublicKey = errors.New("failed to parse PEM block containing the public key")
+	// ErrNotRsaKey indicates the invalid RSA key.
+	ErrNotRsaKey = errors.New("key type is not RSA")
 )
 
 type (
+	// RsaDecrypter represents a RSA decrypter.
 	RsaDecrypter interface {
 		Decrypt(input []byte) ([]byte, error)
 		DecryptBase64(input string) ([]byte, error)
 	}
 
+	// RsaEncrypter represents a RSA encrypter.
 	RsaEncrypter interface {
 		Encrypt(input []byte) ([]byte, error)
 	}
@@ -41,6 +46,7 @@ type (
 	}
 )
 
+// NewRsaDecrypter returns a RsaDecrypter with the given file.
 func NewRsaDecrypter(file string) (RsaDecrypter, error) {
 	content, err := ioutil.ReadFile(file)
 	if err != nil {
@@ -84,6 +90,7 @@ func (r *rsaDecrypter) DecryptBase64(input string) ([]byte, error) {
 	return r.Decrypt(base64Decoded)
 }
 
+// NewRsaEncrypter returns a RsaEncrypter with the given key.
 func NewRsaEncrypter(key []byte) (RsaEncrypter, error) {
 	block, _ := pem.Decode(key)
 	if block == nil {

core/codec/rsa_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/fs"
+	"github.com/zeromicro/go-zero/core/fs"
 )
 
 const (

core/collection/cache.go

@@ -6,9 +6,9 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/tal-tech/go-zero/core/logx"
-	"github.com/tal-tech/go-zero/core/mathx"
-	"github.com/tal-tech/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/mathx"
+	"github.com/zeromicro/go-zero/core/syncx"
 )
 
 const (
@@ -23,8 +23,10 @@ const (
 var emptyLruCache = emptyLru{}
 
 type (
+	// CacheOption defines the method to customize a Cache.
 	CacheOption func(cache *Cache)
 
+	// A Cache object is a in-memory cache.
 	Cache struct {
 		name           string
 		lock           sync.Mutex
@@ -32,18 +34,19 @@ type (
 		expire         time.Duration
 		timingWheel    *TimingWheel
 		lruCache       lru
-		barrier        syncx.SharedCalls
+		barrier        syncx.SingleFlight
 		unstableExpiry mathx.Unstable
 		stats          *cacheStat
 	}
 )
 
+// NewCache returns a Cache with given expire.
 func NewCache(expire time.Duration, opts ...CacheOption) (*Cache, error) {
 	cache := &Cache{
 		data:           make(map[string]interface{}),
 		expire:         expire,
 		lruCache:       emptyLruCache,
-		barrier:        syncx.NewSharedCalls(),
+		barrier:        syncx.NewSingleFlight(),
 		unstableExpiry: mathx.NewUnstable(expiryDeviation),
 	}
 
@@ -72,6 +75,7 @@ func NewCache(expire time.Duration, opts ...CacheOption) (*Cache, error) {
 	return cache, nil
 }
 
+// Del deletes the item with the given key from c.
 func (c *Cache) Del(key string) {
 	c.lock.Lock()
 	delete(c.data, key)
@@ -80,6 +84,7 @@ func (c *Cache) Del(key string) {
 	c.timingWheel.RemoveTimer(key)
 }
 
+// Get returns the item with the given key from c.
 func (c *Cache) Get(key string) (interface{}, bool) {
 	value, ok := c.doGet(key)
 	if ok {
@@ -91,6 +96,7 @@ func (c *Cache) Get(key string) (interface{}, bool) {
 	return value, ok
 }
 
+// Set sets value into c with key.
 func (c *Cache) Set(key string, value interface{}) {
 	c.lock.Lock()
 	_, ok := c.data[key]
@@ -106,6 +112,9 @@ func (c *Cache) Set(key string, value interface{}) {
 	}
 }
 
+// Take returns the item with the given key.
+// If the item is in c, return it directly.
+// If not, use fetch method to get the item, set into c and return it.
 func (c *Cache) Take(key string, fetch func() (interface{}, error)) (interface{}, error) {
 	if val, ok := c.doGet(key); ok {
 		c.stats.IncrementHit()
@@ -136,11 +145,10 @@ func (c *Cache) Take(key string, fetch func() (interface{}, error)) (interface{}
 	if fresh {
 		c.stats.IncrementMiss()
 		return val, nil
-	} else {
-		// got the result from previous ongoing query
-		c.stats.IncrementHit()
 	}
 
+	// got the result from previous ongoing query
+	c.stats.IncrementHit()
 	return val, nil
 }
 
@@ -168,6 +176,7 @@ func (c *Cache) size() int {
 	return len(c.data)
 }
 
+// WithLimit customizes a Cache with items up to limit.
 func WithLimit(limit int) CacheOption {
 	return func(cache *Cache) {
 		if limit > 0 {
@@ -176,6 +185,7 @@ func WithLimit(limit int) CacheOption {
 	}
 }
 
+// WithName customizes a Cache with the given name.
 func WithName(name string) CacheOption {
 	return func(cache *Cache) {
 		cache.name = name

core/collection/fifo.go

@@ -2,6 +2,7 @@ package collection
 
 import "sync"
 
+// A Queue is a FIFO queue.
 type Queue struct {
 	lock     sync.Mutex
 	elements []interface{}
@@ -11,6 +12,7 @@ type Queue struct {
 	count    int
 }
 
+// NewQueue returns a Queue object.
 func NewQueue(size int) *Queue {
 	return &Queue{
 		elements: make([]interface{}, size),
@@ -18,6 +20,7 @@ func NewQueue(size int) *Queue {
 	}
 }
 
+// Empty checks if q is empty.
 func (q *Queue) Empty() bool {
 	q.lock.Lock()
 	empty := q.count == 0
@@ -26,6 +29,7 @@ func (q *Queue) Empty() bool {
 	return empty
 }
 
+// Put puts element into q at the last position.
 func (q *Queue) Put(element interface{}) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
@@ -44,6 +48,7 @@ func (q *Queue) Put(element interface{}) {
 	q.count++
 }
 
+// Take takes the first element out of q if not empty.
 func (q *Queue) Take() (interface{}, bool) {
 	q.lock.Lock()
 	defer q.lock.Unlock()

core/collection/fifo_test.go

@@ -61,3 +61,41 @@ func TestPutMore(t *testing.T) {
 		assert.Equal(t, string(element), string(body.([]byte)))
 	}
 }
+
+func TestPutMoreWithHeaderNotZero(t *testing.T) {
+	elements := [][]byte{
+		[]byte("hello"),
+		[]byte("world"),
+		[]byte("again"),
+	}
+	queue := NewQueue(4)
+	for i := range elements {
+		queue.Put(elements[i])
+	}
+
+	// take 1
+	body, ok := queue.Take()
+	assert.True(t, ok)
+	element, ok := body.([]byte)
+	assert.True(t, ok)
+	assert.Equal(t, element, []byte("hello"))
+
+	// put more
+	queue.Put([]byte("b4"))
+	queue.Put([]byte("b5")) // will store in elements[0]
+	queue.Put([]byte("b6")) // cause expansion
+
+	results := [][]byte{
+		[]byte("world"),
+		[]byte("again"),
+		[]byte("b4"),
+		[]byte("b5"),
+		[]byte("b6"),
+	}
+
+	for _, element := range results {
+		body, ok := queue.Take()
+		assert.True(t, ok)
+		assert.Equal(t, string(element), string(body.([]byte)))
+	}
+}

core/collection/ring.go

@@ -1,10 +1,15 @@
 package collection
 
+import "sync"
+
+// A Ring can be used as fixed size ring.
 type Ring struct {
 	elements []interface{}
 	index    int
+	lock     sync.Mutex
 }
 
+// NewRing returns a Ring object with the given size n.
 func NewRing(n int) *Ring {
 	if n < 1 {
 		panic("n should be greater than 0")
@@ -15,12 +20,20 @@ func NewRing(n int) *Ring {
 	}
 }
 
+// Add adds v into r.
 func (r *Ring) Add(v interface{}) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
 	r.elements[r.index%len(r.elements)] = v
 	r.index++
 }
 
+// Take takes all items from r.
 func (r *Ring) Take() []interface{} {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
 	var size int
 	var start int
 	if r.index > len(r.elements) {

core/collection/ring_test.go

@@ -1,6 +1,7 @@
 package collection
 
 import (
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -29,3 +30,30 @@ func TestRingMore(t *testing.T) {
 	elements := ring.Take()
 	assert.ElementsMatch(t, []interface{}{6, 7, 8, 9, 10}, elements)
 }
+
+func TestRingAdd(t *testing.T) {
+	ring := NewRing(5051)
+	wg := sync.WaitGroup{}
+	for i := 1; i <= 100; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			for j := 1; j <= i; j++ {
+				ring.Add(i)
+			}
+		}(i)
+	}
+	wg.Wait()
+	assert.Equal(t, 5050, len(ring.Take()))
+}
+
+func BenchmarkRingAdd(b *testing.B) {
+	ring := NewRing(500)
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			for i := 0; i < b.N; i++ {
+				ring.Add(i)
+			}
+		}
+	})
+}

core/collection/rollingwindow.go

@@ -4,7 +4,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 type (
@@ -73,9 +73,9 @@ func (rw *RollingWindow) span() int {
 	offset := int(timex.Since(rw.lastTime) / rw.interval)
 	if 0 <= offset && offset < rw.size {
 		return offset
-	} else {
-		return rw.size
 	}
+
+	return rw.size
 }
 
 func (rw *RollingWindow) updateOffset() {

core/collection/rollingwindow_test.go

@@ -6,7 +6,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/stringx"
+	"github.com/zeromicro/go-zero/core/stringx"
 )
 
 const duration = time.Millisecond * 50
@@ -139,7 +139,7 @@ func TestRollingWindowBucketTimeBoundary(t *testing.T) {
 func TestRollingWindowDataRace(t *testing.T) {
 	const size = 3
 	r := NewRollingWindow(size, duration)
-	var stop = make(chan bool)
+	stop := make(chan bool)
 	go func() {
 		for {
 			select {

core/collection/safemap.go

@@ -18,6 +18,7 @@ type SafeMap struct {
 	dirtyNew    map[interface{}]interface{}
 }
 
+// NewSafeMap returns a SafeMap.
 func NewSafeMap() *SafeMap {
 	return &SafeMap{
 		dirtyOld: make(map[interface{}]interface{}),
@@ -25,6 +26,7 @@ func NewSafeMap() *SafeMap {
 	}
 }
 
+// Del deletes the value with the given key from m.
 func (m *SafeMap) Del(key interface{}) {
 	m.lock.Lock()
 	if _, ok := m.dirtyOld[key]; ok {
@@ -53,18 +55,20 @@ func (m *SafeMap) Del(key interface{}) {
 	m.lock.Unlock()
 }
 
+// Get gets the value with the given key from m.
 func (m *SafeMap) Get(key interface{}) (interface{}, bool) {
 	m.lock.RLock()
 	defer m.lock.RUnlock()
 
 	if val, ok := m.dirtyOld[key]; ok {
 		return val, true
-	} else {
-		val, ok := m.dirtyNew[key]
-		return val, ok
 	}
+
+	val, ok := m.dirtyNew[key]
+	return val, ok
 }
 
+// Set sets the value into m with the given key.
 func (m *SafeMap) Set(key, value interface{}) {
 	m.lock.Lock()
 	if m.deletionOld <= maxDeletion {
@@ -83,6 +87,7 @@ func (m *SafeMap) Set(key, value interface{}) {
 	m.lock.Unlock()
 }
 
+// Size returns the size of m.
 func (m *SafeMap) Size() int {
 	m.lock.RLock()
 	size := len(m.dirtyOld) + len(m.dirtyNew)

core/collection/safemap_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/stringx"
+	"github.com/zeromicro/go-zero/core/stringx"
 )
 
 func TestSafeMap(t *testing.T) {

core/collection/set.go

@@ -1,8 +1,8 @@
 package collection
 
 import (
-	"github.com/tal-tech/go-zero/core/lang"
-	"github.com/tal-tech/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/logx"
 )
 
 const (
@@ -21,6 +21,7 @@ type Set struct {
 	tp   int
 }
 
+// NewSet returns a managed Set, can only put the values with the same type.
 func NewSet() *Set {
 	return &Set{
 		data: make(map[interface{}]lang.PlaceholderType),
@@ -28,6 +29,7 @@ func NewSet() *Set {
 	}
 }
 
+// NewUnmanagedSet returns a unmanaged Set, which can put values with different types.
 func NewUnmanagedSet() *Set {
 	return &Set{
 		data: make(map[interface{}]lang.PlaceholderType),
@@ -35,42 +37,49 @@ func NewUnmanagedSet() *Set {
 	}
 }
 
+// Add adds i into s.
 func (s *Set) Add(i ...interface{}) {
 	for _, each := range i {
 		s.add(each)
 	}
 }
 
+// AddInt adds int values ii into s.
 func (s *Set) AddInt(ii ...int) {
 	for _, each := range ii {
 		s.add(each)
 	}
 }
 
+// AddInt64 adds int64 values ii into s.
 func (s *Set) AddInt64(ii ...int64) {
 	for _, each := range ii {
 		s.add(each)
 	}
 }
 
+// AddUint adds uint values ii into s.
 func (s *Set) AddUint(ii ...uint) {
 	for _, each := range ii {
 		s.add(each)
 	}
 }
 
+// AddUint64 adds uint64 values ii into s.
 func (s *Set) AddUint64(ii ...uint64) {
 	for _, each := range ii {
 		s.add(each)
 	}
 }
 
+// AddStr adds string values ss into s.
 func (s *Set) AddStr(ss ...string) {
 	for _, each := range ss {
 		s.add(each)
 	}
 }
 
+// Contains checks if i is in s.
 func (s *Set) Contains(i interface{}) bool {
 	if len(s.data) == 0 {
 		return false
@@ -81,6 +90,7 @@ func (s *Set) Contains(i interface{}) bool {
 	return ok
 }
 
+// Keys returns the keys in s.
 func (s *Set) Keys() []interface{} {
 	var keys []interface{}
 
@@ -91,13 +101,12 @@ func (s *Set) Keys() []interface{} {
 	return keys
 }
 
+// KeysInt returns the int keys in s.
 func (s *Set) KeysInt() []int {
 	var keys []int
 
 	for key := range s.data {
-		if intKey, ok := key.(int); !ok {
-			continue
-		} else {
+		if intKey, ok := key.(int); ok {
 			keys = append(keys, intKey)
 		}
 	}
@@ -105,13 +114,12 @@ func (s *Set) KeysInt() []int {
 	return keys
 }
 
+// KeysInt64 returns int64 keys in s.
 func (s *Set) KeysInt64() []int64 {
 	var keys []int64
 
 	for key := range s.data {
-		if intKey, ok := key.(int64); !ok {
-			continue
-		} else {
+		if intKey, ok := key.(int64); ok {
 			keys = append(keys, intKey)
 		}
 	}
@@ -119,13 +127,12 @@ func (s *Set) KeysInt64() []int64 {
 	return keys
 }
 
+// KeysUint returns uint keys in s.
 func (s *Set) KeysUint() []uint {
 	var keys []uint
 
 	for key := range s.data {
-		if intKey, ok := key.(uint); !ok {
-			continue
-		} else {
+		if intKey, ok := key.(uint); ok {
 			keys = append(keys, intKey)
 		}
 	}
@@ -133,13 +140,12 @@ func (s *Set) KeysUint() []uint {
 	return keys
 }
 
+// KeysUint64 returns uint64 keys in s.
 func (s *Set) KeysUint64() []uint64 {
 	var keys []uint64
 
 	for key := range s.data {
-		if intKey, ok := key.(uint64); !ok {
-			continue
-		} else {
+		if intKey, ok := key.(uint64); ok {
 			keys = append(keys, intKey)
 		}
 	}
@@ -147,13 +153,12 @@ func (s *Set) KeysUint64() []uint64 {
 	return keys
 }
 
+// KeysStr returns string keys in s.
 func (s *Set) KeysStr() []string {
 	var keys []string
 
 	for key := range s.data {
-		if strKey, ok := key.(string); !ok {
-			continue
-		} else {
+		if strKey, ok := key.(string); ok {
 			keys = append(keys, strKey)
 		}
 	}
@@ -161,11 +166,13 @@ func (s *Set) KeysStr() []string {
 	return keys
 }
 
+// Remove removes i from s.
 func (s *Set) Remove(i interface{}) {
 	s.validate(i)
 	delete(s.data, i)
 }
 
+// Count returns the number of items in s.
 func (s *Set) Count() int {
 	return len(s.data)
 }

core/collection/set_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/logx"
 )
 
 func init() {

core/collection/timingwheel.go

@@ -5,16 +5,18 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/tal-tech/go-zero/core/lang"
-	"github.com/tal-tech/go-zero/core/threading"
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/threading"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 const drainWorkers = 8
 
 type (
+	// Execute defines the method to execute the task.
 	Execute func(key, value interface{})
 
+	// A TimingWheel is a timing wheel object to schedule tasks.
 	TimingWheel struct {
 		interval      time.Duration
 		ticker        timex.Ticker
@@ -54,6 +56,7 @@ type (
 	}
 )
 
+// NewTimingWheel returns a TimingWheel.
 func NewTimingWheel(interval time.Duration, numSlots int, execute Execute) (*TimingWheel, error) {
 	if interval <= 0 || numSlots <= 0 || execute == nil {
 		return nil, fmt.Errorf("interval: %v, slots: %d, execute: %p", interval, numSlots, execute)
@@ -85,10 +88,12 @@ func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execu
 	return tw, nil
 }
 
+// Drain drains all items and executes them.
 func (tw *TimingWheel) Drain(fn func(key, value interface{})) {
 	tw.drainChannel <- fn
 }
 
+// MoveTimer moves the task with the given key to the given delay.
 func (tw *TimingWheel) MoveTimer(key interface{}, delay time.Duration) {
 	if delay <= 0 || key == nil {
 		return
@@ -100,6 +105,7 @@ func (tw *TimingWheel) MoveTimer(key interface{}, delay time.Duration) {
 	}
 }
 
+// RemoveTimer removes the task with the given key.
 func (tw *TimingWheel) RemoveTimer(key interface{}) {
 	if key == nil {
 		return
@@ -108,6 +114,7 @@ func (tw *TimingWheel) RemoveTimer(key interface{}) {
 	tw.removeChannel <- key
 }
 
+// SetTimer sets the task value with the given key to the delay.
 func (tw *TimingWheel) SetTimer(key, value interface{}, delay time.Duration) {
 	if delay <= 0 || key == nil {
 		return
@@ -122,6 +129,7 @@ func (tw *TimingWheel) SetTimer(key, value interface{}, delay time.Duration) {
 	}
 }
 
+// Stop stops tw.
 func (tw *TimingWheel) Stop() {
 	close(tw.stopChannel)
 }
@@ -143,7 +151,7 @@ func (tw *TimingWheel) drainAll(fn func(key, value interface{})) {
 	}
 }
 
-func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos int, circle int) {
+func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos, circle int) {
 	steps := int(d / tw.interval)
 	pos = (tw.tickedPos + steps) % tw.numSlots
 	circle = (steps - 1) / tw.numSlots

core/collection/timingwheel_test.go

@@ -8,10 +8,10 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/lang"
-	"github.com/tal-tech/go-zero/core/stringx"
-	"github.com/tal-tech/go-zero/core/syncx"
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/stringx"
+	"github.com/zeromicro/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 const (

core/conf/config.go

@@ -7,7 +7,7 @@ import (
 	"os"
 	"path"
 
-	"github.com/tal-tech/go-zero/core/mapping"
+	"github.com/zeromicro/go-zero/core/mapping"
 )
 
 var loaders = map[string]func([]byte, interface{}) error{
@@ -16,26 +16,43 @@ var loaders = map[string]func([]byte, interface{}) error{
 	".yml":  LoadConfigFromYamlBytes,
 }
 
-func LoadConfig(file string, v interface{}) error {
-	if content, err := ioutil.ReadFile(file); err != nil {
+// LoadConfig loads config into v from file, .json, .yaml and .yml are acceptable.
+func LoadConfig(file string, v interface{}, opts ...Option) error {
+	content, err := ioutil.ReadFile(file)
+	if err != nil {
 		return err
-	} else if loader, ok := loaders[path.Ext(file)]; ok {
-		return loader([]byte(os.ExpandEnv(string(content))), v)
-	} else {
-		return fmt.Errorf("unrecoginized file type: %s", file)
 	}
+
+	loader, ok := loaders[path.Ext(file)]
+	if !ok {
+		return fmt.Errorf("unrecognized file type: %s", file)
+	}
+
+	var opt options
+	for _, o := range opts {
+		o(&opt)
+	}
+
+	if opt.env {
+		return loader([]byte(os.ExpandEnv(string(content))), v)
+	}
+
+	return loader(content, v)
 }
 
+// LoadConfigFromJsonBytes loads config into v from content json bytes.
 func LoadConfigFromJsonBytes(content []byte, v interface{}) error {
 	return mapping.UnmarshalJsonBytes(content, v)
 }
 
+// LoadConfigFromYamlBytes loads config into v from content yaml bytes.
 func LoadConfigFromYamlBytes(content []byte, v interface{}) error {
 	return mapping.UnmarshalYamlBytes(content, v)
 }
 
-func MustLoad(path string, v interface{}) {
-	if err := LoadConfig(path, v); err != nil {
+// MustLoad loads config into v from path, exits on error.
+func MustLoad(path string, v interface{}, opts ...Option) {
+	if err := LoadConfig(path, v, opts...); err != nil {
 		log.Fatalf("error: config file %s, %s", path, err.Error())
 	}
 }

core/conf/config_test.go

@@ -6,8 +6,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/fs"
-	"github.com/tal-tech/go-zero/core/hash"
+	"github.com/zeromicro/go-zero/core/fs"
+	"github.com/zeromicro/go-zero/core/hash"
 )
 
 func TestLoadConfig_notExists(t *testing.T) {
@@ -30,7 +30,8 @@ func TestConfigJson(t *testing.T) {
 	text := `{
 	"a": "foo",
 	"b": 1,
-	"c": "${FOO}"
+	"c": "${FOO}",
+	"d": "abcd!@#$112"
 }`
 	for _, test := range tests {
 		test := test
@@ -45,11 +46,49 @@ func TestConfigJson(t *testing.T) {
 				A string `json:"a"`
 				B int    `json:"b"`
 				C string `json:"c"`
+				D string `json:"d"`
 			}
 			MustLoad(tmpfile, &val)
 			assert.Equal(t, "foo", val.A)
 			assert.Equal(t, 1, val.B)
+			assert.Equal(t, "${FOO}", val.C)
+			assert.Equal(t, "abcd!@#$112", val.D)
+		})
+	}
+}
+
+func TestConfigJsonEnv(t *testing.T) {
+	tests := []string{
+		".json",
+		".yaml",
+		".yml",
+	}
+	text := `{
+	"a": "foo",
+	"b": 1,
+	"c": "${FOO}",
+	"d": "abcd!@#$a12 3"
+}`
+	for _, test := range tests {
+		test := test
+		t.Run(test, func(t *testing.T) {
+			os.Setenv("FOO", "2")
+			defer os.Unsetenv("FOO")
+			tmpfile, err := createTempFile(test, text)
+			assert.Nil(t, err)
+			defer os.Remove(tmpfile)
+
+			var val struct {
+				A string `json:"a"`
+				B int    `json:"b"`
+				C string `json:"c"`
+				D string `json:"d"`
+			}
+			MustLoad(tmpfile, &val, UseEnv())
+			assert.Equal(t, "foo", val.A)
+			assert.Equal(t, 1, val.B)
 			assert.Equal(t, "2", val.C)
+			assert.Equal(t, "abcd!@# 3", val.D)
 		})
 	}
 }

core/conf/options.go

@@ -0,0 +1,17 @@
+package conf
+
+type (
+	// Option defines the method to customize the config options.
+	Option func(opt *options)
+
+	options struct {
+		env bool
+	}
+)
+
+// UseEnv customizes the config to use environment variables.
+func UseEnv() Option {
+	return func(opt *options) {
+		opt.env = true
+	}
+}

core/conf/properties.go

@@ -2,11 +2,12 @@ package conf
 
 import (
 	"fmt"
+	"os"
 	"strconv"
 	"strings"
 	"sync"
 
-	"github.com/tal-tech/go-zero/core/iox"
+	"github.com/zeromicro/go-zero/core/iox"
 )
 
 // PropertyError represents a configuration error message.
@@ -30,14 +31,19 @@ type mapBasedProperties struct {
 	lock       sync.RWMutex
 }
 
-// Loads the properties into a properties configuration instance.
+// LoadProperties loads the properties into a properties configuration instance.
 // Returns an error that indicates if there was a problem loading the configuration.
-func LoadProperties(filename string) (Properties, error) {
+func LoadProperties(filename string, opts ...Option) (Properties, error) {
 	lines, err := iox.ReadTextLines(filename, iox.WithoutBlank(), iox.OmitWithPrefix("#"))
 	if err != nil {
 		return nil, err
 	}
 
+	var opt options
+	for _, o := range opts {
+		o(&opt)
+	}
+
 	raw := make(map[string]string)
 	for i := range lines {
 		pair := strings.Split(lines[i], "=")
@@ -50,7 +56,11 @@ func LoadProperties(filename string) (Properties, error) {
 
 		key := strings.TrimSpace(pair[0])
 		value := strings.TrimSpace(pair[1])
-		raw[key] = value
+		if opt.env {
+			raw[key] = os.ExpandEnv(value)
+		} else {
+			raw[key] = value
+		}
 	}
 
 	return &mapBasedProperties{
@@ -87,7 +97,7 @@ func (config *mapBasedProperties) SetInt(key string, value int) {
 	config.lock.Unlock()
 }
 
-// Dumps the configuration internal map into a string.
+// ToString dumps the configuration internal map into a string.
 func (config *mapBasedProperties) ToString() string {
 	config.lock.RLock()
 	ret := fmt.Sprintf("%s", config.properties)
@@ -96,12 +106,12 @@ func (config *mapBasedProperties) ToString() string {
 	return ret
 }
 
-// Returns the error message.
+// Error returns the error message.
 func (configError *PropertyError) Error() string {
 	return configError.message
 }
 
-// Builds a new properties configuration structure
+// NewProperties builds a new properties configuration structure.
 func NewProperties() Properties {
 	return &mapBasedProperties{
 		properties: make(map[string]string),

core/conf/properties_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/fs"
+	"github.com/zeromicro/go-zero/core/fs"
 )
 
 func TestProperties(t *testing.T) {
@@ -31,6 +31,39 @@ func TestProperties(t *testing.T) {
 	assert.Contains(t, val, "app.threads")
 }
 
+func TestPropertiesEnv(t *testing.T) {
+	text := `app.name = test
+
+    app.program=app
+
+	app.env1 = ${FOO}
+	app.env2 = $none
+
+    # this is comment
+    app.threads = 5`
+	tmpfile, err := fs.TempFilenameWithText(text)
+	assert.Nil(t, err)
+	defer os.Remove(tmpfile)
+
+	os.Setenv("FOO", "2")
+	defer os.Unsetenv("FOO")
+
+	props, err := LoadProperties(tmpfile, UseEnv())
+	assert.Nil(t, err)
+	assert.Equal(t, "test", props.GetString("app.name"))
+	assert.Equal(t, "app", props.GetString("app.program"))
+	assert.Equal(t, 5, props.GetInt("app.threads"))
+	assert.Equal(t, "2", props.GetString("app.env1"))
+	assert.Equal(t, "", props.GetString("app.env2"))
+
+	val := props.ToString()
+	assert.Contains(t, val, "app.name")
+	assert.Contains(t, val, "app.program")
+	assert.Contains(t, val, "app.threads")
+	assert.Contains(t, val, "app.env1")
+	assert.Contains(t, val, "app.env2")
+}
+
 func TestLoadProperties_badContent(t *testing.T) {
 	filename, err := fs.TempFilenameWithText("hello")
 	assert.Nil(t, err)

core/contextx/deadline.go

@@ -1,17 +0,0 @@
-package contextx
-
-import (
-	"context"
-	"time"
-)
-
-func ShrinkDeadline(ctx context.Context, timeout time.Duration) (context.Context, func()) {
-	if deadline, ok := ctx.Deadline(); ok {
-		leftTime := time.Until(deadline)
-		if leftTime < timeout {
-			timeout = leftTime
-		}
-	}
-
-	return context.WithDeadline(ctx, time.Now().Add(timeout))
-}

core/contextx/deadline_test.go

@@ -1,31 +0,0 @@
-package contextx
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestShrinkDeadlineLess(t *testing.T) {
-	deadline := time.Now().Add(time.Second)
-	ctx, cancel := context.WithDeadline(context.Background(), deadline)
-	defer cancel()
-	ctx, cancel = ShrinkDeadline(ctx, time.Minute)
-	defer cancel()
-	dl, ok := ctx.Deadline()
-	assert.True(t, ok)
-	assert.Equal(t, deadline, dl)
-}
-
-func TestShrinkDeadlineMore(t *testing.T) {
-	deadline := time.Now().Add(time.Minute)
-	ctx, cancel := context.WithDeadline(context.Background(), deadline)
-	defer cancel()
-	ctx, cancel = ShrinkDeadline(ctx, time.Second)
-	defer cancel()
-	dl, ok := ctx.Deadline()
-	assert.True(t, ok)
-	assert.True(t, dl.Before(deadline))
-}

core/contextx/unmarshaler.go

@@ -3,7 +3,7 @@ package contextx
 import (
 	"context"
 
-	"github.com/tal-tech/go-zero/core/mapping"
+	"github.com/zeromicro/go-zero/core/mapping"
 )
 
 const contextTagKey = "ctx"
@@ -19,6 +19,7 @@ func (cv contextValuer) Value(key string) (interface{}, bool) {
 	return v, v != nil
 }
 
+// For unmarshals ctx into v.
 func For(ctx context.Context, v interface{}) error {
 	return unmarshaler.UnmarshalValuer(contextValuer{
 		Context: ctx,

core/contextx/unmarshaler_test.go

@@ -47,9 +47,11 @@ func TestUnmarshalContextWithMissing(t *testing.T) {
 		Name string `ctx:"name"`
 		Age  int    `ctx:"age"`
 	}
+	type name string
+	const PersonNameKey name = "name"
 
 	ctx := context.Background()
-	ctx = context.WithValue(ctx, "name", "kevin")
+	ctx = context.WithValue(ctx, PersonNameKey, "kevin")
 
 	var person Person
 	err := For(ctx, &person)

core/contextx/valueonlycontext.go

@@ -21,6 +21,7 @@ func (valueOnlyContext) Err() error {
 	return nil
 }
 
+// ValueOnlyFrom takes all values from the given ctx, without deadline and error control.
 func ValueOnlyFrom(ctx context.Context) context.Context {
 	return valueOnlyContext{
 		Context: ctx,

core/contextx/valueonlycontext_test.go

@@ -9,7 +9,9 @@ import (
 )
 
 func TestContextCancel(t *testing.T) {
-	c := context.WithValue(context.Background(), "key", "value")
+	type key string
+	var nameKey key = "name"
+	c := context.WithValue(context.Background(), nameKey, "value")
 	c1, cancel := context.WithCancel(c)
 	o := ValueOnlyFrom(c1)
 	c2, cancel2 := context.WithCancel(o)

core/discov/accountregistry.go

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

core/discov/accountregistry_test.go

@@ -0,0 +1,21 @@
+package discov
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/zeromicro/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/stringx"
+)
+
+func TestRegisterAccount(t *testing.T) {
+	endpoints := []string{
+		"localhost:2379",
+	}
+	user := "foo" + stringx.Rand()
+	RegisterAccount(endpoints, user, "bar")
+	account, ok := internal.GetAccount(endpoints)
+	assert.True(t, ok)
+	assert.Equal(t, user, account.User)
+	assert.Equal(t, "bar", account.Pass)
+}

core/discov/clients.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/tal-tech/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/discov/internal"
 )
 
 const (
@@ -14,6 +14,7 @@ const (
 
 const timeToLive int64 = 10
 
+// TimeToLive is seconds to live in etcd.
 var TimeToLive = timeToLive
 
 func extract(etcdKey string, index int) (string, bool) {

core/discov/clients_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/discov/internal"
 )
 
 var mockLock sync.Mutex
@@ -28,6 +28,9 @@ func TestExtract(t *testing.T) {
 
 	_, ok = extract("any", -1)
 	assert.False(t, ok)
+
+	_, ok = extract("any", 10)
+	assert.False(t, ok)
 }
 
 func TestMakeKey(t *testing.T) {

core/discov/config.go

@@ -2,16 +2,41 @@ package discov
 
 import "errors"
 
+var (
+	// errEmptyEtcdHosts indicates that etcd hosts are empty.
+	errEmptyEtcdHosts = errors.New("empty etcd hosts")
+	// errEmptyEtcdKey indicates that etcd key is empty.
+	errEmptyEtcdKey = errors.New("empty etcd key")
+)
+
+// EtcdConf is the config item with the given key on etcd.
 type EtcdConf struct {
-	Hosts []string
-	Key   string
+	Hosts              []string
+	Key                string
+	User               string `json:",optional"`
+	Pass               string `json:",optional"`
+	CertFile           string `json:",optional"`
+	CertKeyFile        string `json:",optional=CertFile"`
+	CACertFile         string `json:",optional=CertFile"`
+	InsecureSkipVerify bool   `json:",optional"`
 }
 
+// HasAccount returns if account provided.
+func (c EtcdConf) HasAccount() bool {
+	return len(c.User) > 0 && len(c.Pass) > 0
+}
+
+// HasTLS returns if TLS CertFile/CertKeyFile/CACertFile are provided.
+func (c EtcdConf) HasTLS() bool {
+	return len(c.CertFile) > 0 && len(c.CertKeyFile) > 0 && len(c.CACertFile) > 0
+}
+
+// Validate validates c.
 func (c EtcdConf) Validate() error {
 	if len(c.Hosts) == 0 {
-		return errors.New("empty etcd hosts")
+		return errEmptyEtcdHosts
 	} else if len(c.Key) == 0 {
-		return errors.New("empty etcd key")
+		return errEmptyEtcdKey
 	} else {
 		return nil
 	}

core/discov/config_test.go

@@ -44,3 +44,39 @@ func TestConfig(t *testing.T) {
 		}
 	}
 }
+
+func TestEtcdConf_HasAccount(t *testing.T) {
+	tests := []struct {
+		EtcdConf
+		hasAccount bool
+	}{
+		{
+			EtcdConf: EtcdConf{
+				Hosts: []string{"any"},
+				Key:   "key",
+			},
+			hasAccount: false,
+		},
+		{
+			EtcdConf: EtcdConf{
+				Hosts: []string{"any"},
+				Key:   "key",
+				User:  "foo",
+			},
+			hasAccount: false,
+		},
+		{
+			EtcdConf: EtcdConf{
+				Hosts: []string{"any"},
+				Key:   "key",
+				User:  "foo",
+				Pass:  "bar",
+			},
+			hasAccount: true,
+		},
+	}
+
+	for _, test := range tests {
+		assert.Equal(t, test.hasAccount, test.EtcdConf.HasAccount())
+	}
+}

core/discov/facade.go

@@ -1,47 +0,0 @@
-package discov
-
-import (
-	"github.com/tal-tech/go-zero/core/discov/internal"
-	"github.com/tal-tech/go-zero/core/logx"
-)
-
-type (
-	Facade struct {
-		endpoints []string
-		registry  *internal.Registry
-	}
-
-	FacadeListener interface {
-		OnAdd(key, val string)
-		OnDelete(key string)
-	}
-)
-
-func NewFacade(endpoints []string) Facade {
-	return Facade{
-		endpoints: endpoints,
-		registry:  internal.GetRegistry(),
-	}
-}
-
-func (f Facade) Client() internal.EtcdClient {
-	conn, err := f.registry.GetConn(f.endpoints)
-	logx.Must(err)
-	return conn
-}
-
-func (f Facade) Monitor(key string, l FacadeListener) {
-	f.registry.Monitor(f.endpoints, key, listenerAdapter{l})
-}
-
-type listenerAdapter struct {
-	l FacadeListener
-}
-
-func (la listenerAdapter) OnAdd(kv internal.KV) {
-	la.l.OnAdd(kv.Key, kv.Val)
-}
-
-func (la listenerAdapter) OnDelete(kv internal.KV) {
-	la.l.OnDelete(kv.Key)
-}

core/discov/internal/accountmanager.go

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

core/discov/internal/accountmanager_test.go

@@ -0,0 +1,34 @@
+package internal
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/zeromicro/go-zero/core/stringx"
+)
+
+func TestAccount(t *testing.T) {
+	endpoints := []string{
+		"192.168.0.2:2379",
+		"192.168.0.3:2379",
+		"192.168.0.4:2379",
+	}
+	username := "foo" + stringx.Rand()
+	password := "bar"
+	anotherPassword := "any"
+
+	_, ok := GetAccount(endpoints)
+	assert.False(t, ok)
+
+	AddAccount(endpoints, username, password)
+	account, ok := GetAccount(endpoints)
+	assert.True(t, ok)
+	assert.Equal(t, username, account.User)
+	assert.Equal(t, password, account.Pass)
+
+	AddAccount(endpoints, username, anotherPassword)
+	account, ok = GetAccount(endpoints)
+	assert.True(t, ok)
+	assert.Equal(t, username, account.User)
+	assert.Equal(t, anotherPassword, account.Pass)
+}

core/discov/internal/etcdclient.go

@@ -1,13 +1,15 @@
 //go:generate mockgen -package internal -destination etcdclient_mock.go -source etcdclient.go EtcdClient
+
 package internal
 
 import (
 	"context"
 
-	"go.etcd.io/etcd/clientv3"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"google.golang.org/grpc"
 )
 
+// EtcdClient interface represents an etcd client.
 type EtcdClient interface {
 	ActiveConnection() *grpc.ClientConn
 	Close() error

core/discov/internal/etcdclient_mock.go

@@ -6,10 +6,11 @@ package internal
 
 import (
 	context "context"
-	gomock "github.com/golang/mock/gomock"
-	clientv3 "go.etcd.io/etcd/clientv3"
-	grpc "google.golang.org/grpc"
 	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	grpc "google.golang.org/grpc"
 )
 
 // MockEtcdClient is a mock of EtcdClient interface

core/discov/internal/listener.go

@@ -1,5 +1,6 @@
 package internal
 
+// Listener interface wraps the OnUpdate method.
 type Listener interface {
-	OnUpdate(keys []string, values []string, newKey string)
+	OnUpdate(keys, values []string, newKey string)
 }

core/discov/internal/registry.go

@@ -9,49 +9,63 @@ import (
 	"sync"
 	"time"
 
-	"github.com/tal-tech/go-zero/core/contextx"
-	"github.com/tal-tech/go-zero/core/lang"
-	"github.com/tal-tech/go-zero/core/logx"
-	"github.com/tal-tech/go-zero/core/syncx"
-	"github.com/tal-tech/go-zero/core/threading"
-	"go.etcd.io/etcd/clientv3"
+	"github.com/zeromicro/go-zero/core/contextx"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/threading"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 var (
-	registryInstance = Registry{
+	registry = Registry{
 		clusters: make(map[string]*cluster),
 	}
 	connManager = syncx.NewResourceManager()
 )
 
+// A Registry is a registry that manages the etcd client connections.
 type Registry struct {
 	clusters map[string]*cluster
 	lock     sync.Mutex
 }
 
+// GetRegistry returns a global Registry.
 func GetRegistry() *Registry {
-	return &registryInstance
+	return &registry
 }
 
-func (r *Registry) getCluster(endpoints []string) *cluster {
+// GetConn returns an etcd client connection associated with given endpoints.
+func (r *Registry) GetConn(endpoints []string) (EtcdClient, error) {
+	c, _ := r.getCluster(endpoints)
+	return c.getClient()
+}
+
+// Monitor monitors the key on given etcd endpoints, notify with the given UpdateListener.
+func (r *Registry) Monitor(endpoints []string, key string, l UpdateListener) error {
+	c, exists := r.getCluster(endpoints)
+	// if exists, the existing values should be updated to the listener.
+	if exists {
+		kvs := c.getCurrent(key)
+		for _, kv := range kvs {
+			l.OnAdd(kv)
+		}
+	}
+
+	return c.monitor(key, l)
+}
+
+func (r *Registry) getCluster(endpoints []string) (c *cluster, exists bool) {
 	clusterKey := getClusterKey(endpoints)
 	r.lock.Lock()
 	defer r.lock.Unlock()
-	c, ok := r.clusters[clusterKey]
-	if !ok {
+	c, exists = r.clusters[clusterKey]
+	if !exists {
 		c = newCluster(endpoints)
 		r.clusters[clusterKey] = c
 	}
 
-	return c
-}
-
-func (r *Registry) GetConn(endpoints []string) (EtcdClient, error) {
-	return r.getCluster(endpoints).getClient()
-}
-
-func (r *Registry) Monitor(endpoints []string, key string, l UpdateListener) error {
-	return r.getCluster(endpoints).monitor(key, l)
+	return
 }
 
 type cluster struct {
@@ -90,6 +104,21 @@ func (c *cluster) getClient() (EtcdClient, error) {
 	return val.(EtcdClient), nil
 }
 
+func (c *cluster) getCurrent(key string) []KV {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	var kvs []KV
+	for k, v := range c.values[key] {
+		kvs = append(kvs, KV{
+			Key: k,
+			Val: v,
+		})
+	}
+
+	return kvs
+}
+
 func (c *cluster) handleChanges(key string, kvs []KV) {
 	var add []KV
 	var remove []KV
@@ -193,14 +222,12 @@ func (c *cluster) load(cli EtcdClient, key string) {
 	}
 
 	var kvs []KV
-	c.lock.Lock()
 	for _, ev := range resp.Kvs {
 		kvs = append(kvs, KV{
 			Key: string(ev.Key),
 			Val: string(ev.Value),
 		})
 	}
-	c.lock.Unlock()
 
 	c.handleChanges(key, kvs)
 }
@@ -256,26 +283,34 @@ func (c *cluster) reload(cli EtcdClient) {
 }
 
 func (c *cluster) watch(cli EtcdClient, key string) {
+	for {
+		if c.watchStream(cli, key) {
+			return
+		}
+	}
+}
+
+func (c *cluster) watchStream(cli EtcdClient, key string) bool {
 	rch := cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key), clientv3.WithPrefix())
 	for {
 		select {
 		case wresp, ok := <-rch:
 			if !ok {
 				logx.Error("etcd monitor chan has been closed")
-				return
+				return false
 			}
 			if wresp.Canceled {
-				logx.Error("etcd monitor chan has been canceled")
-				return
+				logx.Errorf("etcd monitor chan has been canceled, error: %v", wresp.Err())
+				return false
 			}
 			if wresp.Err() != nil {
 				logx.Error(fmt.Sprintf("etcd monitor chan error: %v", wresp.Err()))
-				return
+				return false
 			}
 
 			c.handleWatchEvents(key, wresp.Events)
 		case <-c.done:
-			return
+			return true
 		}
 	}
 }
@@ -288,15 +323,25 @@ func (c *cluster) watchConnState(cli EtcdClient) {
 	watcher.watch(cli.ActiveConnection())
 }
 
+// DialClient dials an etcd cluster with given endpoints.
 func DialClient(endpoints []string) (EtcdClient, error) {
-	return clientv3.New(clientv3.Config{
+	cfg := clientv3.Config{
 		Endpoints:            endpoints,
 		AutoSyncInterval:     autoSyncInterval,
 		DialTimeout:          DialTimeout,
 		DialKeepAliveTime:    dialKeepAliveTime,
 		DialKeepAliveTimeout: DialTimeout,
 		RejectOldCluster:     true,
-	})
+	}
+	if account, ok := GetAccount(endpoints); ok {
+		cfg.Username = account.User
+		cfg.Password = account.Pass
+	}
+	if tlsCfg, ok := GetTLS(endpoints); ok {
+		cfg.TLS = tlsCfg
+	}
+
+	return clientv3.New(cfg)
 }
 
 func getClusterKey(endpoints []string) string {

core/discov/internal/registry_test.go

@@ -7,11 +7,12 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/contextx"
-	"github.com/tal-tech/go-zero/core/logx"
-	"github.com/tal-tech/go-zero/core/stringx"
-	"go.etcd.io/etcd/clientv3"
-	"go.etcd.io/etcd/mvcc/mvccpb"
+	"github.com/zeromicro/go-zero/core/contextx"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/stringx"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 var mockLock sync.Mutex
@@ -32,9 +33,10 @@ func setMockClient(cli EtcdClient) func() {
 }
 
 func TestGetCluster(t *testing.T) {
-	c1 := GetRegistry().getCluster([]string{"first"})
-	c2 := GetRegistry().getCluster([]string{"second"})
-	c3 := GetRegistry().getCluster([]string{"first"})
+	AddAccount([]string{"first"}, "foo", "bar")
+	c1, _ := GetRegistry().getCluster([]string{"first"})
+	c2, _ := GetRegistry().getCluster([]string{"second"})
+	c3, _ := GetRegistry().getCluster([]string{"first"})
 	assert.Equal(t, c1, c3)
 	assert.NotEqual(t, c1, c2)
 }
@@ -202,11 +204,13 @@ func TestClusterWatch_RespFailures(t *testing.T) {
 			restore := setMockClient(cli)
 			defer restore()
 			ch := make(chan clientv3.WatchResponse)
-			cli.EXPECT().Watch(gomock.Any(), "any/", gomock.Any()).Return(ch)
+			cli.EXPECT().Watch(gomock.Any(), "any/", gomock.Any()).Return(ch).AnyTimes()
 			cli.EXPECT().Ctx().Return(context.Background()).AnyTimes()
 			c := new(cluster)
+			c.done = make(chan lang.PlaceholderType)
 			go func() {
 				ch <- resp
+				close(c.done)
 			}()
 			c.watch(cli, "any")
 		})
@@ -220,11 +224,13 @@ func TestClusterWatch_CloseChan(t *testing.T) {
 	restore := setMockClient(cli)
 	defer restore()
 	ch := make(chan clientv3.WatchResponse)
-	cli.EXPECT().Watch(gomock.Any(), "any/", gomock.Any()).Return(ch)
+	cli.EXPECT().Watch(gomock.Any(), "any/", gomock.Any()).Return(ch).AnyTimes()
 	cli.EXPECT().Ctx().Return(context.Background()).AnyTimes()
 	c := new(cluster)
+	c.done = make(chan lang.PlaceholderType)
 	go func() {
 		close(ch)
+		close(c.done)
 	}()
 	c.watch(cli, "any")
 }

core/discov/internal/statewatcher.go

@@ -1,4 +1,5 @@
 //go:generate mockgen -package internal -destination statewatcher_mock.go -source statewatcher.go etcdConn
+
 package internal
 
 import (

core/discov/internal/statewatcher_mock.go

@@ -6,9 +6,10 @@ package internal
 
 import (
 	context "context"
+	reflect "reflect"
+
 	gomock "github.com/golang/mock/gomock"
 	connectivity "google.golang.org/grpc/connectivity"
-	reflect "reflect"
 )
 
 // MocketcdConn is a mock of etcdConn interface

core/discov/internal/updatelistener.go

@@ -1,12 +1,15 @@
 //go:generate mockgen -package internal -destination updatelistener_mock.go -source updatelistener.go UpdateListener
+
 package internal
 
 type (
+	// A KV is used to store an etcd entry with key and value.
 	KV struct {
 		Key string
 		Val string
 	}
 
+	// UpdateListener wraps the OnAdd and OnDelete methods.
 	UpdateListener interface {
 		OnAdd(kv KV)
 		OnDelete(kv KV)

core/discov/internal/updatelistener_mock.go

@@ -5,8 +5,9 @@
 package internal
 
 import (
-	gomock "github.com/golang/mock/gomock"
 	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
 )
 
 // MockUpdateListener is a mock of UpdateListener interface

core/discov/internal/vars.go

@@ -3,17 +3,22 @@ package internal
 import "time"
 
 const (
+	// Delimiter is a separator that separates the etcd path.
+	Delimiter = '/'
+
 	autoSyncInterval   = time.Minute
 	coolDownInterval   = time.Second
 	dialTimeout        = 5 * time.Second
 	dialKeepAliveTime  = 5 * time.Second
 	requestTimeout     = 3 * time.Second
-	Delimiter          = '/'
 	endpointsSeparator = ","
 )
 
 var (
-	DialTimeout    = dialTimeout
+	// DialTimeout is the dial timeout.
+	DialTimeout = dialTimeout
+	// RequestTimeout is the request timeout.
 	RequestTimeout = requestTimeout
-	NewClient      = DialClient
+	// NewClient is used to create etcd clients.
+	NewClient = DialClient
 )

core/discov/kubernetes/etcd-statefulset.yaml

@@ -0,0 +1,64 @@
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: "etcd"
+  namespace: discov
+  labels:
+    app: "etcd"
+spec:
+  serviceName: "etcd"
+  replicas: 5
+  template:
+    metadata:
+      name: "etcd"
+      labels:
+        app: "etcd"
+    spec:
+      volumes:
+      - name: etcd-pvc
+        persistentVolumeClaim:
+          claimName: etcd-pvc
+      containers:
+      - name: "etcd"
+        image: quay.io/coreos/etcd:latest
+        ports:
+        - containerPort: 2379
+          name: client
+        - containerPort: 2380
+          name: peer
+        env:
+        - name: CLUSTER_SIZE
+          value: "5"
+        - name: SET_NAME
+          value: "etcd"
+        - name: VOLNAME
+          valueFrom:
+            fieldRef:
+              apiVersion: v1
+              fieldPath: metadata.name
+        volumeMounts:
+        - name: etcd-pvc
+          mountPath: /var/lib/etcd
+          subPathExpr: $(VOLNAME)  # data mounted respectively in each pod
+        command:
+          - "/bin/sh"
+          - "-ecx"
+          - |
+
+            chmod 700 /var/lib/etcd
+
+            IP=$(hostname -i)
+            PEERS=""
+            for i in $(seq 0 $((${CLUSTER_SIZE} - 1))); do
+                PEERS="${PEERS}${PEERS:+,}${SET_NAME}-${i}=http://${SET_NAME}-${i}.${SET_NAME}:2380"
+            done
+            exec etcd --name ${HOSTNAME} \
+              --listen-peer-urls http://0.0.0.0:2380 \
+              --listen-client-urls http://0.0.0.0:2379 \
+              --advertise-client-urls http://${HOSTNAME}.${SET_NAME}.discov:2379 \
+              --initial-advertise-peer-urls http://${HOSTNAME}.${SET_NAME}:2380 \
+              --initial-cluster ${PEERS} \
+              --initial-cluster-state new \
+              --logger zap \
+              --data-dir /var/lib/etcd \
+              --auto-compaction-retention 1

core/discov/publisher.go

@@ -1,18 +1,20 @@
 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"
-	"github.com/tal-tech/go-zero/core/proc"
-	"github.com/tal-tech/go-zero/core/syncx"
-	"github.com/tal-tech/go-zero/core/threading"
-	"go.etcd.io/etcd/clientv3"
+	"github.com/zeromicro/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/proc"
+	"github.com/zeromicro/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/threading"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 type (
-	PublisherOption func(client *Publisher)
+	// PubOption defines the method to customize a Publisher.
+	PubOption func(client *Publisher)
 
+	// A Publisher can be used to publish the value to an etcd cluster on the given key.
 	Publisher struct {
 		endpoints  []string
 		key        string
@@ -26,7 +28,11 @@ type (
 	}
 )
 
-func NewPublisher(endpoints []string, key, value string, opts ...PublisherOption) *Publisher {
+// NewPublisher returns a Publisher.
+// endpoints is the hosts of the etcd cluster.
+// key:value are a pair to be published.
+// opts are used to customize the Publisher.
+func NewPublisher(endpoints []string, key, value string, opts ...PubOption) *Publisher {
 	publisher := &Publisher{
 		endpoints:  endpoints,
 		key:        key,
@@ -43,6 +49,7 @@ func NewPublisher(endpoints []string, key, value string, opts ...PublisherOption
 	return publisher
 }
 
+// KeepAlive keeps key:value alive.
 func (p *Publisher) KeepAlive() error {
 	cli, err := internal.GetRegistry().GetConn(p.endpoints)
 	if err != nil {
@@ -61,14 +68,17 @@ func (p *Publisher) KeepAlive() error {
 	return p.keepAliveAsync(cli)
 }
 
+// Pause pauses the renewing of key:value.
 func (p *Publisher) Pause() {
 	p.pauseChan <- lang.Placeholder
 }
 
+// Resume resumes the renewing of key:value.
 func (p *Publisher) Resume() {
 	p.resumeChan <- lang.Placeholder
 }
 
+// Stop stops the renewing and revokes the registration.
 func (p *Publisher) Stop() {
 	p.quit.Close()
 }
@@ -135,8 +145,23 @@ func (p *Publisher) revoke(cli internal.EtcdClient) {
 	}
 }
 
-func WithId(id int64) PublisherOption {
+// WithId customizes a Publisher with the id.
+func WithId(id int64) PubOption {
 	return func(publisher *Publisher) {
 		publisher.id = id
 	}
 }
+
+// WithPubEtcdAccount provides the etcd username/password.
+func WithPubEtcdAccount(user, pass string) PubOption {
+	return func(pub *Publisher) {
+		RegisterAccount(pub.endpoints, user, pass)
+	}
+}
+
+// WithPubEtcdTLS provides the etcd CertFile/CertKeyFile/CACertFile.
+func WithPubEtcdTLS(certFile, certKeyFile, caFile string, insecureSkipVerify bool) PubOption {
+	return func(pub *Publisher) {
+		logx.Must(RegisterTLS(pub.endpoints, certFile, certKeyFile, caFile, insecureSkipVerify))
+	}
+}

core/discov/publisher_test.go

@@ -4,12 +4,15 @@ import (
 	"errors"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/discov/internal"
-	"github.com/tal-tech/go-zero/core/logx"
-	"go.etcd.io/etcd/clientv3"
+	"github.com/zeromicro/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/stringx"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 func init() {
@@ -28,7 +31,8 @@ func TestPublisher_register(t *testing.T) {
 		ID: id,
 	}, nil)
 	cli.EXPECT().Put(gomock.Any(), makeEtcdKey("thekey", id), "thevalue", gomock.Any())
-	pub := NewPublisher(nil, "thekey", "thevalue")
+	pub := NewPublisher(nil, "thekey", "thevalue",
+		WithPubEtcdAccount(stringx.Rand(), "bar"))
 	_, err := pub.register(cli)
 	assert.Nil(t, err)
 }
@@ -152,3 +156,16 @@ func TestPublisher_keepAliveAsyncPause(t *testing.T) {
 	pub.Pause()
 	wg.Wait()
 }
+
+func TestPublisher_Resume(t *testing.T) {
+	publisher := new(Publisher)
+	publisher.resumeChan = make(chan lang.PlaceholderType)
+	go func() {
+		publisher.Resume()
+	}()
+	go func() {
+		time.Sleep(time.Minute)
+		t.Fail()
+	}()
+	<-publisher.resumeChan
+}

core/discov/subscriber.go

@@ -4,31 +4,36 @@ import (
 	"sync"
 	"sync/atomic"
 
-	"github.com/tal-tech/go-zero/core/discov/internal"
-	"github.com/tal-tech/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/syncx"
 )
 
 type (
-	subOptions struct {
-		exclusive bool
-	}
-
-	SubOption func(opts *subOptions)
+	// SubOption defines the method to customize a Subscriber.
+	SubOption func(sub *Subscriber)
 
+	// A Subscriber is used to subscribe the given key on a etcd cluster.
 	Subscriber struct {
-		items *container
+		endpoints []string
+		exclusive bool
+		items     *container
 	}
 )
 
+// NewSubscriber returns a Subscriber.
+// endpoints is the hosts of the etcd cluster.
+// key is the key to subscribe.
+// opts are used to customize the Subscriber.
 func NewSubscriber(endpoints []string, key string, opts ...SubOption) (*Subscriber, error) {
-	var subOpts subOptions
-	for _, opt := range opts {
-		opt(&subOpts)
-	}
-
 	sub := &Subscriber{
-		items: newContainer(subOpts.exclusive),
+		endpoints: endpoints,
 	}
+	for _, opt := range opts {
+		opt(sub)
+	}
+	sub.items = newContainer(sub.exclusive)
+
 	if err := internal.GetRegistry().Monitor(endpoints, key, sub.items); err != nil {
 		return nil, err
 	}
@@ -36,19 +41,35 @@ func NewSubscriber(endpoints []string, key string, opts ...SubOption) (*Subscrib
 	return sub, nil
 }
 
+// AddListener adds listener to s.
 func (s *Subscriber) AddListener(listener func()) {
 	s.items.addListener(listener)
 }
 
+// Values returns all the subscription values.
 func (s *Subscriber) Values() []string {
 	return s.items.getValues()
 }
 
-// exclusive means that key value can only be 1:1,
+// Exclusive means that key value can only be 1:1,
 // which means later added value will remove the keys associated with the same value previously.
 func Exclusive() SubOption {
-	return func(opts *subOptions) {
-		opts.exclusive = true
+	return func(sub *Subscriber) {
+		sub.exclusive = true
+	}
+}
+
+// WithSubEtcdAccount provides the etcd username/password.
+func WithSubEtcdAccount(user, pass string) SubOption {
+	return func(sub *Subscriber) {
+		RegisterAccount(sub.endpoints, user, pass)
+	}
+}
+
+// WithSubEtcdTLS provides the etcd CertFile/CertKeyFile/CACertFile.
+func WithSubEtcdTLS(certFile, certKeyFile, caFile string, insecureSkipVerify bool) SubOption {
+	return func(sub *Subscriber) {
+		logx.Must(RegisterTLS(sub.endpoints, certFile, certKeyFile, caFile, insecureSkipVerify))
 	}
 }
 
@@ -100,9 +121,9 @@ func (c *container) addKv(key, value string) ([]string, bool) {
 
 	if early {
 		return previous, true
-	} else {
-		return nil, false
 	}
+
+	return nil, false
 }
 
 func (c *container) addListener(listener func()) {

core/discov/subscriber_test.go

@@ -1,10 +1,12 @@
 package discov
 
 import (
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/discov/internal"
+	"github.com/zeromicro/go-zero/core/stringx"
 )
 
 const (
@@ -198,3 +200,28 @@ func TestContainer(t *testing.T) {
 		}
 	}
 }
+
+func TestSubscriber(t *testing.T) {
+	sub := new(Subscriber)
+	Exclusive()(sub)
+	sub.items = newContainer(sub.exclusive)
+	var count int32
+	sub.AddListener(func() {
+		atomic.AddInt32(&count, 1)
+	})
+	sub.items.notifyChange()
+	assert.Empty(t, sub.Values())
+	assert.Equal(t, int32(1), atomic.LoadInt32(&count))
+}
+
+func TestWithSubEtcdAccount(t *testing.T) {
+	endpoints := []string{"localhost:2379"}
+	user := stringx.Rand()
+	WithSubEtcdAccount(user, "bar")(&Subscriber{
+		endpoints: endpoints,
+	})
+	account, ok := internal.GetAccount(endpoints)
+	assert.True(t, ok)
+	assert.Equal(t, user, account.User)
+	assert.Equal(t, "bar", account.Pass)
+}

core/errorx/atomicerror.go

@@ -2,14 +2,19 @@ package errorx
 
 import "sync/atomic"
 
+// AtomicError defines an atomic error.
 type AtomicError struct {
 	err atomic.Value // error
 }
 
+// Set sets the error.
 func (ae *AtomicError) Set(err error) {
-	ae.err.Store(err)
+	if err != nil {
+		ae.err.Store(err)
+	}
 }
 
+// Load returns the error.
 func (ae *AtomicError) Load() error {
 	if v := ae.err.Load(); v != nil {
 		return v.(error)

core/errorx/atomicerror_test.go

@@ -17,6 +17,15 @@ func TestAtomicError(t *testing.T) {
 	assert.Equal(t, errDummy, err.Load())
 }
 
+func TestAtomicErrorSetNil(t *testing.T) {
+	var (
+		errNil error
+		err    AtomicError
+	)
+	err.Set(errNil)
+	assert.Equal(t, errNil, err.Load())
+}
+
 func TestAtomicErrorNil(t *testing.T) {
 	var err AtomicError
 	assert.Nil(t, err.Load())

core/errorx/batcherror.go

@@ -3,6 +3,7 @@ package errorx
 import "bytes"
 
 type (
+	// A BatchError is an error that can hold multiple errors.
 	BatchError struct {
 		errs errorArray
 	}
@@ -10,12 +11,16 @@ type (
 	errorArray []error
 )
 
-func (be *BatchError) Add(err error) {
-	if err != nil {
-		be.errs = append(be.errs, err)
+// Add adds errs to be, nil errors are ignored.
+func (be *BatchError) Add(errs ...error) {
+	for _, err := range errs {
+		if err != nil {
+			be.errs = append(be.errs, err)
+		}
 	}
 }
 
+// Err returns an error that represents all errors.
 func (be *BatchError) Err() error {
 	switch len(be.errs) {
 	case 0:
@@ -27,10 +32,12 @@ func (be *BatchError) Err() error {
 	}
 }
 
+// NotNil checks if any error inside.
 func (be *BatchError) NotNil() bool {
 	return len(be.errs) > 0
 }
 
+// Error returns a string that represents inside errors.
 func (ea errorArray) Error() string {
 	var buf bytes.Buffer
 

core/errorx/callchain.go

@@ -1,5 +1,6 @@
 package errorx
 
+// Chain runs funs one by one until an error occurred.
 func Chain(fns ...func() error) error {
 	for _, fn := range fns {
 		if err := fn(); err != nil {

core/errorx/callchain_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestChain(t *testing.T) {
-	var errDummy = errors.New("dummy")
+	errDummy := errors.New("dummy")
 	assert.Nil(t, Chain(func() error {
 		return nil
 	}, func() error {

core/executors/bulkexecutor.go

@@ -5,8 +5,12 @@ import "time"
 const defaultBulkTasks = 1000
 
 type (
+	// BulkOption defines the method to customize a BulkExecutor.
 	BulkOption func(options *bulkOptions)
 
+	// A BulkExecutor is an executor that can execute tasks on either requirement meets:
+	// 1. up to given size of tasks
+	// 2. flush interval time elapsed
 	BulkExecutor struct {
 		executor  *PeriodicalExecutor
 		container *bulkContainer
@@ -18,6 +22,7 @@ type (
 	}
 )
 
+// NewBulkExecutor returns a BulkExecutor.
 func NewBulkExecutor(execute Execute, opts ...BulkOption) *BulkExecutor {
 	options := newBulkOptions()
 	for _, opt := range opts {
@@ -36,25 +41,30 @@ func NewBulkExecutor(execute Execute, opts ...BulkOption) *BulkExecutor {
 	return executor
 }
 
+// Add adds task into be.
 func (be *BulkExecutor) Add(task interface{}) error {
 	be.executor.Add(task)
 	return nil
 }
 
+// Flush forces be to flush and execute tasks.
 func (be *BulkExecutor) Flush() {
 	be.executor.Flush()
 }
 
+// Wait waits be to done with the task execution.
 func (be *BulkExecutor) Wait() {
 	be.executor.Wait()
 }
 
+// WithBulkTasks customizes a BulkExecutor with given tasks limit.
 func WithBulkTasks(tasks int) BulkOption {
 	return func(options *bulkOptions) {
 		options.cachedTasks = tasks
 	}
 }
 
+// WithBulkInterval customizes a BulkExecutor with given flush interval.
 func WithBulkInterval(duration time.Duration) BulkOption {
 	return func(options *bulkOptions) {
 		options.flushInterval = duration

core/executors/chunkexecutor.go

@@ -5,8 +5,12 @@ import "time"
 const defaultChunkSize = 1024 * 1024 // 1M
 
 type (
+	// ChunkOption defines the method to customize a ChunkExecutor.
 	ChunkOption func(options *chunkOptions)
 
+	// A ChunkExecutor is an executor to execute tasks when either requirement meets:
+	// 1. up to given chunk size
+	// 2. flush interval elapsed
 	ChunkExecutor struct {
 		executor  *PeriodicalExecutor
 		container *chunkContainer
@@ -18,6 +22,7 @@ type (
 	}
 )
 
+// NewChunkExecutor returns a ChunkExecutor.
 func NewChunkExecutor(execute Execute, opts ...ChunkOption) *ChunkExecutor {
 	options := newChunkOptions()
 	for _, opt := range opts {
@@ -36,6 +41,7 @@ func NewChunkExecutor(execute Execute, opts ...ChunkOption) *ChunkExecutor {
 	return executor
 }
 
+// Add adds task with given chunk size into ce.
 func (ce *ChunkExecutor) Add(task interface{}, size int) error {
 	ce.executor.Add(chunk{
 		val:  task,
@@ -44,20 +50,24 @@ func (ce *ChunkExecutor) Add(task interface{}, size int) error {
 	return nil
 }
 
+// Flush forces ce to flush and execute tasks.
 func (ce *ChunkExecutor) Flush() {
 	ce.executor.Flush()
 }
 
+// Wait waits the execution to be done.
 func (ce *ChunkExecutor) Wait() {
 	ce.executor.Wait()
 }
 
+// WithChunkBytes customizes a ChunkExecutor with the given chunk size.
 func WithChunkBytes(size int) ChunkOption {
 	return func(options *chunkOptions) {
 		options.chunkSize = size
 	}
 }
 
+// WithFlushInterval customizes a ChunkExecutor with the given flush interval.
 func WithFlushInterval(duration time.Duration) ChunkOption {
 	return func(options *chunkOptions) {
 		options.flushInterval = duration

core/executors/delayexecutor.go

@@ -4,9 +4,10 @@ import (
 	"sync"
 	"time"
 
-	"github.com/tal-tech/go-zero/core/threading"
+	"github.com/zeromicro/go-zero/core/threading"
 )
 
+// A DelayExecutor delays a tasks on given delay interval.
 type DelayExecutor struct {
 	fn        func()
 	delay     time.Duration
@@ -14,6 +15,7 @@ type DelayExecutor struct {
 	lock      sync.Mutex
 }
 
+// NewDelayExecutor returns a DelayExecutor with given fn and delay.
 func NewDelayExecutor(fn func(), delay time.Duration) *DelayExecutor {
 	return &DelayExecutor{
 		fn:    fn,
@@ -21,6 +23,7 @@ func NewDelayExecutor(fn func(), delay time.Duration) *DelayExecutor {
 	}
 }
 
+// Trigger triggers the task to be executed after given delay, safe to trigger more than once.
 func (de *DelayExecutor) Trigger() {
 	de.lock.Lock()
 	defer de.lock.Unlock()

core/executors/lessexecutor.go

@@ -3,15 +3,17 @@ package executors
 import (
 	"time"
 
-	"github.com/tal-tech/go-zero/core/syncx"
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
+// A LessExecutor is an executor to limit execution once within given time interval.
 type LessExecutor struct {
 	threshold time.Duration
 	lastTime  *syncx.AtomicDuration
 }
 
+// NewLessExecutor returns a LessExecutor with given threshold as time interval.
 func NewLessExecutor(threshold time.Duration) *LessExecutor {
 	return &LessExecutor{
 		threshold: threshold,
@@ -19,6 +21,8 @@ func NewLessExecutor(threshold time.Duration) *LessExecutor {
 	}
 }
 
+// DoOrDiscard executes or discards the task depends on if
+// another task was executed within the time interval.
 func (le *LessExecutor) DoOrDiscard(execute func()) bool {
 	now := timex.Now()
 	lastTime := le.lastTime.Load()

core/executors/lessexecutor_test.go

@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 func TestLessExecutor_DoOrDiscard(t *testing.T) {

core/executors/periodicalexecutor.go

@@ -6,17 +6,17 @@ import (
 	"sync/atomic"
 	"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"
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/proc"
+	"github.com/zeromicro/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/threading"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 const idleRound = 10
 
 type (
-	// A type that satisfies executors.TaskContainer can be used as the underlying
+	// TaskContainer interface defines a type that can be used as the underlying
 	// container that used to do periodical executions.
 	TaskContainer interface {
 		// AddTask adds the task into the container.
@@ -28,6 +28,7 @@ type (
 		RemoveAll() interface{}
 	}
 
+	// A PeriodicalExecutor is an executor that periodically execute tasks.
 	PeriodicalExecutor struct {
 		commander chan interface{}
 		interval  time.Duration
@@ -43,6 +44,7 @@ type (
 	}
 )
 
+// NewPeriodicalExecutor returns a PeriodicalExecutor with given interval and container.
 func NewPeriodicalExecutor(interval time.Duration, container TaskContainer) *PeriodicalExecutor {
 	executor := &PeriodicalExecutor{
 		// buffer 1 to let the caller go quickly
@@ -51,7 +53,7 @@ func NewPeriodicalExecutor(interval time.Duration, container TaskContainer) *Per
 		container:   container,
 		confirmChan: make(chan lang.PlaceholderType),
 		newTicker: func(d time.Duration) timex.Ticker {
-			return timex.NewTicker(interval)
+			return timex.NewTicker(d)
 		},
 	}
 	proc.AddShutdownListener(func() {
@@ -61,6 +63,7 @@ func NewPeriodicalExecutor(interval time.Duration, container TaskContainer) *Per
 	return executor
 }
 
+// Add adds tasks into pe.
 func (pe *PeriodicalExecutor) Add(task interface{}) {
 	if vals, ok := pe.addAndCheck(task); ok {
 		pe.commander <- vals
@@ -68,6 +71,7 @@ func (pe *PeriodicalExecutor) Add(task interface{}) {
 	}
 }
 
+// Flush forces pe to execute tasks.
 func (pe *PeriodicalExecutor) Flush() bool {
 	pe.enterExecution()
 	return pe.executeTasks(func() interface{} {
@@ -77,12 +81,14 @@ func (pe *PeriodicalExecutor) Flush() bool {
 	}())
 }
 
+// Sync lets caller to run fn thread-safe with pe, especially for the underlying container.
 func (pe *PeriodicalExecutor) Sync(fn func()) {
 	pe.lock.Lock()
 	defer pe.lock.Unlock()
 	fn()
 }
 
+// Wait waits the execution to be done.
 func (pe *PeriodicalExecutor) Wait() {
 	pe.Flush()
 	pe.wgBarrier.Guard(func() {

core/executors/periodicalexecutor_test.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/timex"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 const threshold = 10

core/executors/vars.go

@@ -4,4 +4,5 @@ import "time"
 
 const defaultFlushInterval = time.Second
 
+// Execute defines the method to execute tasks.
 type Execute func(tasks []interface{})

core/filex/file.go

@@ -7,6 +7,7 @@ import (
 
 const bufSize = 1024
 
+// FirstLine returns the first line of the file.
 func FirstLine(filename string) (string, error) {
 	file, err := os.Open(filename)
 	if err != nil {
@@ -17,6 +18,7 @@ func FirstLine(filename string) (string, error) {
 	return firstLine(file)
 }
 
+// LastLine returns the last line of the file.
 func LastLine(filename string) (string, error) {
 	file, err := os.Open(filename)
 	if err != nil {
@@ -69,11 +71,11 @@ func lastLine(filename string, file *os.File) (string, error) {
 
 		if buf[n-1] == '\n' {
 			buf = buf[:n-1]
-			n -= 1
+			n--
 		} else {
 			buf = buf[:n]
 		}
-		for n -= 1; n >= 0; n-- {
+		for n--; n >= 0; n-- {
 			if buf[n] == '\n' {
 				return string(append(buf[n+1:], last...)), nil
 			}

core/filex/file_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/fs"
+	"github.com/zeromicro/go-zero/core/fs"
 )
 
 const (

core/filex/lookup.go

@@ -5,12 +5,15 @@ import (
 	"os"
 )
 
+// OffsetRange represents a content block of a file.
 type OffsetRange struct {
 	File  string
 	Start int64
 	Stop  int64
 }
 
+// SplitLineChunks splits file into chunks.
+// The whole line are guaranteed to be split in the same chunk.
 func SplitLineChunks(filename string, chunks int) ([]OffsetRange, error) {
 	info, err := os.Stat(filename)
 	if err != nil {

core/filex/lookup_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/fs"
+	"github.com/zeromicro/go-zero/core/fs"
 )
 
 func TestSplitLineChunks(t *testing.T) {

core/filex/progressscanner.go

@@ -3,8 +3,11 @@ package filex
 import "gopkg.in/cheggaaa/pb.v1"
 
 type (
+	// A Scanner is used to read lines.
 	Scanner interface {
+		// Scan checks if has remaining to read.
 		Scan() bool
+		// Text returns next line.
 		Text() string
 	}
 
@@ -14,6 +17,7 @@ type (
 	}
 )
 
+// NewProgressScanner returns a Scanner with progress indicator.
 func NewProgressScanner(scanner Scanner, bar *pb.ProgressBar) Scanner {
 	return &progressScanner{
 		Scanner: scanner,

core/filex/rangereader.go

@@ -5,12 +5,17 @@ import (
 	"os"
 )
 
+// errExceedFileSize indicates that the file size is exceeded.
+var errExceedFileSize = errors.New("exceed file size")
+
+// A RangeReader is used to read a range of content from a file.
 type RangeReader struct {
 	file  *os.File
 	start int64
 	stop  int64
 }
 
+// NewRangeReader returns a RangeReader, which will read the range of content from file.
 func NewRangeReader(file *os.File, start, stop int64) *RangeReader {
 	return &RangeReader{
 		file:  file,
@@ -19,6 +24,7 @@ func NewRangeReader(file *os.File, start, stop int64) *RangeReader {
 	}
 }
 
+// Read reads the range of content into p.
 func (rr *RangeReader) Read(p []byte) (n int, err error) {
 	stat, err := rr.file.Stat()
 	if err != nil {
@@ -26,7 +32,7 @@ func (rr *RangeReader) Read(p []byte) (n int, err error) {
 	}
 
 	if rr.stop < rr.start || rr.start >= stat.Size() {
-		return 0, errors.New("exceed file size")
+		return 0, errExceedFileSize
 	}
 
 	if rr.stop-rr.start < int64(len(p)) {

core/filex/rangereader_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tal-tech/go-zero/core/fs"
+	"github.com/zeromicro/go-zero/core/fs"
 )
 
 func TestRangeReader(t *testing.T) {

core/fs/files+polyfill.go

@@ -1,3 +1,4 @@
+//go:build windows
 // +build windows
 
 package fs

core/fs/files.go

@@ -1,3 +1,4 @@
+//go:build linux || darwin
 // +build linux darwin
 
 package fs
@@ -7,6 +8,7 @@ import (
 	"syscall"
 )
 
+// CloseOnExec makes sure closing the file on process forking.
 func CloseOnExec(file *os.File) {
 	if file != nil {
 		syscall.CloseOnExec(int(file.Fd()))