unistack-org/micro-broker-service
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity

Compare commits

base: unistack-org:v3
Branches Tags
unistack-org:v3
unistack-org:dependabot/go_modules/google.golang.org/grpc-1.53.0
unistack-org:master
unistack-org:upstream
unistack-org:v3.8.3
unistack-org:v3.8.2
unistack-org:v3.8.1
unistack-org:v3.8.0
unistack-org:v3.3.1
unistack-org:v3.3.0
unistack-org:v3.2.4
unistack-org:v3.1.3
..
compare: unistack-org:v3.8.0
Branches Tags
unistack-org:v3
unistack-org:dependabot/go_modules/google.golang.org/grpc-1.53.0
unistack-org:master
unistack-org:upstream
unistack-org:v3.8.3
unistack-org:v3.8.2
unistack-org:v3.8.1
unistack-org:v3.8.0
unistack-org:v3.3.1
unistack-org:v3.3.0
unistack-org:v3.2.4
unistack-org:v3.1.3

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

17 changed files with 598 additions and 752 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.

2
generate.go
View File

@ -1,3 +1,3 @@
package service package service
//go:generate protoc -I./proto -I. --go_out=paths=source_relative:./proto --go-micro_out=components=micro|grpc,standalone=true,paths=source_relative:./micro proto/broker.proto //go:generate protoc -I./proto -I. --go-grpc_out=paths=source_relative:./proto --go_out=paths=source_relative:./proto --go-micro_out=components=micro|rpc,standalone=true,paths=source_relative:./micro proto/broker.proto

20
go.mod
View File

@ -1,20 +1,10 @@
module go.unistack.org/micro-broker-service/v3 module go.unistack.org/micro-broker-service/v3
go 1.21 go 1.16
toolchain go1.23.1
require ( require (
go.unistack.org/micro-server-http/v3 v3.11.29 github.com/golang/protobuf v1.5.1 // indirect
go.unistack.org/micro/v3 v3.10.91 go.unistack.org/micro/v3 v3.8.6
google.golang.org/protobuf v1.34.2 google.golang.org/grpc v1.40.0
) google.golang.org/protobuf v1.27.1
require (
github.com/google/go-cmp v0.6.0 // indirect
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
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
google.golang.org/grpc v1.67.0 // indirect
) )

155
go.sum
View File

@ -1,28 +1,135 @@
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/ef-ds/deque v1.0.4/go.mod h1:gXDnTC3yqvBcHbq2lcExjtAcVrOnJCbMcZXmuj8Z4tg=
github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
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/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.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/golang/protobuf v1.5.1 h1:jAbXjIeW2ZSW2AwFxlGTDoc2CjI2XujLkV3ArsZFCvc=
github.com/golang/protobuf v1.5.1/go.mod h1:DopwsBzvsk0Fs44TXzsVbJyPhcCPeIwnvohx4u74HPM=
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/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q= github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo= github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
go.unistack.org/micro-server-http/v3 v3.10.14 h1:yWeT/Ljag/ueiV6eSK3Xi620b+HzPOgZLCpwx+tQxns= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
go.unistack.org/micro-server-http/v3 v3.11.29 h1:IncjMVunK3I4yJ8fIcUjy5ttEVaXskfGb6QXYOV3b5M= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
go.unistack.org/micro-server-http/v3 v3.11.29/go.mod h1:GaVKQVMSeHlcBvL2l1kHRxOWuSJ/l+uhUCLHYGidaE0= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
go.unistack.org/micro/v3 v3.10.23 h1:4BE7NwwyJbCWOfzjzztamBxJSgRHHW1uQtMGNDLHG3s= github.com/silas/dag v0.0.0-20210626123444-3804bac2d6d4/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
go.unistack.org/micro/v3 v3.10.91 h1:vuJY4tXwpqimwIkEJ3TozMYNVQQs+C5QMlQWPgSY/YM= github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
go.unistack.org/micro/v3 v3.10.91/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo= go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0= go.unistack.org/micro-proto/v3 v3.1.0 h1:q39FwjFiRZn+Ux/tt+d3bJTmDtsQQWa+3SLYVo1vLfA=
golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34= go.unistack.org/micro-proto/v3 v3.1.0/go.mod h1:DpRhYCBXlmSJ/AAXTmntvlh7kQkYU6eFvlmYAx4BQS8=
golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= go.unistack.org/micro/v3 v3.8.6 h1:nU6dpNRF7iF8dtcTTfzdPan8RkhiDuTsfMrXBmpzG/g=
go.unistack.org/micro/v3 v3.8.6/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-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/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-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
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/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-20210423082822-04245dca01da h1:b3NXsE2LusjYGGjL5bxEVZZORm/YEFFrWFjR8eFrw/c=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
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.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/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/grpc v1.67.0 h1:IdH9y6PF5MPSdAntIcpjQ+tXO41pcQsfZV2RxtQgVcw= google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
google.golang.org/grpc v1.67.0/go.mod h1:1gLDyUQU7CTLJI90u3nXZ9ekeghjeM7pTDZlqFNg2AA= 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 h1:+kGHl1aib/qcwaRi1CbqBZ1rk19r85MNUf8HaBghugY=
google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
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.40.0 h1:AGJ0Ih4mHjSeibYkFGh1dD9KJ/eOtZ93I6hoHhukQ5Q=
google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
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-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= 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=
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=

53
micro/broker_micro.pb.go
View File

