unistack-org/micro-client-grpc
4
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 35 Wiki Activity

Compare commits

base: unistack-org:v3
Branches Tags
unistack-org:master
unistack-org:v3
unistack-org:v3.11.10
unistack-org:v3.11.9
unistack-org:v3.11.8
unistack-org:v3.11.7
unistack-org:v3.11.6
unistack-org:v3.11.5
unistack-org:v3.11.4
unistack-org:v3.11.3
unistack-org:v3.11.2
unistack-org:v3.11.1
unistack-org:v3.10.0
unistack-org:v3.9.0
unistack-org:v3.8.7
unistack-org:v3.8.6
unistack-org:v3.8.5
unistack-org:v3.8.4
unistack-org:v3.8.3
unistack-org:v3.8.2
unistack-org:v3.8.1
unistack-org:v3.8.0
unistack-org:v3.7.8
unistack-org:v3.7.4
unistack-org:v3.4.0
unistack-org:v3.3.6
unistack-org:v3.3.5
unistack-org:v3.3.4
unistack-org:v3.3.3
unistack-org:v3.3.2
unistack-org:v3.3.1
unistack-org:v3.3.0
unistack-org:v3.2.4
unistack-org:v3.2.3
unistack-org:v3.1.1
unistack-org:v0.0.2
unistack-org:v0.0.3
unistack-org:v0.0.1
..
compare: unistack-org:v3.8.1
Branches Tags
unistack-org:v3
unistack-org:master
unistack-org:v3.11.10
unistack-org:v3.11.9
unistack-org:v3.11.8
unistack-org:v3.11.7
unistack-org:v3.11.6
unistack-org:v3.11.5
unistack-org:v3.11.4
unistack-org:v3.11.3
unistack-org:v3.11.2
unistack-org:v3.11.1
unistack-org:v3.10.0
unistack-org:v3.9.0
unistack-org:v3.8.7
unistack-org:v3.8.6
unistack-org:v3.8.5
unistack-org:v3.8.4
unistack-org:v3.8.3
unistack-org:v3.8.2
unistack-org:v3.8.1
unistack-org:v3.8.0
unistack-org:v3.7.8
unistack-org:v3.7.4
unistack-org:v3.4.0
unistack-org:v3.3.6
unistack-org:v3.3.5
unistack-org:v3.3.4
unistack-org:v3.3.3
unistack-org:v3.3.2
unistack-org:v3.3.1
unistack-org:v3.3.0
unistack-org:v3.2.4
unistack-org:v3.2.3
unistack-org:v3.1.1
unistack-org:v0.0.2
unistack-org:v0.0.3
unistack-org:v0.0.1

No commits in common. "v3" and "v3.8.1" have entirely different histories.
v3 ... v3.8.1

17 changed files with 439 additions and 509 deletions
Show Stats Expand all files Collapse all files

20
.github/workflows/autoapprove.yml vendored
View File

@ -1,20 +0,0 @@
name: "autoapprove"
on:
pull_request_target:
types: [assigned, opened, synchronize, reopened]
permissions:
pull-requests: write
contents: write
jobs:
autoapprove:
runs-on: ubuntu-latest
steps:
- name: approve
uses: hmarr/auto-approve-action@v3
if: github.actor == 'vtolstov' || github.actor == 'dependabot[bot]'
id: approve
with:
github-token: ${{ secrets.GITHUB_TOKEN }}

21
.github/workflows/automerge.yml vendored
View File

@ -1,21 +0,0 @@
name: "automerge"
on:
pull_request_target:
types: [assigned, opened, synchronize, reopened]
permissions:
pull-requests: write
contents: write
jobs:
automerge:
runs-on: ubuntu-latest
if: github.actor == 'vtolstov'
steps:
- name: merge
id: merge
run: gh pr merge --auto --merge "$PR_URL"
env:
PR_URL: ${{github.event.pull_request.html_url}}
GITHUB_TOKEN: ${{secrets.TOKEN}}

13
.github/workflows/build.yml vendored
View File

@ -3,20 +3,19 @@ on:
push: push:
branches: branches:
- master - master
- v3
jobs: jobs:
test: test:
name: test name: test
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: setup - name: setup
uses: actions/setup-go@v3 uses: actions/setup-go@v2
with: with:
go-version: 1.17 go-version: 1.16
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v2
- name: cache - name: cache
uses: actions/cache@v3 uses: actions/cache@v2
with: with:
path: ~/go/pkg/mod path: ~/go/pkg/mod
key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }} key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
@ -32,9 +31,9 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v2
- name: lint - name: lint
uses: golangci/golangci-lint-action@v3.4.0 uses: golangci/golangci-lint-action@v2
continue-on-error: true continue-on-error: true
with: with:
# Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version. # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.

12
.github/workflows/codeql-analysis.yml vendored
View File

@ -43,14 +43,14 @@ jobs:
steps: steps:
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v2
- name: setup - name: setup
uses: actions/setup-go@v3 uses: actions/setup-go@v2
with: with:
go-version: 1.17 go-version: 1.16
# Initializes the CodeQL tools for scanning. # Initializes the CodeQL tools for scanning.
- name: init - name: init
uses: github/codeql-action/init@v2 uses: github/codeql-action/init@v1
with: with:
languages: ${{ matrix.language }} languages: ${{ matrix.language }}
# If you wish to specify custom queries, you can do so here or in a config file. # If you wish to specify custom queries, you can do so here or in a config file.
@ -61,7 +61,7 @@ jobs:
# Autobuild attempts to build any compiled languages (C/C++, C#, or Java). # Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
# If this step fails, then you should remove it and run the build manually (see below) # If this step fails, then you should remove it and run the build manually (see below)
- name: autobuild - name: autobuild
uses: github/codeql-action/autobuild@v2 uses: github/codeql-action/autobuild@v1
# ℹ️ Command-line programs to run using the OS shell. # ℹ️ Command-line programs to run using the OS shell.
# 📚 https://git.io/JvXDl # 📚 https://git.io/JvXDl
@ -75,4 +75,4 @@ jobs:
# make release # make release
- name: analyze - name: analyze
uses: github/codeql-action/analyze@v2 uses: github/codeql-action/analyze@v1

