8 Commits

Author SHA1 Message Date
72f3f19b2f [v3] update ci (#208)
All checks were successful
coverage / build (push) Successful in 1m16s
test / test (push) Successful in 3m38s
* rename .github to .github.old

* update readme

* rename .gitea to .github

* update ci

* fix linters
2025-05-05 18:01:26 +03:00
f17bcbf046 Merge pull request #206 from unistack-org/dependabot/go_modules/go.unistack.org/micro/v3-3.11.43
All checks were successful
test / test (push) Successful in 3m40s
Bump go.unistack.org/micro/v3 from 3.11.37 to 3.11.43
2025-04-29 03:08:29 +03:00
dependabot[bot]
8af0ad942d Bump go.unistack.org/micro/v3 from 3.11.37 to 3.11.43
Bumps go.unistack.org/micro/v3 from 3.11.37 to 3.11.43.

---
updated-dependencies:
- dependency-name: go.unistack.org/micro/v3
  dependency-version: 3.11.43
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-04-29 00:08:09 +00:00
cd84ab56af update for latest micro
All checks were successful
test / test (push) Successful in 1m58s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-01-15 13:06:43 +03:00
4470cb69a2 update for latest micro
All checks were successful
test / test (push) Successful in 2m40s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-01-15 12:17:26 +03:00
027268df75 add workflow
All checks were successful
test / test (push) Successful in 13m32s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-12-12 09:37:18 +03:00
1544956100 update go.mod
Some checks failed
build / test (push) Failing after 25s
build / lint (push) Successful in 22s
codeql / analyze (go) (push) Failing after 1m13s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-12-12 09:36:12 +03:00
db00f3bb91 add waitGroups for waiting finish all connects (#131)
closes #130

Reviewed-on: #131
Co-authored-by: Evstigneev Denis <danteevstigneev@yandex.ru>
Co-committed-by: Evstigneev Denis <danteevstigneev@yandex.ru>
2024-03-13 15:36:34 +03:00
29 changed files with 441 additions and 155 deletions

View File

@@ -0,0 +1,24 @@
---
name: Bug report
about: For reporting bugs in go-micro
title: "[BUG]"
labels: ''
assignees: ''
---
**Describe the bug**
1. What are you trying to do?
2. What did you expect to happen?
3. What happens instead?
**How to reproduce the bug:**
If possible, please include a minimal code snippet here.
**Environment:**
Go Version: please paste `go version` output here
```
please paste `go env` output here
```

View File

@@ -0,0 +1,17 @@
---
name: Feature request / Enhancement
about: If you have a need not served by go-micro
title: "[FEATURE]"
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.
**Additional context**
Add any other context or screenshots about the feature request here.

View File

@@ -0,0 +1,14 @@
---
name: Question
about: Ask a question about go-micro
title: ''
labels: ''
assignees: ''
---
Before asking, please check if your question has already been answered:
1. Check the documentation - https://micro.mu/docs/
2. Check the examples and plugins - https://github.com/micro/examples & https://github.com/micro/go-plugins
3. Search existing issues

View File

@@ -0,0 +1,9 @@
## Pull Request template
Please, go through these steps before clicking submit on this PR.
1. Give a descriptive title to your PR.
2. Provide a description of your changes.
3. Make sure you have some relevant tests.
4. Put `closes #XXXX` in your comment to auto-close the issue that your PR fixes (if applicable).
**PLEASE REMOVE THIS TEMPLATE BEFORE SUBMITTING**

View File

@@ -1,6 +1,6 @@
---
name: Bug report
about: For reporting bugs in go-micro
about: For reporting bugs in micro
title: "[BUG]"
labels: ''
assignees: ''
@@ -16,9 +16,3 @@ assignees: ''
**How to reproduce the bug:**
If possible, please include a minimal code snippet here.
**Environment:**
Go Version: please paste `go version` output here
```
please paste `go env` output here
```

View File

@@ -1,6 +1,6 @@
---
name: Feature request / Enhancement
about: If you have a need not served by go-micro
about: If you have a need not served by micro
title: "[FEATURE]"
labels: ''
assignees: ''
@@ -14,4 +14,4 @@ A clear and concise description of what the problem is. Ex. I'm always frustrate
A clear and concise description of what you want to happen.
**Additional context**
Add any other context or screenshots about the feature request here.
Add any other context or screenshots about the feature request here.

View File

@@ -1,14 +1,8 @@
---
name: Question
about: Ask a question about go-micro
about: Ask a question about micro
title: ''
labels: ''
assignees: ''
---
Before asking, please check if your question has already been answered:
1. Check the documentation - https://micro.mu/docs/
2. Check the examples and plugins - https://github.com/micro/examples & https://github.com/micro/go-plugins
3. Search existing issues
---

28
.github/autoapprove.yml vendored Normal file
View File

@@ -0,0 +1,28 @@
name: "autoapprove"
on:
pull_request_target:
types: [assigned, opened, synchronize, reopened]
workflow_run:
workflows: ["prbuild"]
types:
- completed
permissions:
pull-requests: write
contents: write
jobs:
autoapprove:
runs-on: ubuntu-latest
steps:
- name: approve
run: [ "curl -o tea https://dl.gitea.com/tea/main/tea-main-linux-amd64",
"chmod +x ./tea",
"./tea login add --name unistack --token ${{ secrets.GITHUB_TOKEN }} --url https://git.unistack.org",
"./tea pr --repo ${{ github.event.repository.name }}"
]
if: github.actor == 'vtolstov'
id: approve
with:
github-token: ${{ secrets.GITHUB_TOKEN }}

53
.github/workflows/job_coverage.yml vendored Normal file
View File

@@ -0,0 +1,53 @@
name: coverage
on:
push:
branches: [ main, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
pull_request:
branches: [ main, v3, v4 ]
jobs:
build:
if: github.server_url != 'https://github.com'
runs-on: ubuntu-latest
steps:
- name: checkout code
uses: actions/checkout@v4
with:
filter: 'blob:none'
- name: setup go
uses: actions/setup-go@v5
with:
cache-dependency-path: "**/*.sum"
go-version: 'stable'
- name: test coverage
run: |
go test -v -cover ./... -covermode=count -coverprofile coverage.out -coverpkg ./...
go tool cover -func coverage.out -o coverage.out
- name: coverage badge
uses: tj-actions/coverage-badge-go@v2
with:
green: 80
filename: coverage.out
- uses: stefanzweifel/git-auto-commit-action@v4
name: autocommit
with:
commit_message: Apply Code Coverage Badge
skip_fetch: false
skip_checkout: false
file_pattern: ./README.md
- name: push
if: steps.auto-commit-action.outputs.changes_detected == 'true'
uses: ad-m/github-push-action@master
with:
github_token: ${{ github.token }}
branch: ${{ github.ref }}

29
.github/workflows/job_lint.yml vendored Normal file
View File

@@ -0,0 +1,29 @@
name: lint
on:
pull_request:
types: [opened, reopened, synchronize]
branches: [ master, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
jobs:
lint:
runs-on: ubuntu-latest
steps:
- name: checkout code
uses: actions/checkout@v4
with:
filter: 'blob:none'
- name: setup go
uses: actions/setup-go@v5
with:
cache-dependency-path: "**/*.sum"
go-version: 'stable'
- name: setup deps
run: go get -v ./...
- name: run lint
uses: golangci/golangci-lint-action@v6
with:
version: 'latest'

31
.github/workflows/job_test.yml vendored Normal file
View File

@@ -0,0 +1,31 @@
name: test
on:
pull_request:
types: [opened, reopened, synchronize]
branches: [ master, v3, v4 ]
push:
branches: [ master, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: checkout code
uses: actions/checkout@v4
with:
filter: 'blob:none'
- name: setup go
uses: actions/setup-go@v5
with:
cache-dependency-path: "**/*.sum"
go-version: 'stable'
- name: setup deps
run: go get -v ./...
- name: run test
env:
INTEGRATION_TESTS: yes
run: go test -mod readonly -v ./...

50
.github/workflows/job_tests.yml vendored Normal file
View File

@@ -0,0 +1,50 @@
name: test
on:
pull_request:
types: [opened, reopened, synchronize]
branches: [ master, v3, v4 ]
push:
branches: [ master, v3, v4 ]
paths-ignore:
- '.github/**'
- '.gitea/**'
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: checkout code
uses: actions/checkout@v4
with:
filter: 'blob:none'
- name: checkout tests
uses: actions/checkout@v4
with:
ref: master
filter: 'blob:none'
repository: unistack-org/micro-tests
path: micro-tests
- name: setup go
uses: actions/setup-go@v5
with:
cache-dependency-path: "**/*.sum"
go-version: 'stable'
- name: setup go work
env:
GOWORK: ${{ github.workspace }}/go.work
run: |
go work init
go work use .
go work use micro-tests
- name: setup deps
env:
GOWORK: ${{ github.workspace }}/go.work
run: go get -v ./...
- name: run tests
env:
INTEGRATION_TESTS: yes
GOWORK: ${{ github.workspace }}/go.work
run: |
cd micro-tests
go test -mod readonly -v ./... || true

24
.gitignore vendored Normal file
View File

@@ -0,0 +1,24 @@
# Binaries for programs and plugins
*.exe
*.exe~
*.dll
*.so
*.dylib
bin
# Test binary, built with `go test -c`
*.test
# Output of the go coverage tool, specifically when used with LiteIDE
*.out
# Dependency directories (remove the comment below to include it)
# vendor/
# Go workspace file
go.work
# General
.DS_Store
.idea
.vscode

View File

@@ -1,27 +1,18 @@
run:
concurrency: 4
deadline: 5m
timeout: 5m
issues-exit-code: 1
tests: true
linters-settings:
govet:
check-shadowing: true
enable:
- fieldalignment
linters:
enable:
- govet
- deadcode
- errcheck
- govet
- ineffassign
- staticcheck
- structcheck
- typecheck
- unused
- varcheck
- bodyclose
- gci
- goconst
@@ -30,7 +21,6 @@ linters:
- gofmt
- gofumpt
- goimports
- golint
- gosec
- makezero
- misspell

View File

@@ -1,7 +1,7 @@
# TCP Server
![Coverage](https://img.shields.io/badge/Coverage-0.0%25-red)
The TCP micro.Server implemtation. It's a partial implementation which strips out codecs, transports, etc but enables you
The TCP micro.Server implemtation. It's a partial implementation which strips out codecs, transports, etc but enables you
to create a TCP Server that could potentially be used for some TCP based API services.
## Usage

21
go.mod
View File

@@ -1,8 +1,23 @@
module go.unistack.org/micro-server-tcp/v3
go 1.16
go 1.22.0
toolchain go1.23.4
require (
go.unistack.org/micro/v3 v3.10.14
golang.org/x/net v0.0.0-20220225172249-27dd8689420f
go.unistack.org/micro/v3 v3.11.43
golang.org/x/net v0.34.0
)
require (
github.com/ash3in/uuidv8 v1.2.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/kr/pretty v0.3.1 // indirect
github.com/matoous/go-nanoid v1.5.1 // indirect
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
golang.org/x/sys v0.29.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20250106144421-5f5ef82da422 // indirect
google.golang.org/grpc v1.69.4 // indirect
google.golang.org/protobuf v1.36.2 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)

53
go.sum
View File

@@ -1,15 +1,42 @@
github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
go.unistack.org/micro/v3 v3.10.14 h1:7fgLpwGlCN67twhwtngJDEQvrMkUBDSA5vzZqxIDqNE=
go.unistack.org/micro/v3 v3.10.14/go.mod h1:uMAc0U/x7dmtICCrblGf0ZLgYegu3VwQAquu+OFCw1Q=
golang.org/x/net v0.0.0-20220225172249-27dd8689420f h1:oA4XRj0qtSt8Yo1Zms0CUlsT3KG69V2UGQWPBxujDmc=
golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
github.com/ash3in/uuidv8 v1.2.0 h1:2oogGdtCPwaVtyvPPGin4TfZLtOGE5F+W++E880G6SI=
github.com/ash3in/uuidv8 v1.2.0/go.mod h1:BnU0wJBxnzdEKmVg4xckBkD+VZuecTFTUP3M0dWgyY4=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/matoous/go-nanoid v1.5.1 h1:aCjdvTyO9LLnTIi0fgdXhOPPvOHjpXN6Ik9DaNjIct4=
github.com/matoous/go-nanoid v1.5.1/go.mod h1:zyD2a71IubI24efhpvkJz+ZwfwagzgSO6UNiFsZKN7U=
github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
go.unistack.org/micro/v3 v3.11.43 h1:+zIt+NjMhHHNbOpllQVilz96UWV5qsSGV+JFgTC4xFM=
go.unistack.org/micro/v3 v3.11.43/go.mod h1:POGU5hstnAT9LH70m8FalyQSNi2GfIew71K75JenIZk=
golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250106144421-5f5ef82da422 h1:3UsHvIr4Wc2aW4brOaSCmcxh9ksica6fHEr8P1XhkYw=
google.golang.org/genproto/googleapis/rpc v0.0.0-20250106144421-5f5ef82da422/go.mod h1:3ENsm/5D1mzDyhpzeRi1NR784I0BcofWBoSc5QqqMK4=
google.golang.org/grpc v1.69.4 h1:MF5TftSMkd8GLw/m0KM6V8CMOCY6NZ1NQDPGFgbTt4A=
google.golang.org/grpc v1.69.4/go.mod h1:vyjdE6jLBI76dgpDojsFGNaHlxdjXN9ghpnd2o7JGZ4=
google.golang.org/protobuf v1.36.2 h1:R8FeyR1/eLmkutZOM5CWghmo5itiG9z0ktFlTVLuTmU=
google.golang.org/protobuf v1.36.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

View File

@@ -3,7 +3,6 @@ package tcp
import (
"net"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/server"
)
@@ -14,7 +13,6 @@ type Handler interface {
type tcpHandler struct {
opts server.HandlerOptions
hd interface{}
eps []*register.Endpoint
maxMsgSize int
}
@@ -26,10 +24,6 @@ func (h *tcpHandler) Handler() interface{} {
return h.hd
}
func (h *tcpHandler) Endpoints() []*register.Endpoint {
return h.eps
}
func (h *tcpHandler) Options() server.HandlerOptions {
return h.opts
}

View File

@@ -14,7 +14,7 @@ type tcpMessage struct {
header metadata.Metadata
topic string
contentType string
body []byte
body []byte //nolint:unused
}
func (r *tcpMessage) Topic() string {

View File

@@ -18,10 +18,8 @@ type (
netListener struct{}
)
//
// MaxMsgSize set the maximum message in bytes the server can receive and
// send. Default maximum message size is 8K
//
func MaxMsgSize(s int) server.Option {
return server.SetOption(maxMsgSizeKey{}, s)
}

View File

@@ -1,7 +1,6 @@
package tcp
import (
"bytes"
"context"
"fmt"
"reflect"
@@ -10,9 +9,8 @@ import (
"unicode/utf8"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/options"
"go.unistack.org/micro/v3/server"
)
@@ -34,7 +32,6 @@ type tcpSubscriber struct {
typ reflect.Type
subscriber interface{}
handlers []*handler
endpoints []*register.Endpoint
opts server.SubscriberOptions
}
@@ -63,7 +60,6 @@ func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOptio
o(&options)
}
var endpoints []*register.Endpoint
var handlers []*handler
if typ := reflect.TypeOf(sub); typ.Kind() == reflect.Func {
@@ -80,18 +76,7 @@ func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOptio
}
handlers = append(handlers, h)
ep := &register.Endpoint{
Name: "Func",
Request: register.ExtractSubValue(typ),
Metadata: metadata.New(2),
}
ep.Metadata.Set("topic", topic)
ep.Metadata.Set("subscriber", "true")
endpoints = append(endpoints, ep)
} else {
hdlr := reflect.ValueOf(sub)
name := reflect.Indirect(hdlr).Type().Name()
for m := 0; m < typ.NumMethod(); m++ {
method := typ.Method(m)
h := &handler{
@@ -107,14 +92,6 @@ func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOptio
}
handlers = append(handlers, h)
ep := &register.Endpoint{
Name: name + "." + method.Name,
Request: register.ExtractSubValue(method.Type),
Metadata: metadata.New(2),
}
ep.Metadata.Set("topic", topic)
ep.Metadata.Set("subscriber", "true")
endpoints = append(endpoints, ep)
}
}
@@ -124,7 +101,6 @@ func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOptio
topic: topic,
subscriber: sub,
handlers: handlers,
endpoints: endpoints,
opts: options,
}
}
@@ -184,11 +160,11 @@ func validateSubscriber(sub server.Subscriber) error {
return nil
}
func (s *tcpServer) createSubHandler(sb *tcpSubscriber, opts server.Options) broker.Handler {
func (h *Server) createSubHandler(sb *tcpSubscriber, opts server.Options) broker.Handler {
return func(p broker.Event) error {
msg := p.Message()
ct := msg.Header["Content-Type"]
cf, err := s.newCodec(ct)
cf, err := h.newCodec(ct)
if err != nil {
return err
}
@@ -220,16 +196,6 @@ func (s *tcpServer) createSubHandler(sb *tcpSubscriber, opts server.Options) bro
req = req.Elem()
}
buf := bytes.NewBuffer(msg.Body)
if err := cf.ReadHeader(buf, &codec.Message{}, codec.Event); err != nil {
return err
}
if err := cf.ReadBody(buf, req.Interface()); err != nil {
return err
}
fn := func(ctx context.Context, msg server.Message) error {
var vals []reflect.Value
if sb.typ.Kind() != reflect.Func {
@@ -248,9 +214,11 @@ func (s *tcpServer) createSubHandler(sb *tcpSubscriber, opts server.Options) bro
return nil
}
for i := len(opts.SubWrappers); i > 0; i-- {
fn = opts.SubWrappers[i-1](fn)
}
opts.Hooks.EachNext(func(hook options.Hook) {
if h, ok := hook.(server.HookSubHandler); ok {
fn = h(fn)
}
})
go func() {
results <- fn(ctx, &tcpMessage{
@@ -287,10 +255,6 @@ func (s *tcpSubscriber) Subscriber() interface{} {
return s.subscriber
}
func (s *tcpSubscriber) Endpoints() []*register.Endpoint {
return s.endpoints
}
func (s *tcpSubscriber) Options() server.SubscriberOptions {
return s.opts
}

143
tcp.go
View File

@@ -1,12 +1,14 @@
// Package tcp implements a go-micro.Server
package tcp // import "go.unistack.org/micro-server-tcp/v3"
package tcp
import (
"context"
"crypto/tls"
"fmt"
"net"
"sort"
"sync"
"sync/atomic"
"time"
"go.unistack.org/micro/v3/broker"
@@ -14,34 +16,55 @@ import (
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/register"
"go.unistack.org/micro/v3/server"
msync "go.unistack.org/micro/v3/sync"
"golang.org/x/net/netutil"
)
type tcpServer struct {
var _ server.Server = (*Server)(nil)
type Server struct {
hd server.Handler
rsvc *register.Service
exit chan chan error
subscribers map[*tcpSubscriber][]broker.Subscriber
opts server.Options
sync.RWMutex
registered bool
init bool
registered bool
init bool
stateLive *atomic.Uint32
stateReady *atomic.Uint32
stateHealth *atomic.Uint32
}
func (h *tcpServer) newCodec(ct string) (codec.Codec, error) {
func (h *Server) Live() bool {
return h.stateLive.Load() == 1
}
func (h *Server) Ready() bool {
return h.stateReady.Load() == 1
}
func (h *Server) Health() bool {
return h.stateHealth.Load() == 1
}
func (h *Server) newCodec(ct string) (codec.Codec, error) {
if cf, ok := h.opts.Codecs[ct]; ok {
return cf, nil
}
return nil, codec.ErrUnknownContentType
}
func (h *tcpServer) Options() server.Options {
func (h *Server) Options() server.Options {
h.RLock()
defer h.RUnlock()
return h.opts
}
func (h *tcpServer) Init(opts ...server.Option) error {
func (h *Server) Init(opts ...server.Option) error {
if h.opts.Wait == nil {
h.opts.Wait = msync.NewWaitGroup()
}
if len(opts) == 0 && h.init {
return nil
}
@@ -66,33 +89,21 @@ func (h *tcpServer) Init(opts ...server.Option) error {
if err := h.opts.Meter.Init(); err != nil {
return err
}
if err := h.opts.Transport.Init(); err != nil {
return err
}
return nil
}
func (h *tcpServer) Handle(handler server.Handler) error {
func (h *Server) Handle(handler server.Handler) error {
h.Lock()
h.hd = handler
h.Unlock()
return nil
}
func (h *tcpServer) NewHandler(handler interface{}, opts ...server.HandlerOption) server.Handler {
func (h *Server) NewHandler(handler interface{}, opts ...server.HandlerOption) server.Handler {
options := server.NewHandlerOptions(opts...)
eps := make([]*register.Endpoint, 0, len(options.Metadata))
for name, metadata := range options.Metadata {
eps = append(eps, &register.Endpoint{
Name: name,
Metadata: metadata,
})
}
th := &tcpHandler{
eps: eps,
hd: handler,
opts: options,
}
@@ -104,11 +115,11 @@ func (h *tcpServer) NewHandler(handler interface{}, opts ...server.HandlerOption
return th
}
func (h *tcpServer) NewSubscriber(topic string, handler interface{}, opts ...server.SubscriberOption) server.Subscriber {
func (h *Server) NewSubscriber(topic string, handler interface{}, opts ...server.SubscriberOption) server.Subscriber {
return newSubscriber(topic, handler, opts...)
}
func (h *tcpServer) Subscribe(sb server.Subscriber) error {
func (h *Server) Subscribe(sb server.Subscriber) error {
sub, ok := sb.(*tcpSubscriber)
if !ok {
return fmt.Errorf("invalid subscriber: expected *tcpSubscriber")
@@ -131,11 +142,11 @@ func (h *tcpServer) Subscribe(sb server.Subscriber) error {
return nil
}
func (h *tcpServer) Register() error {
func (h *Server) Register() error {
h.Lock()
config := h.opts
rsvc := h.rsvc
eps := h.hd.Endpoints()
h.Unlock()
// if service already filled, reuse it and return early
@@ -153,7 +164,6 @@ func (h *tcpServer) Register() error {
service.Nodes[0].Metadata["protocol"] = "tcp"
service.Nodes[0].Metadata["transport"] = service.Nodes[0].Metadata["protocol"]
service.Endpoints = eps
h.Lock()
@@ -165,9 +175,7 @@ func (h *tcpServer) Register() error {
sort.Slice(subscriberList, func(i, j int) bool {
return subscriberList[i].topic > subscriberList[j].topic
})
for _, e := range subscriberList {
service.Endpoints = append(service.Endpoints, e.Endpoints()...)
}
h.Unlock()
h.RLock()
@@ -176,7 +184,7 @@ func (h *tcpServer) Register() error {
if !registered {
if config.Logger.V(logger.InfoLevel) {
config.Logger.Infof(config.Context, "Register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID)
config.Logger.Info(config.Context, fmt.Sprintf("Register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID))
}
}
@@ -212,7 +220,7 @@ func (h *tcpServer) Register() error {
opts = append(opts, broker.SubscribeAutoAck(sb.Options().AutoAck))
if config.Logger.V(logger.InfoLevel) {
config.Logger.Infof(config.Context, "Subscribing to topic: %s", sb.Topic())
config.Logger.Info(config.Context, "Subscribing to topic: "+sb.Topic())
}
sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), handler, opts...)
@@ -228,7 +236,7 @@ func (h *tcpServer) Register() error {
return nil
}
func (h *tcpServer) Deregister() error {
func (h *Server) Deregister() error {
h.Lock()
config := h.opts
h.Unlock()
@@ -239,7 +247,7 @@ func (h *tcpServer) Deregister() error {
}
if config.Logger.V(logger.InfoLevel) {
config.Logger.Infof(config.Context, "Deregistering node: %s", service.Nodes[0].ID)
config.Logger.Info(config.Context, "Deregistering node: "+service.Nodes[0].ID)
}
if err := server.DefaultDeregisterFunc(service, config); err != nil {
@@ -266,11 +274,11 @@ func (h *tcpServer) Deregister() error {
go func(s broker.Subscriber) {
defer wg.Done()
if config.Logger.V(logger.InfoLevel) {
config.Logger.Infof(config.Context, "Unsubscribing from topic: %s", s.Topic())
config.Logger.Info(config.Context, "Unsubscribing from topic: "+s.Topic())
}
if err := s.Unsubscribe(subCtx); err != nil {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Unsubscribing from topic: %s err: %v", s.Topic(), err)
config.Logger.Error(config.Context, "Unsubscribing from errot topic: "+s.Topic(), err)
}
}
}(sub)
@@ -283,7 +291,7 @@ func (h *tcpServer) Deregister() error {
return nil
}
func (h *tcpServer) getListener() net.Listener {
func (h *Server) getListener() net.Listener {
if h.opts.Context == nil {
return nil
}
@@ -296,7 +304,7 @@ func (h *tcpServer) getListener() net.Listener {
return l
}
func (h *tcpServer) Start() error {
func (h *Server) Start() error {
h.RLock()
config := h.opts
hd := h.hd.Handler()
@@ -330,7 +338,7 @@ func (h *tcpServer) Start() error {
}
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Infof(config.Context, "Listening on %s", ts.Addr().String())
config.Logger.Info(config.Context, "Listening on "+ts.Addr().String())
}
h.Lock()
@@ -351,6 +359,9 @@ func (h *tcpServer) Start() error {
return fmt.Errorf("invalid handler %T", hd)
}
go h.serve(ts, handle)
h.stateLive.Store(1)
h.stateReady.Store(1)
h.stateHealth.Store(1)
go func() {
t := new(time.Ticker)
@@ -376,23 +387,23 @@ func (h *tcpServer) Start() error {
// nolint: nestif
if rerr != nil && registered {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s, deregister it", config.Name, config.ID, rerr)
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s deregister, check error", config.Name, config.ID), rerr)
}
// deregister self in case of error
if err := h.Deregister(); err != nil {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Server %s-%s deregister error: %s", config.Name, config.ID, err)
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s deregister error", config.Name, config.ID), err)
}
}
} else if rerr != nil && !registered {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Server %s-%s register check error: %s", config.Name, config.ID, rerr)
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register check error", config.Name, config.ID), rerr)
}
continue
}
if err := h.Register(); err != nil {
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "Server %s-%s register error: %s", config.Name, config.ID, err)
config.Logger.Error(config.Context, fmt.Sprintf("Server %s-%s register error", config.Name, config.ID), err)
}
}
// wait for exit
@@ -401,36 +412,49 @@ func (h *tcpServer) Start() error {
}
}
h.gracefulStop()
ch <- ts.Close()
h.stateLive.Store(0)
h.stateReady.Store(0)
h.stateHealth.Store(0)
// deregister
if cerr := h.Deregister(); cerr != nil {
config.Logger.Errorf(config.Context, "Register deregister error: %v", cerr)
config.Logger.Error(config.Context, "Register deregister error", cerr)
}
if cerr := config.Broker.Disconnect(config.Context); cerr != nil {
config.Logger.Errorf(config.Context, "Broker disconnect error: %v", cerr)
config.Logger.Error(config.Context, "Broker disconnect error", cerr)
}
}()
return nil
}
func (h *tcpServer) Stop() error {
func (h *Server) Stop() error {
ch := make(chan error)
h.exit <- ch
return <-ch
}
func (h *tcpServer) String() string {
func (h *Server) gracefulStop() {
ctx, cancel := context.WithTimeout(context.Background(), h.opts.GracefulTimeout)
defer cancel()
h.opts.Wait.WaitContext(ctx)
}
func (h *Server) String() string {
return "tcp"
}
func (h *tcpServer) Name() string {
func (h *Server) Name() string {
return h.opts.Name
}
func (h *tcpServer) serve(ln net.Listener, hd Handler) {
func (h *Server) serve(ln net.Listener, hd Handler) {
var tempDelay time.Duration // how long to sleep on accept failure
h.RLock()
config := h.opts
@@ -444,7 +468,7 @@ func (h *tcpServer) serve(ln net.Listener, hd Handler) {
return
default:
}
if ne, ok := err.(net.Error); ok && ne.Temporary() {
if ne, ok := err.(net.Error); ok && ne.Temporary() { //nolint:staticcheck
if tempDelay == 0 {
tempDelay = 5 * time.Millisecond
} else {
@@ -454,27 +478,34 @@ func (h *tcpServer) serve(ln net.Listener, hd Handler) {
tempDelay = max
}
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "tcp: Accept error: %v; retrying in %v", err, tempDelay)
config.Logger.Error(config.Context, fmt.Sprintf("tcp: Accept error: %v; retrying in %v", err, tempDelay))
}
time.Sleep(tempDelay)
continue
}
if config.Logger.V(logger.ErrorLevel) {
config.Logger.Errorf(config.Context, "tcp: Accept error: %v", err)
config.Logger.Error(config.Context, "tcp: Accept error", err)
}
return
}
if err != nil {
config.Logger.Errorf(config.Context, "tcp: accept err: %v", err)
return
if h.opts.Wait != nil {
h.opts.Wait.Add(1)
}
go hd.Serve(c)
go func() {
hd.Serve(c)
if h.opts.Wait != nil {
h.opts.Wait.Done()
}
}()
}
}
func NewServer(opts ...server.Option) server.Server {
return &tcpServer{
return &Server{
stateLive: &atomic.Uint32{},
stateReady: &atomic.Uint32{},
stateHealth: &atomic.Uint32{},
opts: server.NewOptions(opts...),
exit: make(chan chan error),
subscribers: make(map[*tcpSubscriber][]broker.Subscriber),