@ -1,7 +1,5 @@
// Code generated by protoc-gen-go-micro. DO NOT EDIT. // Code generated by protoc-gen-go-micro. DO NOT EDIT.
// versions: // protoc-gen-go-micro version: v3.5.3
// - protoc-gen-go-micro v3.10.2
// - protoc v3.21.12
// source: broker.proto // source: broker.proto
package servicepb package servicepb
@ -9,25 +7,26 @@ package servicepb
import ( import (
context "context" context "context"
proto "go.unistack.org/micro-broker-service/v3/proto" proto "go.unistack.org/micro-broker-service/v3/proto"
v3 "go.unistack.org/micro-server-http/v3" api "go.unistack.org/micro/v3/api"
client "go.unistack.org/micro/v3/client" client "go.unistack.org/micro/v3/client"
) )
var ( var (
BrokerServiceName = "BrokerService" BrokerName = "Broker"
)
var ( BrokerEndpoints = []api.Endpoint{}
BrokerServiceServerEndpoints = []v3.EndpointMetadata{}
) )
type BrokerServiceClient interface { func NewBrokerEndpoints() []api.Endpoint {
Publish(ctx context.Context, req *proto.PublishRequest, opts ...client.CallOption) (*proto.PublishResponse, error) return BrokerEndpoints
BatchPublish(ctx context.Context, req *proto.BatchPublishRequest, opts ...client.CallOption) (*proto.BatchPublishResponse, error)
Subscribe(ctx context.Context, req *proto.SubscribeRequest, opts ...client.CallOption) (BrokerService_SubscribeClient, error)
BatchSubscribe(ctx context.Context, req *proto.BatchSubscribeRequest, opts ...client.CallOption) (BrokerService_BatchSubscribeClient, error)
} }
type BrokerService_SubscribeClient interface { type BrokerClient interface {
Publish(ctx context.Context, req *proto.PublishRequest, opts ...client.CallOption) (*proto.Empty, error)
Subscribe(ctx context.Context, req *proto.SubscribeRequest, opts ...client.CallOption) (Broker_SubscribeClient, error)
}
type Broker_SubscribeClient interface {
Context() context.Context Context() context.Context
SendMsg(msg interface{}) error SendMsg(msg interface{}) error
RecvMsg(msg interface{}) error RecvMsg(msg interface{}) error
@ -35,30 +34,12 @@ type BrokerService_SubscribeClient interface {
Recv() (*proto.Message, error) Recv() (*proto.Message, error)
} }
type BrokerService_BatchSubscribeClient interface { type BrokerServer interface {
Context() context.Context Publish(ctx context.Context, req *proto.PublishRequest, rsp *proto.Empty) error
SendMsg(msg interface{}) error Subscribe(ctx context.Context, req *proto.SubscribeRequest, stream Broker_SubscribeStream) error
RecvMsg(msg interface{}) error
Close() error
Recv() (*proto.Message, error)
} }
type BrokerServiceServer interface { type Broker_SubscribeStream interface {
Publish(ctx context.Context, req *proto.PublishRequest, rsp *proto.PublishResponse) error
BatchPublish(ctx context.Context, req *proto.BatchPublishRequest, rsp *proto.BatchPublishResponse) error
Subscribe(ctx context.Context, req *proto.SubscribeRequest, stream BrokerService_SubscribeStream) error
BatchSubscribe(ctx context.Context, req *proto.BatchSubscribeRequest, stream BrokerService_BatchSubscribeStream) error
}
type BrokerService_SubscribeStream interface {
Context() context.Context
SendMsg(msg interface{}) error
RecvMsg(msg interface{}) error
Close() error
Send(msg *proto.Message) error
}
type BrokerService_BatchSubscribeStream interface {
Context() context.Context Context() context.Context
SendMsg(msg interface{}) error SendMsg(msg interface{}) error
RecvMsg(msg interface{}) error RecvMsg(msg interface{}) error

216
micro/broker_micro_grpc.pb.go
View File

@ -1,216 +0,0 @@
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
// protoc-gen-go-micro version: v3.10.2
// source: broker.proto
package servicepb
import (
context "context"
proto "go.unistack.org/micro-broker-service/v3/proto"
client "go.unistack.org/micro/v3/client"
server "go.unistack.org/micro/v3/server"
)
type brokerServiceClient struct {
c client.Client
name string
}
func NewBrokerServiceClient(name string, c client.Client) BrokerServiceClient {
return &brokerServiceClient{c: c, name: name}
}
func (c *brokerServiceClient) Publish(ctx context.Context, req *proto.PublishRequest, opts ...client.CallOption) (*proto.PublishResponse, error) {
rsp := &proto.PublishResponse{}
err := c.c.Call(ctx, c.c.NewRequest(c.name, "BrokerService.Publish", req), rsp, opts...)
if err != nil {
return nil, err
}
return rsp, nil
}
func (c *brokerServiceClient) BatchPublish(ctx context.Context, req *proto.BatchPublishRequest, opts ...client.CallOption) (*proto.BatchPublishResponse, error) {
rsp := &proto.BatchPublishResponse{}
err := c.c.Call(ctx, c.c.NewRequest(c.name, "BrokerService.BatchPublish", req), rsp, opts...)
if err != nil {
return nil, err
}
return rsp, nil
}
func (c *brokerServiceClient) Subscribe(ctx context.Context, req *proto.SubscribeRequest, opts ...client.CallOption) (BrokerService_SubscribeClient, error) {
stream, err := c.c.Stream(ctx, c.c.NewRequest(c.name, "BrokerService.Subscribe", &proto.SubscribeRequest{}), opts...)
if err != nil {
return nil, err
}
if err := stream.Send(req); err != nil {
return nil, err
}
return &brokerServiceClientSubscribe{stream}, nil
}
type brokerServiceClientSubscribe struct {
stream client.Stream
}
func (s *brokerServiceClientSubscribe) Close() error {
return s.stream.Close()
}
func (s *brokerServiceClientSubscribe) CloseSend() error {
return s.stream.CloseSend()
}
func (s *brokerServiceClientSubscribe) Context() context.Context {
return s.stream.Context()
}
func (s *brokerServiceClientSubscribe) SendMsg(msg interface{}) error {
return s.stream.Send(msg)
}
func (s *brokerServiceClientSubscribe) RecvMsg(msg interface{}) error {
return s.stream.Recv(msg)
}
func (s *brokerServiceClientSubscribe) Recv() (*proto.Message, error) {
msg := &proto.Message{}
if err := s.stream.Recv(msg); err != nil {
return nil, err
}
return msg, nil
}
func (c *brokerServiceClient) BatchSubscribe(ctx context.Context, req *proto.BatchSubscribeRequest, opts ...client.CallOption) (BrokerService_BatchSubscribeClient, error) {
stream, err := c.c.Stream(ctx, c.c.NewRequest(c.name, "BrokerService.BatchSubscribe", &proto.BatchSubscribeRequest{}), opts...)
if err != nil {
return nil, err
}
if err := stream.Send(req); err != nil {
return nil, err
}
return &brokerServiceClientBatchSubscribe{stream}, nil
}
type brokerServiceClientBatchSubscribe struct {
stream client.Stream
}
func (s *brokerServiceClientBatchSubscribe) Close() error {
return s.stream.Close()
}
func (s *brokerServiceClientBatchSubscribe) CloseSend() error {
return s.stream.CloseSend()
}
func (s *brokerServiceClientBatchSubscribe) Context() context.Context {
return s.stream.Context()
}
func (s *brokerServiceClientBatchSubscribe) SendMsg(msg interface{}) error {
return s.stream.Send(msg)
}
func (s *brokerServiceClientBatchSubscribe) RecvMsg(msg interface{}) error {
return s.stream.Recv(msg)
}
func (s *brokerServiceClientBatchSubscribe) Recv() (*proto.Message, error) {
msg := &proto.Message{}
if err := s.stream.Recv(msg); err != nil {
return nil, err
}
return msg, nil
}
type brokerServiceServer struct {
BrokerServiceServer
}
func (h *brokerServiceServer) Publish(ctx context.Context, req *proto.PublishRequest, rsp *proto.PublishResponse) error {
return h.BrokerServiceServer.Publish(ctx, req, rsp)
}
func (h *brokerServiceServer) BatchPublish(ctx context.Context, req *proto.BatchPublishRequest, rsp *proto.BatchPublishResponse) error {
return h.BrokerServiceServer.BatchPublish(ctx, req, rsp)
}
func (h *brokerServiceServer) Subscribe(ctx context.Context, stream server.Stream) error {
msg := &proto.SubscribeRequest{}
if err := stream.Recv(msg); err != nil {
return err
}
return h.BrokerServiceServer.Subscribe(ctx, msg, &brokerServiceSubscribeStream{stream})
}
type brokerServiceSubscribeStream struct {
stream server.Stream
}
func (s *brokerServiceSubscribeStream) Close() error {
return s.stream.Close()
}
func (s *brokerServiceSubscribeStream) Context() context.Context {
return s.stream.Context()
}
func (s *brokerServiceSubscribeStream) SendMsg(msg interface{}) error {
return s.stream.Send(msg)
}
func (s *brokerServiceSubscribeStream) RecvMsg(msg interface{}) error {
return s.stream.Recv(msg)
}
func (s *brokerServiceSubscribeStream) Send(msg *proto.Message) error {
return s.stream.Send(msg)
}
func (h *brokerServiceServer) BatchSubscribe(ctx context.Context, stream server.Stream) error {
msg := &proto.BatchSubscribeRequest{}
if err := stream.Recv(msg); err != nil {
return err
}
return h.BrokerServiceServer.BatchSubscribe(ctx, msg, &brokerServiceBatchSubscribeStream{stream})
}
type brokerServiceBatchSubscribeStream struct {
stream server.Stream
}
func (s *brokerServiceBatchSubscribeStream) Close() error {
return s.stream.Close()
}
func (s *brokerServiceBatchSubscribeStream) Context() context.Context {
return s.stream.Context()
}
func (s *brokerServiceBatchSubscribeStream) SendMsg(msg interface{}) error {
return s.stream.Send(msg)
}
func (s *brokerServiceBatchSubscribeStream) RecvMsg(msg interface{}) error {
return s.stream.Recv(msg)
}
func (s *brokerServiceBatchSubscribeStream) Send(msg *proto.Message) error {
return s.stream.Send(msg)
}
func RegisterBrokerServiceServer(s server.Server, sh BrokerServiceServer, opts ...server.HandlerOption) error {
type brokerService interface {
Publish(ctx context.Context, req *proto.PublishRequest, rsp *proto.PublishResponse) error
BatchPublish(ctx context.Context, req *proto.BatchPublishRequest, rsp *proto.BatchPublishResponse) error
Subscribe(ctx context.Context, stream server.Stream) error
BatchSubscribe(ctx context.Context, stream server.Stream) error
}
type BrokerService struct {
brokerService
}
h := &brokerServiceServer{sh}
var nopts []server.HandlerOption
return s.Handle(s.NewHandler(&BrokerService{h}, append(nopts, opts...)...))
}

126
micro/broker_micro_rpc.pb.go Normal file
View File

@ -0,0 +1,126 @@
// Code generated by protoc-gen-go-micro. DO NOT EDIT.
// protoc-gen-go-micro version: v3.5.3
// source: broker.proto
package servicepb
import (
context "context"
proto "go.unistack.org/micro-broker-service/v3/proto"
api "go.unistack.org/micro/v3/api"
client "go.unistack.org/micro/v3/client"
server "go.unistack.org/micro/v3/server"
)
type brokerClient struct {
c client.Client
name string
}
func NewBrokerClient(name string, c client.Client) BrokerClient {
return &brokerClient{c: c, name: name}
}
func (c *brokerClient) Publish(ctx context.Context, req *proto.PublishRequest, opts ...client.CallOption) (*proto.Empty, error) {
rsp := &proto.Empty{}
err := c.c.Call(ctx, c.c.NewRequest(c.name, "Broker.Publish", req), rsp, opts...)
if err != nil {
return nil, err
}
return rsp, nil
}
func (c *brokerClient) Subscribe(ctx context.Context, req *proto.SubscribeRequest, opts ...client.CallOption) (Broker_SubscribeClient, error) {
stream, err := c.c.Stream(ctx, c.c.NewRequest(c.name, "Broker.Subscribe", &proto.SubscribeRequest{}), opts...)
if err != nil {
return nil, err
}
if err := stream.Send(req); err != nil {
return nil, err
}
return &brokerClientSubscribe{stream}, nil
}
type brokerClientSubscribe struct {
stream client.Stream
}
func (s *brokerClientSubscribe) Close() error {
return s.stream.Close()
}
func (s *brokerClientSubscribe) Context() context.Context {
return s.stream.Context()
}
func (s *brokerClientSubscribe) SendMsg(msg interface{}) error {
return s.stream.Send(msg)
}
func (s *brokerClientSubscribe) RecvMsg(msg interface{}) error {
return s.stream.Recv(msg)
}
func (s *brokerClientSubscribe) Recv() (*proto.Message, error) {
msg := &proto.Message{}
if err := s.stream.Recv(msg); err != nil {
return nil, err
}
return msg, nil
}
type brokerServer struct {
BrokerServer
}
func (h *brokerServer) Publish(ctx context.Context, req *proto.PublishRequest, rsp *proto.Empty) error {
return h.BrokerServer.Publish(ctx, req, rsp)
}
func (h *brokerServer) Subscribe(ctx context.Context, stream server.Stream) error {
msg := &proto.SubscribeRequest{}
if err := stream.Recv(msg); err != nil {
return err
}
return h.BrokerServer.Subscribe(ctx, msg, &brokerSubscribeStream{stream})
}
type brokerSubscribeStream struct {
stream server.Stream
}
func (s *brokerSubscribeStream) Close() error {
return s.stream.Close()
}
func (s *brokerSubscribeStream) Context() context.Context {
return s.stream.Context()
}
func (s *brokerSubscribeStream) SendMsg(msg interface{}) error {
return s.stream.Send(msg)
}
func (s *brokerSubscribeStream) RecvMsg(msg interface{}) error {
return s.stream.Recv(msg)
}
func (s *brokerSubscribeStream) Send(msg *proto.Message) error {
return s.stream.Send(msg)
}
func RegisterBrokerServer(s server.Server, sh BrokerServer, opts ...server.HandlerOption) error {
type broker interface {
Publish(ctx context.Context, req *proto.PublishRequest, rsp *proto.Empty) error
Subscribe(ctx context.Context, stream server.Stream) error
}
type Broker struct {
broker
}
h := &brokerServer{sh}
var nopts []server.HandlerOption
for _, endpoint := range BrokerEndpoints {
nopts = append(nopts, api.WithEndpoint(&endpoint))
}
return s.Handle(s.NewHandler(&Broker{h}, append(nopts, opts...)...))
}

419
proto/broker.pb.go
View File

@ -1,7 +1,7 @@
// Code generated by protoc-gen-go. DO NOT EDIT. // Code generated by protoc-gen-go. DO NOT EDIT.
// versions: // versions:
// protoc-gen-go v1.26.0 // protoc-gen-go v1.26.0
// protoc v3.21.12 // protoc v3.17.3
// source: broker.proto // source: broker.proto
package servicepb package servicepb
@ -20,6 +20,44 @@ const (
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
) )
type Empty struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
}
func (x *Empty) Reset() {
*x = Empty{}
if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *Empty) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Empty) ProtoMessage() {}
func (x *Empty) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[0]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Empty.ProtoReflect.Descriptor instead.
func (*Empty) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{0}
}
type PublishRequest struct { type PublishRequest struct {
state protoimpl.MessageState state protoimpl.MessageState
sizeCache protoimpl.SizeCache sizeCache protoimpl.SizeCache
@ -32,7 +70,7 @@ type PublishRequest struct {
func (x *PublishRequest) Reset() { func (x *PublishRequest) Reset() {
*x = PublishRequest{} *x = PublishRequest{}
if protoimpl.UnsafeEnabled { if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[0] mi := &file_broker_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi) ms.StoreMessageInfo(mi)
} }
@ -45,7 +83,7 @@ func (x *PublishRequest) String() string {
func (*PublishRequest) ProtoMessage() {} func (*PublishRequest) ProtoMessage() {}
func (x *PublishRequest) ProtoReflect() protoreflect.Message { func (x *PublishRequest) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[0] mi := &file_broker_proto_msgTypes[1]
if protoimpl.UnsafeEnabled && x != nil { if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil { if ms.LoadMessageInfo() == nil {
@ -58,7 +96,7 @@ func (x *PublishRequest) ProtoReflect() protoreflect.Message {
// Deprecated: Use PublishRequest.ProtoReflect.Descriptor instead. // Deprecated: Use PublishRequest.ProtoReflect.Descriptor instead.
func (*PublishRequest) Descriptor() ([]byte, []int) { func (*PublishRequest) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{0} return file_broker_proto_rawDescGZIP(), []int{1}
} }
func (x *PublishRequest) GetTopic() string { func (x *PublishRequest) GetTopic() string {
@ -75,137 +113,6 @@ func (x *PublishRequest) GetMessage() *Message {
return nil return nil
} }
type BatchPublishRequest struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Topic string `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
Message []*Message `protobuf:"bytes,2,rep,name=message,proto3" json:"message,omitempty"`
}
func (x *BatchPublishRequest) Reset() {
*x = BatchPublishRequest{}
if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *BatchPublishRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*BatchPublishRequest) ProtoMessage() {}
func (x *BatchPublishRequest) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[1]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use BatchPublishRequest.ProtoReflect.Descriptor instead.
func (*BatchPublishRequest) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{1}
}
func (x *BatchPublishRequest) GetTopic() string {
if x != nil {
return x.Topic
}
return ""
}
func (x *BatchPublishRequest) GetMessage() []*Message {
if x != nil {
return x.Message
}
return nil
}
type PublishResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
}
func (x *PublishResponse) Reset() {
*x = PublishResponse{}
if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *PublishResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*PublishResponse) ProtoMessage() {}
func (x *PublishResponse) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[2]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use PublishResponse.ProtoReflect.Descriptor instead.
func (*PublishResponse) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{2}
}
type BatchPublishResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
}
func (x *BatchPublishResponse) Reset() {
*x = BatchPublishResponse{}
if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *BatchPublishResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*BatchPublishResponse) ProtoMessage() {}
func (x *BatchPublishResponse) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[3]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use BatchPublishResponse.ProtoReflect.Descriptor instead.
func (*BatchPublishResponse) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{3}
}
type SubscribeRequest struct { type SubscribeRequest struct {
state protoimpl.MessageState state protoimpl.MessageState
sizeCache protoimpl.SizeCache sizeCache protoimpl.SizeCache
@ -218,7 +125,7 @@ type SubscribeRequest struct {
func (x *SubscribeRequest) Reset() { func (x *SubscribeRequest) Reset() {
*x = SubscribeRequest{} *x = SubscribeRequest{}
if protoimpl.UnsafeEnabled { if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[4] mi := &file_broker_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi) ms.StoreMessageInfo(mi)
} }
@ -231,7 +138,7 @@ func (x *SubscribeRequest) String() string {
func (*SubscribeRequest) ProtoMessage() {} func (*SubscribeRequest) ProtoMessage() {}
func (x *SubscribeRequest) ProtoReflect() protoreflect.Message { func (x *SubscribeRequest) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[4] mi := &file_broker_proto_msgTypes[2]
if protoimpl.UnsafeEnabled && x != nil { if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil { if ms.LoadMessageInfo() == nil {
@ -244,7 +151,7 @@ func (x *SubscribeRequest) ProtoReflect() protoreflect.Message {
// Deprecated: Use SubscribeRequest.ProtoReflect.Descriptor instead. // Deprecated: Use SubscribeRequest.ProtoReflect.Descriptor instead.
func (*SubscribeRequest) Descriptor() ([]byte, []int) { func (*SubscribeRequest) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{4} return file_broker_proto_rawDescGZIP(), []int{2}
} }
func (x *SubscribeRequest) GetTopic() string { func (x *SubscribeRequest) GetTopic() string {
@ -261,61 +168,6 @@ func (x *SubscribeRequest) GetGroup() string {
return "" return ""
} }
type BatchSubscribeRequest struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Topic string `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
Group string `protobuf:"bytes,2,opt,name=group,proto3" json:"group,omitempty"`
}
func (x *BatchSubscribeRequest) Reset() {
*x = BatchSubscribeRequest{}
if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *BatchSubscribeRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*BatchSubscribeRequest) ProtoMessage() {}
func (x *BatchSubscribeRequest) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[5]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use BatchSubscribeRequest.ProtoReflect.Descriptor instead.
func (*BatchSubscribeRequest) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{5}
}
func (x *BatchSubscribeRequest) GetTopic() string {
if x != nil {
return x.Topic
}
return ""
}
func (x *BatchSubscribeRequest) GetGroup() string {
if x != nil {
return x.Group
}
return ""
}
type Message struct { type Message struct {
state protoimpl.MessageState state protoimpl.MessageState
sizeCache protoimpl.SizeCache sizeCache protoimpl.SizeCache
@ -328,7 +180,7 @@ type Message struct {
func (x *Message) Reset() { func (x *Message) Reset() {
*x = Message{} *x = Message{}
if protoimpl.UnsafeEnabled { if protoimpl.UnsafeEnabled {
mi := &file_broker_proto_msgTypes[6] mi := &file_broker_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi) ms.StoreMessageInfo(mi)
} }
@ -341,7 +193,7 @@ func (x *Message) String() string {
func (*Message) ProtoMessage() {} func (*Message) ProtoMessage() {}
func (x *Message) ProtoReflect() protoreflect.Message { func (x *Message) ProtoReflect() protoreflect.Message {
mi := &file_broker_proto_msgTypes[6] mi := &file_broker_proto_msgTypes[3]
if protoimpl.UnsafeEnabled && x != nil { if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil { if ms.LoadMessageInfo() == nil {
@ -354,7 +206,7 @@ func (x *Message) ProtoReflect() protoreflect.Message {
// Deprecated: Use Message.ProtoReflect.Descriptor instead. // Deprecated: Use Message.ProtoReflect.Descriptor instead.
func (*Message) Descriptor() ([]byte, []int) { func (*Message) Descriptor() ([]byte, []int) {
return file_broker_proto_rawDescGZIP(), []int{6} return file_broker_proto_rawDescGZIP(), []int{3}
} }
func (x *Message) GetHeader() map[string]string { func (x *Message) GetHeader() map[string]string {
@ -375,62 +227,39 @@ var File_broker_proto protoreflect.FileDescriptor
var file_broker_proto_rawDesc = []byte{ var file_broker_proto_rawDesc = []byte{
0x0a, 0x0c, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09, 0x0a, 0x0c, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09,
0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x22, 0x54, 0x0a, 0x0e, 0x50, 0x75, 0x62, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x22, 0x07, 0x0a, 0x05, 0x45, 0x6d, 0x70,
0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x74, 0x79, 0x22, 0x54, 0x0a, 0x0e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71,
0x6f, 0x70, 0x69, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x01, 0x20,
0x63, 0x12, 0x2c, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x2c, 0x0a, 0x07, 0x6d, 0x65,
0x28, 0x0b, 0x32, 0x12, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x4d, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x73, 0x65,
0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52,
0x59, 0x0a, 0x13, 0x42, 0x61, 0x74, 0x63, 0x68, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x75, 0x62, 0x73,
0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05,
0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x2c, 0x0a, 0x07, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70,
0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x69, 0x63, 0x12, 0x14, 0x0a, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28,
0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x09, 0x52, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x22, 0x90, 0x01, 0x0a, 0x07, 0x4d, 0x65, 0x73,
0x65, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x11, 0x0a, 0x0f, 0x50, 0x75, 0x73, 0x61, 0x67, 0x65, 0x12, 0x36, 0x0a, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x18, 0x01,
0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x16, 0x0a, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1e, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62,
0x14, 0x42, 0x61, 0x74, 0x63, 0x68, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x45,
0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x3e, 0x0a, 0x10, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x12, 0x0a, 0x04,
0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x62, 0x6f, 0x64, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x62, 0x6f, 0x64, 0x79,
0x69, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x1a, 0x39, 0x0a, 0x0b, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12,
0x14, 0x0a, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65,
0x67, 0x72, 0x6f, 0x75, 0x70, 0x22, 0x43, 0x0a, 0x15, 0x42, 0x61, 0x74, 0x63, 0x68, 0x53, 0x75, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x32, 0x84, 0x01, 0x0a, 0x06,
0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x12, 0x38, 0x0a, 0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73,
0x6f, 0x70, 0x69, 0x63, 0x12, 0x14, 0x0a, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x02, 0x20, 0x68, 0x12, 0x19, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x50, 0x75,
0x01, 0x28, 0x09, 0x52, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x22, 0x90, 0x01, 0x0a, 0x07, 0x4d, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x10, 0x2e, 0x73,
0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x36, 0x0a, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00,
0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1e, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x40, 0x0a, 0x09, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x1b, 0x2e,
0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x65, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72,
0x72, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x12, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x73, 0x65, 0x72,
0x0a, 0x04, 0x62, 0x6f, 0x64, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x62, 0x6f, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00,
0x64, 0x79, 0x1a, 0x39, 0x0a, 0x0b, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x45, 0x6e, 0x74, 0x72, 0x30, 0x01, 0x42, 0x39, 0x5a, 0x37, 0x67, 0x6f, 0x2e, 0x75, 0x6e, 0x69, 0x73, 0x74, 0x61, 0x63,
0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x2e, 0x6f, 0x72, 0x67, 0x2f, 0x6d, 0x69, 0x63, 0x72, 0x6f, 0x2d, 0x62, 0x72, 0x6f, 0x6b,
0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x65, 0x72, 0x2d, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x2f, 0x76, 0x33, 0x2f, 0x70, 0x72,
0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x32, 0xb4, 0x02, 0x6f, 0x74, 0x6f, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x62, 0x06, 0x70,
0x0a, 0x0d, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x42, 0x0a, 0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x19, 0x2e, 0x73, 0x65, 0x72,
0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70,
0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
0x65, 0x22, 0x00, 0x12, 0x51, 0x0a, 0x0c, 0x42, 0x61, 0x74, 0x63, 0x68, 0x50, 0x75, 0x62, 0x6c,
0x69, 0x73, 0x68, 0x12, 0x1e, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e,
0x42, 0x61, 0x74, 0x63, 0x68, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e,
0x42, 0x61, 0x74, 0x63, 0x68, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70,
0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x40, 0x0a, 0x09, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72,
0x69, 0x62, 0x65, 0x12, 0x1b, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e,
0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
0x1a, 0x12, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73,
0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x30, 0x01, 0x12, 0x4a, 0x0a, 0x0e, 0x42, 0x61, 0x74, 0x63,
0x68, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x20, 0x2e, 0x73, 0x65, 0x72,
0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x42, 0x61, 0x74, 0x63, 0x68, 0x53, 0x75, 0x62, 0x73,
0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x73,
0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
0x22, 0x00, 0x30, 0x01, 0x42, 0x39, 0x5a, 0x37, 0x67, 0x6f, 0x2e, 0x75, 0x6e, 0x69, 0x73, 0x74,
0x61, 0x63, 0x6b, 0x2e, 0x6f, 0x72, 0x67, 0x2f, 0x6d, 0x69, 0x63, 0x72, 0x6f, 0x2d, 0x62, 0x72,
0x6f, 0x6b, 0x65, 0x72, 0x2d, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x2f, 0x76, 0x33, 0x2f,
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x70, 0x62, 0x62,
0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
} }
var ( var (
@ -445,34 +274,26 @@ func file_broker_proto_rawDescGZIP() []byte {
return file_broker_proto_rawDescData return file_broker_proto_rawDescData
} }
var file_broker_proto_msgTypes = make([]protoimpl.MessageInfo, 8) var file_broker_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
var file_broker_proto_goTypes = []interface{}{ var file_broker_proto_goTypes = []interface{}{
(*PublishRequest)(nil), // 0: servicepb.PublishRequest (*Empty)(nil), // 0: servicepb.Empty
(*BatchPublishRequest)(nil), // 1: servicepb.BatchPublishRequest (*PublishRequest)(nil), // 1: servicepb.PublishRequest
(*PublishResponse)(nil), // 2: servicepb.PublishResponse (*SubscribeRequest)(nil), // 2: servicepb.SubscribeRequest
(*BatchPublishResponse)(nil), // 3: servicepb.BatchPublishResponse (*Message)(nil), // 3: servicepb.Message
(*SubscribeRequest)(nil), // 4: servicepb.SubscribeRequest nil, // 4: servicepb.Message.HeaderEntry
(*BatchSubscribeRequest)(nil), // 5: servicepb.BatchSubscribeRequest
(*Message)(nil), // 6: servicepb.Message
nil, // 7: servicepb.Message.HeaderEntry
} }
var file_broker_proto_depIdxs = []int32{ var file_broker_proto_depIdxs = []int32{
6, // 0: servicepb.PublishRequest.message:type_name -> servicepb.Message 3, // 0: servicepb.PublishRequest.message:type_name -> servicepb.Message
6, // 1: servicepb.BatchPublishRequest.message:type_name -> servicepb.Message 4, // 1: servicepb.Message.header:type_name -> servicepb.Message.HeaderEntry
7, // 2: servicepb.Message.header:type_name -> servicepb.Message.HeaderEntry 1, // 2: servicepb.Broker.Publish:input_type -> servicepb.PublishRequest
0, // 3: servicepb.BrokerService.Publish:input_type -> servicepb.PublishRequest 2, // 3: servicepb.Broker.Subscribe:input_type -> servicepb.SubscribeRequest
1, // 4: servicepb.BrokerService.BatchPublish:input_type -> servicepb.BatchPublishRequest 0, // 4: servicepb.Broker.Publish:output_type -> servicepb.Empty
4, // 5: servicepb.BrokerService.Subscribe:input_type -> servicepb.SubscribeRequest 3, // 5: servicepb.Broker.Subscribe:output_type -> servicepb.Message
5, // 6: servicepb.BrokerService.BatchSubscribe:input_type -> servicepb.BatchSubscribeRequest 4, // [4:6] is the sub-list for method output_type
2, // 7: servicepb.BrokerService.Publish:output_type -> servicepb.PublishResponse 2, // [2:4] is the sub-list for method input_type
3, // 8: servicepb.BrokerService.BatchPublish:output_type -> servicepb.BatchPublishResponse 2, // [2:2] is the sub-list for extension type_name
6, // 9: servicepb.BrokerService.Subscribe:output_type -> servicepb.Message 2, // [2:2] is the sub-list for extension extendee
6, // 10: servicepb.BrokerService.BatchSubscribe:output_type -> servicepb.Message 0, // [0:2] is the sub-list for field type_name
7, // [7:11] is the sub-list for method output_type
3, // [3:7] is the sub-list for method input_type
3, // [3:3] is the sub-list for extension type_name
3, // [3:3] is the sub-list for extension extendee
0, // [0:3] is the sub-list for field type_name
} }
func init() { file_broker_proto_init() } func init() { file_broker_proto_init() }
@ -482,7 +303,7 @@ func file_broker_proto_init() {
} }
if !protoimpl.UnsafeEnabled { if !protoimpl.UnsafeEnabled {
file_broker_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { file_broker_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*PublishRequest); i { switch v := v.(*Empty); i {
case 0: case 0:
return &v.state return &v.state
case 1: case 1:
@ -494,7 +315,7 @@ func file_broker_proto_init() {
} }
} }
file_broker_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { file_broker_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*BatchPublishRequest); i { switch v := v.(*PublishRequest); i {
case 0: case 0:
return &v.state return &v.state
case 1: case 1:
@ -506,30 +327,6 @@ func file_broker_proto_init() {
} }
} }
file_broker_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { file_broker_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*PublishResponse); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_broker_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*BatchPublishResponse); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_broker_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*SubscribeRequest); i { switch v := v.(*SubscribeRequest); i {
case 0: case 0:
return &v.state return &v.state
@ -541,19 +338,7 @@ func file_broker_proto_init() {
return nil return nil
} }
} }
file_broker_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} { file_broker_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*BatchSubscribeRequest); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_broker_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Message); i { switch v := v.(*Message); i {
case 0: case 0:
return &v.state return &v.state
@ -572,7 +357,7 @@ func file_broker_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(), GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_broker_proto_rawDesc, RawDescriptor: file_broker_proto_rawDesc,
NumEnums: 0, NumEnums: 0,
NumMessages: 8, NumMessages: 5,
NumExtensions: 0, NumExtensions: 0,
NumServices: 1, NumServices: 1,
}, },

34
proto/broker.proto
View File

@ -3,42 +3,24 @@ syntax = "proto3";
package servicepb; package servicepb;
option go_package="go.unistack.org/micro-broker-service/v3/proto;servicepb"; option go_package="go.unistack.org/micro-broker-service/v3/proto;servicepb";
service BrokerService { service Broker {
rpc Publish(PublishRequest) returns (PublishResponse) {}; rpc Publish(PublishRequest) returns (Empty) {};
rpc BatchPublish(BatchPublishRequest) returns (BatchPublishResponse) {};
rpc Subscribe(SubscribeRequest) returns (stream Message) {}; rpc Subscribe(SubscribeRequest) returns (stream Message) {};
rpc BatchSubscribe(BatchSubscribeRequest) returns (stream Message) {}; }
};
message Empty {}
message PublishRequest { message PublishRequest {
string topic = 1; string topic = 1;
Message message = 2; Message message = 2;
}; }
message BatchPublishRequest {
string topic = 1;
repeated Message message = 2;
};
message PublishResponse {
};
message BatchPublishResponse {
};
message SubscribeRequest { message SubscribeRequest {
string topic = 1; string topic = 1;
string group = 2; string group = 2;
}; }
message BatchSubscribeRequest {
string topic = 1;
string group = 2;
};
message Message { message Message {
map<string,string> header = 1; map<string,string> header = 1;
bytes body = 2; bytes body = 2;
}; }

165
proto/broker_grpc.pb.go Normal file
View File

@ -0,0 +1,165 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
package servicepb
import (
context "context"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
)
// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.32.0 or later.
const _ = grpc.SupportPackageIsVersion7
// BrokerClient is the client API for Broker service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
type BrokerClient interface {
Publish(ctx context.Context, in *PublishRequest, opts ...grpc.CallOption) (*Empty, error)
Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (Broker_SubscribeClient, error)
}
type brokerClient struct {
cc grpc.ClientConnInterface
}
func NewBrokerClient(cc grpc.ClientConnInterface) BrokerClient {
return &brokerClient{cc}
}
func (c *brokerClient) Publish(ctx context.Context, in *PublishRequest, opts ...grpc.CallOption) (*Empty, error) {
out := new(Empty)
err := c.cc.Invoke(ctx, "/servicepb.Broker/Publish", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *brokerClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (Broker_SubscribeClient, error) {
stream, err := c.cc.NewStream(ctx, &Broker_ServiceDesc.Streams[0], "/servicepb.Broker/Subscribe", opts...)
if err != nil {
return nil, err
}
x := &brokerSubscribeClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
type Broker_SubscribeClient interface {
Recv() (*Message, error)
grpc.ClientStream
}
type brokerSubscribeClient struct {
grpc.ClientStream
}
func (x *brokerSubscribeClient) Recv() (*Message, error) {
m := new(Message)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
// BrokerServer is the server API for Broker service.
// All implementations must embed UnimplementedBrokerServer
// for forward compatibility
type BrokerServer interface {
Publish(context.Context, *PublishRequest) (*Empty, error)
Subscribe(*SubscribeRequest, Broker_SubscribeServer) error
mustEmbedUnimplementedBrokerServer()
}
// UnimplementedBrokerServer must be embedded to have forward compatible implementations.
type UnimplementedBrokerServer struct {
}
func (UnimplementedBrokerServer) Publish(context.Context, *PublishRequest) (*Empty, error) {
return nil, status.Errorf(codes.Unimplemented, "method Publish not implemented")
}
func (UnimplementedBrokerServer) Subscribe(*SubscribeRequest, Broker_SubscribeServer) error {
return status.Errorf(codes.Unimplemented, "method Subscribe not implemented")
}
func (UnimplementedBrokerServer) mustEmbedUnimplementedBrokerServer() {}
// UnsafeBrokerServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to BrokerServer will
// result in compilation errors.
type UnsafeBrokerServer interface {
mustEmbedUnimplementedBrokerServer()
}
func RegisterBrokerServer(s grpc.ServiceRegistrar, srv BrokerServer) {
s.RegisterService(&Broker_ServiceDesc, srv)
}
func _Broker_Publish_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(PublishRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(BrokerServer).Publish(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/servicepb.Broker/Publish",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(BrokerServer).Publish(ctx, req.(*PublishRequest))
}
return interceptor(ctx, in, info, handler)
}
func _Broker_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(SubscribeRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(BrokerServer).Subscribe(m, &brokerSubscribeServer{stream})
}
type Broker_SubscribeServer interface {
Send(*Message) error
grpc.ServerStream
}
type brokerSubscribeServer struct {
grpc.ServerStream
}
func (x *brokerSubscribeServer) Send(m *Message) error {
return x.ServerStream.SendMsg(m)
}
// Broker_ServiceDesc is the grpc.ServiceDesc for Broker service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var Broker_ServiceDesc = grpc.ServiceDesc{
ServiceName: "servicepb.Broker",
HandlerType: (*BrokerServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "Publish",
Handler: _Broker_Publish_Handler,
},
},
Streams: []grpc.StreamDesc{
{
StreamName: "Subscribe",
Handler: _Broker_Subscribe_Handler,
ServerStreams: true,
},
},
Metadata: "broker.proto",
}

51
service.go
View File

@ -11,17 +11,14 @@ import (
"go.unistack.org/micro/v3/broker" "go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client" "go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/logger" "go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
) )
var _ broker.Broker = &serviceBroker{}
type serviceBroker struct { type serviceBroker struct {
client pbmicro.BrokerServiceClient
service string
opts broker.Options
addrs []string addrs []string
service string
client pbmicro.BrokerClient
init bool init bool
opts broker.Options
} }
func (b *serviceBroker) Address() string { func (b *serviceBroker) Address() string {
@ -76,7 +73,7 @@ func (b *serviceBroker) Init(opts ...broker.Option) error {
return fmt.Errorf("missing Client option") return fmt.Errorf("missing Client option")
} }
b.client = pbmicro.NewBrokerServiceClient(b.service, cli) b.client = pbmicro.NewBrokerClient(b.service, cli)
b.init = true b.init = true
return nil return nil
@ -86,38 +83,18 @@ func (b *serviceBroker) Options() broker.Options {
return b.opts return b.opts
} }
func (b *serviceBroker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
return b.publish(ctx, msgs, opts...)
}
func (b *serviceBroker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error { func (b *serviceBroker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
msg.Header.Set(metadata.HeaderTopic, topic) if logger.V(logger.TraceLevel) {
return b.publish(ctx, []*broker.Message{msg}, opts...) logger.Tracef(ctx, "Publishing to topic %s broker %v", topic, b.addrs)
}
func (b *serviceBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
for _, msg := range msgs {
topic, _ := msg.Header.Get(metadata.HeaderTopic)
if logger.V(logger.TraceLevel) {
logger.Tracef(ctx, "Publishing to topic %s broker %v", topic, b.addrs)
}
_, err := b.client.Publish(ctx, &pb.PublishRequest{
Topic: topic,
Message: &pb.Message{
Header: msg.Header,
Body: msg.Body,
},
}, client.WithAddress(b.addrs...))
if err != nil {
return err
}
} }
_, err := b.client.Publish(ctx, &pb.PublishRequest{
return nil Topic: topic,
} Message: &pb.Message{
Header: msg.Header,
func (b *serviceBroker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) { Body: msg.Body,
return nil, nil },
}, client.WithAddress(b.addrs...))
return err
} }
func (b *serviceBroker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) { func (b *serviceBroker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {

16
subscriber.go
View File

@ -6,29 +6,23 @@ import (
pbmicro "go.unistack.org/micro-broker-service/v3/micro" pbmicro "go.unistack.org/micro-broker-service/v3/micro"
"go.unistack.org/micro/v3/broker" "go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/logger" "go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
) )
type serviceSub struct { type serviceSub struct {
topic string topic string
group string group string
handler broker.Handler handler broker.Handler
stream pbmicro.BrokerService_SubscribeClient stream pbmicro.Broker_SubscribeClient
closed chan bool closed chan bool
options broker.SubscribeOptions options broker.SubscribeOptions
} }
type serviceEvent struct { type serviceEvent struct {
ctx context.Context
topic string topic string
err error err error
message *broker.Message message *broker.Message
} }
func (s *serviceEvent) Context() context.Context {
return s.ctx
}
func (s *serviceEvent) Topic() string { func (s *serviceEvent) Topic() string {
return s.topic return s.topic
} }
@ -45,10 +39,6 @@ func (s *serviceEvent) Error() error {
return s.err return s.err
} }
func (s *serviceEvent) SetError(err error) {
s.err = err
}
func (s *serviceSub) isClosed() bool { func (s *serviceSub) isClosed() bool {
select { select {
case <-s.closed: case <-s.closed:
@ -73,10 +63,9 @@ func (s *serviceSub) run(ctx context.Context) error {
for { for {
// TODO: do not fail silently // TODO: do not fail silently
msg, err := s.stream.Recv() msg, err := s.stream.Recv()
ctx := metadata.NewIncomingContext(context.Background(), msg.Header)
if err != nil { if err != nil {
if logger.V(logger.TraceLevel) { if logger.V(logger.TraceLevel) {
logger.Tracef(ctx, "streaming error for subcription to topic %s: %v", s.Topic(), err) logger.Tracef(ctx, "Streaming error for subcription to topic %s: %v", s.Topic(), err)
} }
// close the exit channel // close the exit channel
@ -92,7 +81,6 @@ func (s *serviceSub) run(ctx context.Context) error {
} }
p := &serviceEvent{ p := &serviceEvent{
ctx: ctx,
topic: s.topic, topic: s.topic,
message: &broker.Message{ message: &broker.Message{
Header: msg.Header, Header: msg.Header,