14
.github/workflows/dependabot-automerge.yml vendored
View File

@ -1,4 +1,4 @@
name: "dependabot-automerge" name: "prautomerge"
on: on:
pull_request_target: pull_request_target:
@ -9,17 +9,21 @@ permissions:
contents: write contents: write
jobs: jobs:
automerge: dependabot:
runs-on: ubuntu-latest runs-on: ubuntu-latest
if: github.actor == 'dependabot[bot]' if: ${{ github.actor == 'dependabot[bot]' }}
steps: steps:
- name: metadata - name: metadata
id: metadata id: metadata
uses: dependabot/fetch-metadata@v1.3.6 uses: dependabot/fetch-metadata@v1.1.1
with: with:
github-token: "${{ secrets.TOKEN }}" github-token: "${{ secrets.TOKEN }}"
- name: approve
run: gh pr review --approve "$PR_URL"
env:
PR_URL: ${{github.event.pull_request.html_url}}
GITHUB_TOKEN: ${{secrets.TOKEN}}
- name: merge - name: merge
id: merge
if: ${{contains(steps.metadata.outputs.dependency-names, 'go.unistack.org')}} if: ${{contains(steps.metadata.outputs.dependency-names, 'go.unistack.org')}}
run: gh pr merge --auto --merge "$PR_URL" run: gh pr merge --auto --merge "$PR_URL"
env: env:

13
.github/workflows/pr.yml vendored
View File

@ -3,20 +3,19 @@ on:
pull_request: pull_request:
branches: branches:
- master - master
- v3
jobs: jobs:
test: test:
name: test name: test
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: setup - name: setup
uses: actions/setup-go@v3 uses: actions/setup-go@v2
with: with:
go-version: 1.17 go-version: 1.16
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v2
- name: cache - name: cache
uses: actions/cache@v3 uses: actions/cache@v2
with: with:
path: ~/go/pkg/mod path: ~/go/pkg/mod
key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }} key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
@ -32,9 +31,9 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v2
- name: lint - name: lint
uses: golangci/golangci-lint-action@v3.4.0 uses: golangci/golangci-lint-action@v2
continue-on-error: true continue-on-error: true
with: with:
# Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version. # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.

24
.gitignore vendored
View File

@ -1,24 +0,0 @@
# 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

62
codec.go
View File

@ -1,6 +1,8 @@
package grpc package grpc
import ( import (
"io"
"go.unistack.org/micro/v3/codec" "go.unistack.org/micro/v3/codec"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/encoding" "google.golang.org/grpc/encoding"
@ -63,3 +65,63 @@ func (w *wrapGrpcCodec) Unmarshal(d []byte, v interface{}, opts ...codec.Option)
} }
return w.Codec.Unmarshal(d, v) return w.Codec.Unmarshal(d, v)
} }
/*
type grpcCodec struct {
grpc.ServerStream
// headers
id string
target string
method string
endpoint string
c encoding.Codec
}
*/
func (w *wrapGrpcCodec) ReadHeader(conn io.Reader, m *codec.Message, mt codec.MessageType) error {
/*
if m == nil {
m = codec.NewMessage(codec.Request)
}
if md, ok := metadata.FromIncomingContext(g.ServerStream.Context()); ok {
if m.Header == nil {
m.Header = meta.New(len(md))
}
for k, v := range md {
m.Header[k] = strings.Join(v, ",")
}
}
m.Id = g.id
m.Target = g.target
m.Method = g.method
m.Endpoint = g.endpoint
*/
return nil
}
func (w *wrapGrpcCodec) ReadBody(conn io.Reader, v interface{}) error {
// caller has requested a frame
if m, ok := v.(*codec.Frame); ok {
_, err := conn.Read(m.Data)
return err
}
return codec.ErrInvalidMessage
}
func (w *wrapGrpcCodec) Write(conn io.Writer, m *codec.Message, v interface{}) error {
// if we don't have a body
if v != nil {
b, err := w.Marshal(v)
if err != nil {
return err
}
m.Body = b
}
// write the body using the framing codec
_, err := conn.Write(m.Body)
return err
}

38
error.go
View File

@ -9,46 +9,36 @@ func microError(err error) error {
// no error // no error
if err == nil { if err == nil {
// nothing to do
return nil return nil
} }
if verr, ok := err.(*errors.Error); ok { if verr, ok := err.(*errors.Error); ok {
// micro error
return verr return verr
} }
// grpc error // grpc error
s, ok := status.FromError(err) s, ok := status.FromError(err)
if !ok { if !ok {
// can't get status detals from grpc error, return base error
return err return err
} }
details := s.Details() // return first error from details
switch len(details) { if details := s.Details(); len(details) > 0 {
case 0: if verr, ok := details[0].(error); ok {
if verr := errors.Parse(s.Message()); verr.Code > 0 { return microError(verr)
// return micro error
return verr
} }
// return base error as it not micro error
return err
case 1:
if verr, ok := details[0].(*errors.Error); ok {
// return nested micro error
return verr
}
// return base error as it not holds micro error
return err
} }
// attached messages in details more then 1, try to fallback to micro error // try to decode micro *errors.Error
if verr := errors.Parse(s.Message()); verr.Code > 0 { if e := errors.Parse(s.Message()); e.Code > 0 {
// return micro error return e // actually a micro error
return verr
} }
// not micro error return base error // fallback
return err return &errors.Error{
ID: "go.micro.client",
Code: int32(s.Code()),
Detail: s.Message(),
Status: s.Code().String(),
}
} }

19
go.mod
View File

@ -1,19 +1,10 @@
module go.unistack.org/micro-client-grpc/v3 module go.unistack.org/micro-client-grpc/v3
go 1.21 go 1.16
toolchain go1.23.1
require ( require (
go.unistack.org/micro/v3 v3.10.91 go.unistack.org/micro/v3 v3.8.12
google.golang.org/grpc v1.67.0 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
) google.golang.org/genproto v0.0.0-20210729151513-df9385d47c1b // indirect
google.golang.org/grpc v1.42.0
require (
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
golang.org/x/net v0.29.0 // indirect
golang.org/x/sys v0.25.0 // indirect
golang.org/x/text v0.18.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
google.golang.org/protobuf v1.34.2 // indirect
) )

177
go.sum
View File

@ -1,18 +1,159 @@
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo= github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
go.unistack.org/micro/v3 v3.10.91 h1:vuJY4tXwpqimwIkEJ3TozMYNVQQs+C5QMlQWPgSY/YM= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
go.unistack.org/micro/v3 v3.10.91/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg= github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0= github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34= github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224= github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY= github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ= github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU= github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
google.golang.org/grpc v1.67.0 h1:IdH9y6PF5MPSdAntIcpjQ+tXO41pcQsfZV2RxtQgVcw= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
google.golang.org/grpc v1.67.0/go.mod h1:1gLDyUQU7CTLJI90u3nXZ9ekeghjeM7pTDZlqFNg2AA= github.com/ef-ds/deque v1.0.4/go.mod h1:gXDnTC3yqvBcHbq2lcExjtAcVrOnJCbMcZXmuj8Z4tg=
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
github.com/golang-jwt/jwt/v4 v4.1.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
github.com/silas/dag v0.0.0-20210626123444-3804bac2d6d4/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
go.unistack.org/micro-proto/v3 v3.1.0 h1:q39FwjFiRZn+Ux/tt+d3bJTmDtsQQWa+3SLYVo1vLfA=
go.unistack.org/micro-proto/v3 v3.1.0/go.mod h1:DpRhYCBXlmSJ/AAXTmntvlh7kQkYU6eFvlmYAx4BQS8=
go.unistack.org/micro/v3 v3.8.12 h1:ACaHE8ZIHFXqEGPSRvXzND4hcqCSQf04WkzOFY6Y1gQ=
go.unistack.org/micro/v3 v3.8.12/go.mod h1:KMMmOmbgo/D52/rCAbqeKbBsgEEbSKM69he54J3ZIuA=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
golang.org/x/lint v0.0.0-20210508222113-6edffad5e616/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b h1:eB48h3HiRycXNy8E0Gf5e0hv7YT6Kt14L/D73G1fuwo=
golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I=
golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
google.golang.org/genproto v0.0.0-20200513103714-09dca8ec2884/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
google.golang.org/genproto v0.0.0-20210729151513-df9385d47c1b h1:4xoALQmXxqVdDdLimpPyPeDdsJzo+nFTJw9euAMpqgM=
google.golang.org/genproto v0.0.0-20210729151513-df9385d47c1b/go.mod h1:ob2IJxKrgPT52GcgX759i1sleT07tiKowYBGbczaW48=
google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
google.golang.org/grpc v1.33.1/go.mod h1:fr5YgcSWrqhRRxogOsw7RzIpsmvOZ6IcH4kBYTpR3n0=
google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
google.golang.org/grpc v1.42.0 h1:XT2/MFpuPFsEX2fWh3YQtHkZ+WYZFQRfaUgLZYj/p6A=
google.golang.org/grpc v1.42.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

389
grpc.go
View File

@ -1,5 +1,5 @@
// Package grpc provides a gRPC client for micro framework // Package grpc provides a gRPC client
package grpc package grpc // import "go.unistack.org/micro-client-grpc/v3"
import ( import (
"context" "context"
@ -8,7 +8,6 @@ import (
"net" "net"
"os" "os"
"reflect" "reflect"
"strconv"
"strings" "strings"
"sync" "sync"
"time" "time"
@ -18,29 +17,18 @@ import (
"go.unistack.org/micro/v3/codec" "go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/errors" "go.unistack.org/micro/v3/errors"
"go.unistack.org/micro/v3/metadata" "go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/options"
"go.unistack.org/micro/v3/selector"
"go.unistack.org/micro/v3/semconv"
"go.unistack.org/micro/v3/tracer"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/encoding" "google.golang.org/grpc/encoding"
gmetadata "google.golang.org/grpc/metadata" gmetadata "google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
) )
const ( const (
DefaultContentType = "application/grpc" defaultContentType = "application/grpc+proto"
) )
type grpcClient struct { type grpcClient struct {
funcPublish client.FuncPublish pool *pool
funcBatchPublish client.FuncBatchPublish
funcCall client.FuncCall
funcStream client.FuncStream
pool *ConnPool
opts client.Options opts client.Options
sync.RWMutex sync.RWMutex
init bool init bool
@ -74,7 +62,7 @@ func (g *grpcClient) secure(addr string) grpc.DialOption {
} }
// other fallback to insecure // other fallback to insecure
return grpc.WithTransportCredentials(insecure.NewCredentials()) return grpc.WithInsecure()
} }
func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error { func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
@ -88,15 +76,9 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
} else { } else {
header = make(map[string]string, 2) header = make(map[string]string, 2)
} }
if opts.RequestMetadata != nil {
for k, v := range opts.RequestMetadata {
header[k] = v
}
}
// set timeout in nanoseconds // set timeout in nanoseconds
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout) header["timeout"] = fmt.Sprintf("%d", opts.RequestTimeout)
header["timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
header["content-type"] = req.ContentType()
md := gmetadata.New(header) md := gmetadata.New(header)
ctx = gmetadata.NewOutgoingContext(ctx, md) ctx = gmetadata.NewOutgoingContext(ctx, md)
@ -108,7 +90,6 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
maxRecvMsgSize := g.maxRecvMsgSizeValue() maxRecvMsgSize := g.maxRecvMsgSizeValue()
maxSendMsgSize := g.maxSendMsgSizeValue() maxSendMsgSize := g.maxSendMsgSizeValue()
cfgService := g.serviceConfig()
var grr error var grr error
@ -127,50 +108,31 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
grpc.MaxCallRecvMsgSize(maxRecvMsgSize), grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
grpc.MaxCallSendMsgSize(maxSendMsgSize), grpc.MaxCallSendMsgSize(maxSendMsgSize),
), ),
grpc.WithDefaultServiceConfig(cfgService),
} }
if opts := g.getGrpcDialOptions(g.opts.Context); opts != nil { if opts := g.getGrpcDialOptions(); opts != nil {
grpcDialOptions = append(grpcDialOptions, opts...)
}
if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
grpcDialOptions = append(grpcDialOptions, opts...) grpcDialOptions = append(grpcDialOptions, opts...)
} }
contextDialer := g.opts.ContextDialer cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
if opts.ContextDialer != nil {
contextDialer = opts.ContextDialer
}
if contextDialer != nil {
grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
}
cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...)
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }
defer func() { defer func() {
// defer execution of release // defer execution of release
g.pool.Put(cc, grr) g.pool.release(cc, grr)
}() }()
ch := make(chan error, 1) ch := make(chan error, 1)
var gmd gmetadata.MD
grpcCallOptions := []grpc.CallOption{
grpc.CallContentSubtype((&wrapMicroCodec{cf}).Name()),
}
if opts := g.getGrpcCallOptions(opts.Context); opts != nil {
grpcCallOptions = append(grpcCallOptions, opts...)
}
if opts.ResponseMetadata != nil {
gmd = gmetadata.MD{}
grpcCallOptions = append(grpcCallOptions, grpc.Header(&gmd))
}
go func() { go func() {
grpcCallOptions := []grpc.CallOption{
grpc.ForceCodec(&wrapMicroCodec{cf}),
grpc.CallContentSubtype((&wrapMicroCodec{cf}).Name()),
}
if opts := g.getGrpcCallOptions(); opts != nil {
grpcCallOptions = append(grpcCallOptions, opts...)
}
err := cc.Invoke(ctx, methodToGRPC(req.Service(), req.Endpoint()), req.Body(), rsp, grpcCallOptions...) err := cc.Invoke(ctx, methodToGRPC(req.Service(), req.Endpoint()), req.Body(), rsp, grpcCallOptions...)
ch <- microError(err) ch <- microError(err)
}() }()
@ -182,13 +144,6 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
grr = errors.Timeout("go.micro.client", "%v", ctx.Err()) grr = errors.Timeout("go.micro.client", "%v", ctx.Err())
} }
if opts.ResponseMetadata != nil {
*opts.ResponseMetadata = metadata.New(gmd.Len())
for k, v := range gmd {
opts.ResponseMetadata.Set(k, strings.Join(v, ","))
}
}
return grr return grr
} }
@ -206,11 +161,10 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
// set timeout in nanoseconds // set timeout in nanoseconds
if opts.StreamTimeout > time.Duration(0) { if opts.StreamTimeout > time.Duration(0) {
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout) header["timeout"] = fmt.Sprintf("%d", opts.StreamTimeout)
header["timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
} }
// set the content type for the request // set the content type for the request
header["content-type"] = req.ContentType() header["x-content-type"] = req.ContentType()
md := gmetadata.New(header) md := gmetadata.New(header)
ctx = gmetadata.NewOutgoingContext(ctx, md) ctx = gmetadata.NewOutgoingContext(ctx, md)
@ -233,7 +187,6 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
maxRecvMsgSize := g.maxRecvMsgSizeValue() maxRecvMsgSize := g.maxRecvMsgSizeValue()
maxSendMsgSize := g.maxSendMsgSizeValue() maxSendMsgSize := g.maxSendMsgSizeValue()
cfgService := g.serviceConfig()
grpcDialOptions := []grpc.DialOption{ grpcDialOptions := []grpc.DialOption{
g.secure(addr), g.secure(addr),
@ -241,22 +194,13 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
grpc.MaxCallRecvMsgSize(maxRecvMsgSize), grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
grpc.MaxCallSendMsgSize(maxSendMsgSize), grpc.MaxCallSendMsgSize(maxSendMsgSize),
), ),
grpc.WithDefaultServiceConfig(cfgService),
} }
if opts := g.getGrpcDialOptions(opts.Context); opts != nil { if opts := g.getGrpcDialOptions(); opts != nil {
grpcDialOptions = append(grpcDialOptions, opts...) grpcDialOptions = append(grpcDialOptions, opts...)
} }
contextDialer := g.opts.ContextDialer cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
if opts.ContextDialer != nil {
contextDialer = opts.ContextDialer
}
if contextDialer != nil {
grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
}
cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...)
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }
@ -268,17 +212,12 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
} }
grpcCallOptions := []grpc.CallOption{ grpcCallOptions := []grpc.CallOption{
// grpc.ForceCodec(wc), grpc.ForceCodec(wc),
grpc.CallContentSubtype(wc.Name()), grpc.CallContentSubtype(wc.Name()),
} }
if opts := g.getGrpcCallOptions(opts.Context); opts != nil { if opts := g.getGrpcCallOptions(); opts != nil {
grpcCallOptions = append(grpcCallOptions, opts...) grpcCallOptions = append(grpcCallOptions, opts...)
} }
var gmd gmetadata.MD
if opts.ResponseMetadata != nil {
gmd = gmetadata.MD{}
grpcCallOptions = append(grpcCallOptions, grpc.Header(&gmd))
}
// create a new cancelling context // create a new cancelling context
newCtx, cancel := context.WithCancel(ctx) newCtx, cancel := context.WithCancel(ctx)
@ -289,7 +228,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
// cancel the context // cancel the context
cancel() cancel()
// release the connection // release the connection
g.pool.Put(cc, err) g.pool.release(cc, err)
// now return the error // now return the error
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
} }
@ -317,14 +256,13 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
} }
// defer execution of release // defer execution of release
g.pool.Put(cc, err) g.pool.release(cc, err)
}, },
} }
// set the stream as the response // set the stream as the response
val := reflect.ValueOf(rsp).Elem() val := reflect.ValueOf(rsp).Elem()
val.Set(reflect.ValueOf(stream).Elem()) val.Set(reflect.ValueOf(stream).Elem())
return nil return nil
} }
@ -386,17 +324,6 @@ func (g *grpcClient) newCodec(ct string) (codec.Codec, error) {
return nil, codec.ErrUnknownContentType return nil, codec.ErrUnknownContentType
} }
func (g *grpcClient) serviceConfig() string {
if g.opts.Context == nil {
return DefaultServiceConfig
}
v := g.opts.Context.Value(serviceConfigKey{})
if v == nil {
return DefaultServiceConfig
}
return v.(string)
}
func (g *grpcClient) Init(opts ...client.Option) error { func (g *grpcClient) Init(opts ...client.Option) error {
if len(opts) == 0 && g.init { if len(opts) == 0 && g.init {
return nil return nil
@ -416,23 +343,26 @@ func (g *grpcClient) Init(opts ...client.Option) error {
g.pool.Unlock() g.pool.Unlock()
} }
g.funcCall = g.fnCall if err := g.opts.Broker.Init(); err != nil {
g.funcStream = g.fnStream return err
g.funcPublish = g.fnPublish
g.funcBatchPublish = g.fnBatchPublish
g.opts.Hooks.EachNext(func(hook options.Hook) {
switch h := hook.(type) {
case client.HookCall:
g.funcCall = h(g.funcCall)
case client.HookStream:
g.funcStream = h(g.funcStream)
case client.HookPublish:
g.funcPublish = h(g.funcPublish)
case client.HookBatchPublish:
g.funcBatchPublish = h(g.funcBatchPublish)
} }
}) if err := g.opts.Tracer.Init(); err != nil {
return err
}
if err := g.opts.Router.Init(); err != nil {
return err
}
if err := g.opts.Logger.Init(); err != nil {
return err
}
if err := g.opts.Meter.Init(); err != nil {
return err
}
if err := g.opts.Transport.Init(); err != nil {
return err
}
g.init = true
return nil return nil
} }
@ -455,35 +385,8 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
} else if rsp == nil { } else if rsp == nil {
return errors.InternalServerError("go.micro.client", "rsp is nil") return errors.InternalServerError("go.micro.client", "rsp is nil")
} }
ts := time.Now()
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
var sp tracer.Span
ctx, sp = g.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
tracer.WithSpanKind(tracer.SpanKindClient),
tracer.WithSpanLabels("endpoint", req.Endpoint()),
)
err := g.funcCall(ctx, req, rsp, opts...)
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
te := time.Since(ts)
g.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
g.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
if me := errors.FromError(err); me == nil {
sp.Finish()
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(200))).Inc()
} else {
sp.SetStatus(tracer.SpanStatusError, err.Error())
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code))).Inc()
}
return err
}
func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
// make a copy of call opts // make a copy of call opts
callOpts := g.opts.CallOptions callOpts := g.opts.CallOptions
for _, opt := range opts { for _, opt := range opts {
opt(&callOpts) opt(&callOpts)
} }
@ -512,6 +415,11 @@ func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interfa
// make copy of call method // make copy of call method
gcall := g.call gcall := g.call
// wrap the call in reverse
for i := len(callOpts.CallWrappers); i > 0; i-- {
gcall = callOpts.CallWrappers[i-1](gcall)
}
// use the router passed as a call option, or fallback to the rpc clients router // use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil { if callOpts.Router == nil {
callOpts.Router = g.opts.Router callOpts.Router = g.opts.Router
@ -527,8 +435,20 @@ func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interfa
callOpts.Address = []string{g.opts.Proxy} callOpts.Address = []string{g.opts.Proxy}
} }
var next selector.Next // lookup the route to send the reques to
// TODO apply any filtering here
routes, err := g.opts.Lookup(ctx, req, callOpts)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err := callOpts.Selector.Select(routes)
if err != nil {
return err
}
// return errors.New("go.micro.client", "request timeout", 408)
call := func(i int) error { call := func(i int) error {
// call backoff first. Someone may want an initial start delay // call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, req, i) t, err := callOpts.Backoff(ctx, req, i)
@ -541,23 +461,6 @@ func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interfa
time.Sleep(t) time.Sleep(t)
} }
if next == nil {
var routes []string
// lookup the route to send the reques to
// TODO apply any filtering here
routes, err = g.opts.Lookup(ctx, req, callOpts)
if err != nil {
return errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err = callOpts.Selector.Select(routes)
if err != nil {
return err
}
}
// get the next node // get the next node
node := next() node := next()
@ -611,31 +514,6 @@ func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interfa
} }
func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) { func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
ts := time.Now()
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
var sp tracer.Span
ctx, sp = g.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
tracer.WithSpanKind(tracer.SpanKindClient),
tracer.WithSpanLabels("endpoint", req.Endpoint()),
)
stream, err := g.funcStream(ctx, req, opts...)
g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
te := time.Since(ts)
g.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
g.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
if me := status.Convert(err); me == nil {
sp.Finish()
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(codes.OK))).Inc()
} else {
sp.SetStatus(tracer.SpanStatusError, err.Error())
g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code()))).Inc()
}
return stream, err
}
func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
// make a copy of call opts // make a copy of call opts
callOpts := g.opts.CallOptions callOpts := g.opts.CallOptions
for _, opt := range opts { for _, opt := range opts {
@ -654,6 +532,11 @@ func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...c
// make a copy of stream // make a copy of stream
gstream := g.stream gstream := g.stream
// wrap the call in reverse
for i := len(callOpts.CallWrappers); i > 0; i-- {
gstream = callOpts.CallWrappers[i-1](gstream)
}
// use the router passed as a call option, or fallback to the rpc clients router // use the router passed as a call option, or fallback to the rpc clients router
if callOpts.Router == nil { if callOpts.Router == nil {
callOpts.Router = g.opts.Router callOpts.Router = g.opts.Router
@ -669,7 +552,18 @@ func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...c
callOpts.Address = []string{g.opts.Proxy} callOpts.Address = []string{g.opts.Proxy}
} }
var next selector.Next // lookup the route to send the reques to
// TODO: move to internal lookup func
routes, err := g.opts.Lookup(ctx, req, callOpts)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err := callOpts.Selector.Select(routes)
if err != nil {
return nil, err
}
call := func(i int) (client.Stream, error) { call := func(i int) (client.Stream, error) {
// call backoff first. Someone may want an initial start delay // call backoff first. Someone may want an initial start delay
@ -683,23 +577,6 @@ func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...c
time.Sleep(t) time.Sleep(t)
} }
if next == nil {
var routes []string
// lookup the route to send the reques to
// TODO apply any filtering here
routes, err = g.opts.Lookup(ctx, req, callOpts)
if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error())
}
// balance the list of nodes
next, err = callOpts.Selector.Select(routes)
if err != nil {
return nil, err
}
}
// get the next node // get the next node
node := next() node := next()
@ -764,18 +641,10 @@ func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...c
} }
func (g *grpcClient) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error { func (g *grpcClient) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
return g.funcBatchPublish(ctx, ps, opts...)
}
func (g *grpcClient) fnBatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
return g.publish(ctx, ps, opts...) return g.publish(ctx, ps, opts...)
} }
func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error { func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
return g.funcPublish(ctx, p, opts...)
}
func (g *grpcClient) fnPublish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
return g.publish(ctx, []client.Message{p}, opts...) return g.publish(ctx, []client.Message{p}, opts...)
} }
@ -789,10 +658,6 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
if v, ok := os.LookupEnv("MICRO_PROXY"); ok { if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
exchange = v exchange = v
} }
// get the exchange
if len(options.Exchange) > 0 {
exchange = options.Exchange
}
msgs := make([]*broker.Message, 0, len(ps)) msgs := make([]*broker.Message, 0, len(ps))
@ -804,16 +669,7 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
for _, p := range ps { for _, p := range ps {
md := metadata.Copy(omd) md := metadata.Copy(omd)
md[metadata.HeaderContentType] = p.ContentType() md[metadata.HeaderContentType] = p.ContentType()
topic := p.Topic() md[metadata.HeaderTopic] = p.Topic()
if len(exchange) > 0 {
topic = exchange
}
md.Set(metadata.HeaderTopic, topic)
iter := p.Metadata().Iterator()
var k, v string
for iter.Next(&k, &v) {
md.Set(k, v)
}
// passed in raw data // passed in raw data
if d, ok := p.Payload().(*codec.Frame); ok { if d, ok := p.Payload().(*codec.Frame); ok {
@ -831,6 +687,13 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
} }
body = b body = b
} }
topic := p.Topic()
if len(exchange) > 0 {
topic = exchange
}
md.Set(metadata.HeaderTopic, topic)
msgs = append(msgs, &broker.Message{Header: md, Body: body}) msgs = append(msgs, &broker.Message{Header: md, Body: body})
} }
@ -848,45 +711,41 @@ func (g *grpcClient) Name() string {
return g.opts.Name return g.opts.Name
} }
func (g *grpcClient) getGrpcDialOptions(ctx context.Context) []grpc.DialOption { func (g *grpcClient) getGrpcDialOptions() []grpc.DialOption {
var opts []grpc.DialOption if g.opts.CallOptions.Context == nil {
return nil
if g.opts.CallOptions.Context != nil {
if v := g.opts.CallOptions.Context.Value(grpcDialOptions{}); v != nil {
if vopts, ok := v.([]grpc.DialOption); ok {
opts = append(opts, vopts...)
}
}
} }
if ctx != nil { v := g.opts.CallOptions.Context.Value(grpcDialOptions{})
if v := ctx.Value(grpcDialOptions{}); v != nil {
if vopts, ok := v.([]grpc.DialOption); ok { if v == nil {
opts = append(opts, vopts...) return nil
}
} }
opts, ok := v.([]grpc.DialOption)
if !ok {
return nil
} }
return opts return opts
} }
func (g *grpcClient) getGrpcCallOptions(ctx context.Context) []grpc.CallOption { func (g *grpcClient) getGrpcCallOptions() []grpc.CallOption {
var opts []grpc.CallOption if g.opts.CallOptions.Context == nil {
return nil
if g.opts.CallOptions.Context != nil {
if v := g.opts.CallOptions.Context.Value(grpcCallOptions{}); v != nil {
if vopts, ok := v.([]grpc.CallOption); ok {
opts = append(opts, vopts...)
}
}
} }
if ctx != nil { v := g.opts.CallOptions.Context.Value(grpcCallOptions{})
if v := ctx.Value(grpcCallOptions{}); v != nil {
if vopts, ok := v.([]grpc.CallOption); ok { if v == nil {
opts = append(opts, vopts...) return nil
}
} }
opts, ok := v.([]grpc.CallOption)
if !ok {
return nil
} }
return opts return opts
@ -895,20 +754,25 @@ func (g *grpcClient) getGrpcCallOptions(ctx context.Context) []grpc.CallOption {
func NewClient(opts ...client.Option) client.Client { func NewClient(opts ...client.Option) client.Client {
options := client.NewOptions(opts...) options := client.NewOptions(opts...)
// default content type for grpc // default content type for grpc
if options.ContentType == "" { options.ContentType = defaultContentType
options.ContentType = DefaultContentType
}
c := &grpcClient{ rc := &grpcClient{
opts: options, opts: options,
} }
c.pool = NewConnPool(options.PoolSize, options.PoolTTL, c.poolMaxIdle(), c.poolMaxStreams()) rc.pool = newPool(options.PoolSize, options.PoolTTL, rc.poolMaxIdle(), rc.poolMaxStreams())
if c.opts.Context != nil { c := client.Client(rc)
if codecs, ok := c.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
// wrap in reverse
for i := len(options.Wrappers); i > 0; i-- {
c = options.Wrappers[i-1](c)
}
if rc.opts.Context != nil {
if codecs, ok := rc.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
for k, v := range codecs { for k, v := range codecs {
c.opts.Codecs[k] = &wrapGrpcCodec{v} rc.opts.Codecs[k] = &wrapGrpcCodec{v}
} }
} }
} }
@ -917,10 +781,5 @@ func NewClient(opts ...client.Option) client.Client {
encoding.RegisterCodec(&wrapMicroCodec{k}) encoding.RegisterCodec(&wrapMicroCodec{k})
} }
c.funcCall = c.fnCall
c.funcStream = c.fnStream
c.funcPublish = c.fnPublish
c.funcBatchPublish = c.fnBatchPublish
return c return c
} }

40
grpc_pool.go
View File

@ -2,7 +2,6 @@ package grpc
import ( import (
"context" "context"
"strings"
"sync" "sync"
"time" "time"
@ -10,7 +9,7 @@ import (
"google.golang.org/grpc/connectivity" "google.golang.org/grpc/connectivity"
) )
type ConnPool struct { type pool struct {
conns map[string]*streamsPool conns map[string]*streamsPool
size int size int
ttl int64 ttl int64
@ -21,36 +20,36 @@ type ConnPool struct {
type streamsPool struct { type streamsPool struct {
// head of list // head of list
head *PoolConn head *poolConn
// busy conns list // busy conns list
busy *PoolConn busy *poolConn
// the siza of list // the siza of list
count int count int
// idle conn // idle conn
idle int idle int
} }
type PoolConn struct { type poolConn struct {
err error err error
*grpc.ClientConn *grpc.ClientConn
next *PoolConn next *poolConn
pool *ConnPool pool *pool
sp *streamsPool sp *streamsPool
pre *PoolConn pre *poolConn
addr string addr string
streams int streams int
created int64 created int64
in bool in bool
} }
func NewConnPool(size int, ttl time.Duration, idle int, ms int) *ConnPool { func newPool(size int, ttl time.Duration, idle int, ms int) *pool {
if ms <= 0 { if ms <= 0 {
ms = 1 ms = 1
} }
if idle < 0 { if idle < 0 {
idle = 0 idle = 0
} }
return &ConnPool{ return &pool{
size: size, size: size,
ttl: int64(ttl.Seconds()), ttl: int64(ttl.Seconds()),
maxStreams: ms, maxStreams: ms,
@ -59,15 +58,12 @@ func NewConnPool(size int, ttl time.Duration, idle int, ms int) *ConnPool {
} }
} }
func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption) (*PoolConn, error) { func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption) (*poolConn, error) {
if strings.HasPrefix(addr, "http") {
addr = addr[strings.Index(addr, ":")+3:]
}
now := time.Now().Unix() now := time.Now().Unix()
p.Lock() p.Lock()
sp, ok := p.conns[addr] sp, ok := p.conns[addr]
if !ok { if !ok {
sp = &streamsPool{head: &PoolConn{}, busy: &PoolConn{}, count: 0, idle: 0} sp = &streamsPool{head: &poolConn{}, busy: &poolConn{}, count: 0, idle: 0}
p.conns[addr] = sp p.conns[addr] = sp
} }
// while we have conns check streams and then return one // while we have conns check streams and then return one
@ -130,12 +126,12 @@ func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption
} }
p.Unlock() p.Unlock()
// create new conn) // create new conn
cc, err := grpc.DialContext(ctx, addr, opts...) cc, err := grpc.DialContext(ctx, addr, opts...)
if err != nil { if err != nil {
return nil, err return nil, err
} }
conn = &PoolConn{ClientConn: cc, err: nil, addr: addr, pool: p, sp: sp, streams: 1, created: time.Now().Unix(), pre: nil, next: nil, in: false} conn = &poolConn{ClientConn: cc, err: nil, addr: addr, pool: p, sp: sp, streams: 1, created: time.Now().Unix(), pre: nil, next: nil, in: false}
// add conn to streams pool // add conn to streams pool
p.Lock() p.Lock()
@ -147,7 +143,7 @@ func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption
return conn, nil return conn, nil
} }
func (p *ConnPool) Put(conn *PoolConn, err error) { func (p *pool) release(conn *poolConn, err error) {
p.Lock() p.Lock()
p, sp, created := conn.pool, conn.sp, conn.created p, sp, created := conn.pool, conn.sp, conn.created
// try to add conn // try to add conn
@ -182,11 +178,11 @@ func (p *ConnPool) Put(conn *PoolConn, err error) {
p.Unlock() p.Unlock()
} }
func (conn *PoolConn) Close() { func (conn *poolConn) Close() {
conn.pool.Put(conn, conn.err) conn.pool.release(conn, conn.err)
} }
func removeConn(conn *PoolConn) { func removeConn(conn *poolConn) {
if conn.pre != nil { if conn.pre != nil {
conn.pre.next = conn.next conn.pre.next = conn.next
} }
@ -199,7 +195,7 @@ func removeConn(conn *PoolConn) {
conn.sp.count-- conn.sp.count--
} }
func addConnAfter(conn *PoolConn, after *PoolConn) { func addConnAfter(conn *poolConn, after *poolConn) {
conn.next = after.next conn.next = after.next
conn.pre = after conn.pre = after
if after.next != nil { if after.next != nil {

28
options.go
View File

@ -10,7 +10,7 @@ import (
) )
var ( var (
// DefaultPoolMaxStreams maximum streams on a connection // DefaultPoolMaxStreams maximum streams on a connectioin
// (20) // (20)
DefaultPoolMaxStreams = 20 DefaultPoolMaxStreams = 20
@ -25,9 +25,6 @@ var (
// DefaultMaxSendMsgSize maximum message that client can send // DefaultMaxSendMsgSize maximum message that client can send
// (4 MB). // (4 MB).
DefaultMaxSendMsgSize = 1024 * 1024 * 4 DefaultMaxSendMsgSize = 1024 * 1024 * 4
// DefaultServiceConfig enable load balancing
DefaultServiceConfig = `{"loadBalancingPolicy":"round_robin"}`
) )
type poolMaxStreams struct{} type poolMaxStreams struct{}
@ -73,7 +70,9 @@ func Codec(contentType string, c encoding.Codec) client.Option {
type maxRecvMsgSizeKey struct{} type maxRecvMsgSizeKey struct{}
//
// MaxRecvMsgSize set the maximum size of message that client can receive. // MaxRecvMsgSize set the maximum size of message that client can receive.
//
func MaxRecvMsgSize(s int) client.Option { func MaxRecvMsgSize(s int) client.Option {
return func(o *client.Options) { return func(o *client.Options) {
if o.Context == nil { if o.Context == nil {
@ -85,7 +84,9 @@ func MaxRecvMsgSize(s int) client.Option {
type maxSendMsgSizeKey struct{} type maxSendMsgSizeKey struct{}
//
// MaxSendMsgSize set the maximum size of message that client can send. // MaxSendMsgSize set the maximum size of message that client can send.
//
func MaxSendMsgSize(s int) client.Option { func MaxSendMsgSize(s int) client.Option {
return func(o *client.Options) { return func(o *client.Options) {
if o.Context == nil { if o.Context == nil {
@ -97,9 +98,11 @@ func MaxSendMsgSize(s int) client.Option {
type grpcDialOptions struct{} type grpcDialOptions struct{}
//
// DialOptions to be used to configure gRPC dial options // DialOptions to be used to configure gRPC dial options
func DialOptions(opts ...grpc.DialOption) client.Option { //
return func(o *client.Options) { func DialOptions(opts ...grpc.DialOption) client.CallOption {
return func(o *client.CallOptions) {
if o.Context == nil { if o.Context == nil {
o.Context = context.Background() o.Context = context.Background()
} }
@ -109,7 +112,9 @@ func DialOptions(opts ...grpc.DialOption) client.Option {
type grpcCallOptions struct{} type grpcCallOptions struct{}
//
// CallOptions to be used to configure gRPC call options // CallOptions to be used to configure gRPC call options
//
func CallOptions(opts ...grpc.CallOption) client.CallOption { func CallOptions(opts ...grpc.CallOption) client.CallOption {
return func(o *client.CallOptions) { return func(o *client.CallOptions) {
if o.Context == nil { if o.Context == nil {
@ -118,14 +123,3 @@ func CallOptions(opts ...grpc.CallOption) client.CallOption {
o.Context = context.WithValue(o.Context, grpcCallOptions{}, opts) o.Context = context.WithValue(o.Context, grpcCallOptions{}, opts)
} }
} }
type serviceConfigKey struct{}
func ServiceConfig(str string) client.CallOption {
return func(options *client.CallOptions) {
if options.Context == nil {
options.Context = context.Background()
}
options.Context = context.WithValue(options.Context, serviceConfigKey{}, str)
}
}

13
request.go
View File

@ -38,12 +38,15 @@ func methodToGRPC(service, method string) string {
return fmt.Sprintf("/%s.%s/%s", service, mParts[0], mParts[1]) return fmt.Sprintf("/%s.%s/%s", service, mParts[0], mParts[1])
} }
func newGRPCRequest(service, method string, request interface{}, contentType string, opts ...client.RequestOption) client.Request { func newGRPCRequest(service, method string, request interface{}, contentType string, reqOpts ...client.RequestOption) client.Request {
options := client.NewRequestOptions(opts...) var opts client.RequestOptions
for _, o := range reqOpts {
o(&opts)
}
// set the content-type specified // set the content-type specified
if len(options.ContentType) > 0 { if len(opts.ContentType) > 0 {
contentType = options.ContentType contentType = opts.ContentType
} }
return &grpcRequest{ return &grpcRequest{
@ -51,7 +54,7 @@ func newGRPCRequest(service, method string, request interface{}, contentType str
method: method, method: method,
request: request, request: request,
contentType: contentType, contentType: contentType,
opts: options, opts: opts,
} }
} }

10
response.go
View File

@ -9,7 +9,7 @@ import (
) )
type response struct { type response struct {
conn *PoolConn conn *poolConn
stream grpc.ClientStream stream grpc.ClientStream
codec codec.Codec codec codec.Codec
} }
@ -23,7 +23,7 @@ func (r *response) Codec() codec.Codec {
func (r *response) Header() metadata.Metadata { func (r *response) Header() metadata.Metadata {
meta, err := r.stream.Header() meta, err := r.stream.Header()
if err != nil { if err != nil {
return nil return metadata.New(0)
} }
md := metadata.New(len(meta)) md := metadata.New(len(meta))
for k, v := range meta { for k, v := range meta {
@ -34,5 +34,9 @@ func (r *response) Header() metadata.Metadata {
// Read the undecoded response // Read the undecoded response
func (r *response) Read() ([]byte, error) { func (r *response) Read() ([]byte, error) {
return nil, nil f := &codec.Frame{}
if err := r.codec.ReadBody(&wrapStream{r.stream}, f); err != nil {
return nil, err
}
return f.Data, nil
} }

49
stream.go
View File

@ -6,7 +6,6 @@ import (
"sync" "sync"
"go.unistack.org/micro/v3/client" "go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/tracer"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
@ -18,7 +17,7 @@ type grpcStream struct {
request client.Request request client.Request
response client.Response response client.Response
close func(err error) close func(err error)
conn *PoolConn conn *poolConn
sync.RWMutex sync.RWMutex
closed bool closed bool
} }
@ -43,14 +42,6 @@ func (g *grpcStream) Send(msg interface{}) error {
return nil return nil
} }
func (g *grpcStream) SendMsg(msg interface{}) error {
if err := g.ClientStream.SendMsg(msg); err != nil {
g.setError(err)
return err
}
return nil
}
func (g *grpcStream) Recv(msg interface{}) (err error) { func (g *grpcStream) Recv(msg interface{}) (err error) {
defer g.setError(err) defer g.setError(err)
@ -69,24 +60,6 @@ func (g *grpcStream) Recv(msg interface{}) (err error) {
return return
} }
func (g *grpcStream) RecvMsg(msg interface{}) (err error) {
defer g.setError(err)
if err = g.ClientStream.RecvMsg(msg); err != nil {
// #202 - inconsistent gRPC stream behavior
// the only way to tell if the stream is done is when we get a EOF on the Recv
// here we should close the underlying gRPC ClientConn
closeErr := g.Close()
if err == io.EOF && closeErr != nil {
err = closeErr
}
return err
}
return
}
func (g *grpcStream) Error() error { func (g *grpcStream) Error() error {
g.RLock() g.RLock()
defer g.RUnlock() defer g.RUnlock()
@ -112,26 +85,6 @@ func (g *grpcStream) Close() error {
return nil return nil
} }
if sp, ok := tracer.SpanFromContext(g.context); ok && sp != nil {
if g.err != nil {
sp.SetStatus(tracer.SpanStatusError, g.err.Error())
}
sp.Finish()
}
// close the connection
g.closed = true
g.close(g.err)
return g.ClientStream.CloseSend()
}
func (g *grpcStream) CloseSend() error {
g.Lock()
defer g.Unlock()
if g.closed {
return nil
}
// close the connection // close the connection
g.closed = true g.closed = true
g.close(g.err) g.close(g.err)