Compare commits
16 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e6e8bcfa5d | |||
| 11700d0482 | |||
| 8e582cc1c0 | |||
| 994882e1b2 | |||
| 5b28857fcd | |||
| f94f94b416 | |||
| 8e4b7d1b54 | |||
| 13bb13f432 | |||
| f13d3917d9 | |||
| 7916dafb4d | |||
| 83b037df20 | |||
| 5934963d81 | |||
| 2a935a51f3 | |||
| f39888f6d4 | |||
| b15586163e | |||
| 17404ee935 |
19
.github/renovate.json
vendored
19
.github/renovate.json
vendored
@ -1,19 +0,0 @@
|
||||
{
|
||||
"extends": [
|
||||
"config:base"
|
||||
],
|
||||
"packageRules": [
|
||||
{
|
||||
"matchUpdateTypes": ["minor", "patch", "pin", "digest"],
|
||||
"automerge": true
|
||||
},
|
||||
{
|
||||
"groupName": "all deps",
|
||||
"separateMajorMinor": true,
|
||||
"groupSlug": "all",
|
||||
"packagePatterns": [
|
||||
"*"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
13
.github/stale.sh
vendored
13
.github/stale.sh
vendored
@ -1,13 +0,0 @@
|
||||
#!/bin/bash -ex
|
||||
|
||||
export PATH=$PATH:$(pwd)/bin
|
||||
export GO111MODULE=on
|
||||
export GOBIN=$(pwd)/bin
|
||||
|
||||
#go get github.com/rvflash/goup@v0.4.1
|
||||
|
||||
#goup -v ./...
|
||||
#go get github.com/psampaz/go-mod-outdated@v0.6.0
|
||||
go list -u -m -mod=mod -json all | go-mod-outdated -update -direct -ci || true
|
||||
|
||||
#go list -u -m -json all | go-mod-outdated -update
|
||||
20
.github/workflows/autoapprove.yml
vendored
Normal file
20
.github/workflows/autoapprove.yml
vendored
Normal file
@ -0,0 +1,20 @@
|
||||
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@v2
|
||||
if: github.actor == 'vtolstov' || github.actor == 'dependabot[bot]'
|
||||
id: approve
|
||||
with:
|
||||
github-token: ${{ secrets.GITHUB_TOKEN }}
|
||||
21
.github/workflows/automerge.yml
vendored
Normal file
21
.github/workflows/automerge.yml
vendored
Normal file
@ -0,0 +1,21 @@
|
||||
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}}
|
||||
7
.github/workflows/build.yml
vendored
7
.github/workflows/build.yml
vendored
@ -3,6 +3,7 @@ on:
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
jobs:
|
||||
test:
|
||||
name: test
|
||||
@ -13,7 +14,7 @@ jobs:
|
||||
with:
|
||||
go-version: 1.16
|
||||
- name: checkout
|
||||
uses: actions/checkout@v2
|
||||
uses: actions/checkout@v3
|
||||
- name: cache
|
||||
uses: actions/cache@v2
|
||||
with:
|
||||
@ -31,9 +32,9 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout
|
||||
uses: actions/checkout@v2
|
||||
uses: actions/checkout@v3
|
||||
- name: lint
|
||||
uses: golangci/golangci-lint-action@v2
|
||||
uses: golangci/golangci-lint-action@v3.1.0
|
||||
continue-on-error: true
|
||||
with:
|
||||
# Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
|
||||
|
||||
78
.github/workflows/codeql-analysis.yml
vendored
Normal file
78
.github/workflows/codeql-analysis.yml
vendored
Normal file
@ -0,0 +1,78 @@
|
||||
# For most projects, this workflow file will not need changing; you simply need
|
||||
# to commit it to your repository.
|
||||
#
|
||||
# You may wish to alter this file to override the set of languages analyzed,
|
||||
# or to provide custom queries or build logic.
|
||||
#
|
||||
# ******** NOTE ********
|
||||
# We have attempted to detect the languages in your repository. Please check
|
||||
# the `language` matrix defined below to confirm you have the correct set of
|
||||
# supported CodeQL languages.
|
||||
#
|
||||
name: "codeql"
|
||||
|
||||
on:
|
||||
workflow_run:
|
||||
workflows: ["prbuild"]
|
||||
types:
|
||||
- completed
|
||||
push:
|
||||
branches: [ master, v3 ]
|
||||
pull_request:
|
||||
# The branches below must be a subset of the branches above
|
||||
branches: [ master, v3 ]
|
||||
schedule:
|
||||
- cron: '34 1 * * 0'
|
||||
|
||||
jobs:
|
||||
analyze:
|
||||
name: analyze
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
actions: read
|
||||
contents: read
|
||||
security-events: write
|
||||
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
language: [ 'go' ]
|
||||
# CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python' ]
|
||||
# Learn more:
|
||||
# https://docs.github.com/en/free-pro-team@latest/github/finding-security-vulnerabilities-and-errors-in-your-code/configuring-code-scanning#changing-the-languages-that-are-analyzed
|
||||
|
||||
steps:
|
||||
- name: checkout
|
||||
uses: actions/checkout@v3
|
||||
- name: setup
|
||||
uses: actions/setup-go@v2
|
||||
with:
|
||||
go-version: 1.16
|
||||
# Initializes the CodeQL tools for scanning.
|
||||
- name: init
|
||||
uses: github/codeql-action/init@v1
|
||||
with:
|
||||
languages: ${{ matrix.language }}
|
||||
# If you wish to specify custom queries, you can do so here or in a config file.
|
||||
# By default, queries listed here will override any specified in a config file.
|
||||
# Prefix the list here with "+" to use these queries and those in the config file.
|
||||
# queries: ./path/to/local/query, your-org/your-repo/queries@main
|
||||
|
||||
# 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)
|
||||
- name: autobuild
|
||||
uses: github/codeql-action/autobuild@v1
|
||||
|
||||
# ℹ️ Command-line programs to run using the OS shell.
|
||||
# 📚 https://git.io/JvXDl
|
||||
|
||||
# ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
|
||||
# and modify them (or add more) to build your code if your project
|
||||
# uses a compiled language
|
||||
|
||||
#- run: |
|
||||
# make bootstrap
|
||||
# make release
|
||||
|
||||
- name: analyze
|
||||
uses: github/codeql-action/analyze@v1
|
||||
27
.github/workflows/dependabot-automerge.yml
vendored
Normal file
27
.github/workflows/dependabot-automerge.yml
vendored
Normal file
@ -0,0 +1,27 @@
|
||||
name: "dependabot-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 == 'dependabot[bot]'
|
||||
steps:
|
||||
- name: metadata
|
||||
id: metadata
|
||||
uses: dependabot/fetch-metadata@v1.3.0
|
||||
with:
|
||||
github-token: "${{ secrets.TOKEN }}"
|
||||
- name: merge
|
||||
id: merge
|
||||
if: ${{contains(steps.metadata.outputs.dependency-names, 'go.unistack.org')}}
|
||||
run: gh pr merge --auto --merge "$PR_URL"
|
||||
env:
|
||||
PR_URL: ${{github.event.pull_request.html_url}}
|
||||
GITHUB_TOKEN: ${{secrets.TOKEN}}
|
||||
7
.github/workflows/pr.yml
vendored
7
.github/workflows/pr.yml
vendored
@ -3,6 +3,7 @@ on:
|
||||
pull_request:
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
jobs:
|
||||
test:
|
||||
name: test
|
||||
@ -13,7 +14,7 @@ jobs:
|
||||
with:
|
||||
go-version: 1.16
|
||||
- name: checkout
|
||||
uses: actions/checkout@v2
|
||||
uses: actions/checkout@v3
|
||||
- name: cache
|
||||
uses: actions/cache@v2
|
||||
with:
|
||||
@ -31,9 +32,9 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout
|
||||
uses: actions/checkout@v2
|
||||
uses: actions/checkout@v3
|
||||
- name: lint
|
||||
uses: golangci/golangci-lint-action@v2
|
||||
uses: golangci/golangci-lint-action@v3.1.0
|
||||
continue-on-error: true
|
||||
with:
|
||||
# Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
|
||||
|
||||
@ -1,7 +1,9 @@
|
||||
# Micro Broker Segmentio [](https://opensource.org/licenses/Apache-2.0) [](https://pkg.go.dev/github.com/unistack-org/micro-broker-segmentio?tab=overview) [](https://github.com/unistack-org/micro-broker-segmentio/actions?query=workflow%3Abuild+branch%3Amaster+event%3Apush) [](https://goreportcard.com/report/github.com/unistack-org/micro-broker-segmentio)
|
||||
# Micro Broker Segmentio [](https://opensource.org/licenses/Apache-2.0) [](https://pkg.go.dev/go.unistack.org/micro-broker-segmentio?tab=overview) [](https://go.unistack.org/micro-broker-segmentio/actions?query=workflow%3Abuild+branch%3Amaster+event%3Apush) [](https://goreportcard.com/report/go.unistack.org/micro-broker-segmentio)
|
||||
|
||||
Micro Broker Segmentio implementation.
|
||||
|
||||
DEPRECATION WARNING: not maintained, use go.unistack.org/micro-broker-kgo instead
|
||||
|
||||
## License
|
||||
|
||||
Apache 2.0 licensed.
|
||||
|
||||
@ -6,27 +6,27 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
segmentio "github.com/unistack-org/micro-broker-segmentio/v3"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
"github.com/unistack-org/micro/v3/logger"
|
||||
segmentio "go.unistack.org/micro-broker-segmentio/v3"
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
)
|
||||
|
||||
var (
|
||||
bm = &broker.Message{
|
||||
Header: map[string]string{"hkey": "hval"},
|
||||
Body: []byte("body"),
|
||||
}
|
||||
)
|
||||
var bm = &broker.Message{
|
||||
Header: map[string]string{"hkey": "hval"},
|
||||
Body: []byte(`"body"`),
|
||||
}
|
||||
|
||||
func TestPubSub(t *testing.T) {
|
||||
t.Skip()
|
||||
logger.DefaultLogger.Init(logger.WithLevel(logger.TraceLevel))
|
||||
ctx := context.Background()
|
||||
|
||||
if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
|
||||
t.Skip()
|
||||
}
|
||||
|
||||
if err := logger.DefaultLogger.Init(logger.WithLevel(logger.TraceLevel)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
var addrs []string
|
||||
if addr := os.Getenv("BROKER_ADDRS"); len(addr) == 0 {
|
||||
addrs = []string{"127.0.0.1:9092"}
|
||||
@ -34,7 +34,7 @@ func TestPubSub(t *testing.T) {
|
||||
addrs = strings.Split(addr, ",")
|
||||
}
|
||||
|
||||
b := segmentio.NewBroker(broker.Addrs(addrs...))
|
||||
b := segmentio.NewBroker(broker.Addrs(addrs...), segmentio.ClientID("test"))
|
||||
if err := b.Init(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
7
go.mod
7
go.mod
@ -1,9 +1,8 @@
|
||||
module github.com/unistack-org/micro-broker-segmentio/v3
|
||||
module go.unistack.org/micro-broker-segmentio/v3
|
||||
|
||||
go 1.16
|
||||
|
||||
require (
|
||||
github.com/google/uuid v1.2.0
|
||||
github.com/segmentio/kafka-go v0.4.16
|
||||
github.com/unistack-org/micro/v3 v3.3.20
|
||||
github.com/segmentio/kafka-go v0.4.22
|
||||
go.unistack.org/micro/v3 v3.8.5
|
||||
)
|
||||
|
||||
30
go.sum
30
go.sum
@ -1,17 +1,17 @@
|
||||
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
|
||||
github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw=
|
||||
github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
|
||||
github.com/ef-ds/deque v1.0.4/go.mod h1:gXDnTC3yqvBcHbq2lcExjtAcVrOnJCbMcZXmuj8Z4tg=
|
||||
github.com/frankban/quicktest v1.11.3 h1:8sXhOn0uLys67V8EsXLc6eszDs8VXWxL3iRvebPhedY=
|
||||
github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
|
||||
github.com/golang-jwt/jwt/v4 v4.1.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
|
||||
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
|
||||
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
|
||||
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
|
||||
github.com/google/go-cmp v0.5.4 h1:L8R9j+yAqZuZjsqh/z+F1NCffTKKLShY6zXTItVIZ8M=
|
||||
github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
|
||||
github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
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/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
|
||||
github.com/klauspost/compress v1.9.8 h1:VMAMUUOh+gaxKTMk+zqbjsSjsIcUcL/LF4o63i82QyA=
|
||||
github.com/klauspost/compress v1.9.8/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
|
||||
@ -25,24 +25,26 @@ github.com/pierrec/lz4 v2.6.0+incompatible h1:Ix9yFKn1nSPBLFl/yZknTp8TU5G4Ps0JDm
|
||||
github.com/pierrec/lz4 v2.6.0+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/segmentio/kafka-go v0.4.16 h1:9dt78ehM9qzAkekA60D6A96RlqDzC3hnYYa8y5Szd+U=
|
||||
github.com/segmentio/kafka-go v0.4.16/go.mod h1:19+Eg7KwrNKy/PFhiIthEPkO8k+ac7/ZYXwYM9Df10w=
|
||||
github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
|
||||
github.com/segmentio/kafka-go v0.4.22 h1:F4k2OTm9Y4+zliuoXgNKJZTktE0miQioZZzofsjhRdk=
|
||||
github.com/segmentio/kafka-go v0.4.22/go.mod h1:XzMcoMjSzDGHcIwpWUI7GB43iKZ2fTVmryPSGLf/MPg=
|
||||
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.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
|
||||
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/unistack-org/micro/v3 v3.3.20 h1:gB+sPtvYuEKJQG/k5xnC1TK7MnZbr7wlgyqpYNREdyo=
|
||||
github.com/unistack-org/micro/v3 v3.3.20/go.mod h1:LXmPfbJnJNvL0kQs8HfnkV3Wya2Wb+C7keVq++RCZnk=
|
||||
github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
|
||||
github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
|
||||
github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=
|
||||
github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
|
||||
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.5 h1:DIYWRsQF+NPhKZP45sCtNsUhaRw6u2+Ps7U+pKU7i3s=
|
||||
go.unistack.org/micro/v3 v3.8.5/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-20190506204251-e1dfcc566284 h1:rlLehGeYg6jfoyz/eDqDU1iRXLKfR42nnNh57ytKEWo=
|
||||
golang.org/x/crypto v0.0.0-20190506204251-e1dfcc566284/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20210510120150-4163338589ed h1:p9UgmWI9wKpfYmgaV/IZKGdXc5qEK45tDwwwDyjS26I=
|
||||
golang.org/x/net v0.0.0-20210510120150-4163338589ed/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
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/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-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
@ -52,8 +54,12 @@ 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/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
|
||||
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/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
|
||||
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.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
|
||||
|
||||
42
options.go
42
options.go
@ -2,19 +2,25 @@ package segmentio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
kafka "github.com/segmentio/kafka-go"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
"github.com/unistack-org/micro/v3/client"
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/client"
|
||||
)
|
||||
|
||||
var (
|
||||
DefaultReaderConfig = kafka.WriterConfig{}
|
||||
DefaultWriterConfig = kafka.ReaderConfig{}
|
||||
DefaultReaderConfig = kafka.ReaderConfig{}
|
||||
DefaultWriterConfig = kafka.WriterConfig{}
|
||||
DefaultStatsInterval = time.Second * 10
|
||||
DefaultCommitInterval = time.Second * 0
|
||||
DefaultCommitQueueSize = 2000
|
||||
)
|
||||
|
||||
type readerConfigKey struct{}
|
||||
type writerConfigKey struct{}
|
||||
type (
|
||||
readerConfigKey struct{}
|
||||
writerConfigKey struct{}
|
||||
)
|
||||
|
||||
func ReaderConfig(c kafka.ReaderConfig) broker.Option {
|
||||
return broker.SetOption(readerConfigKey{}, c)
|
||||
@ -52,3 +58,27 @@ func PublishKey(key []byte) broker.PublishOption {
|
||||
func ClientPublishKey(key []byte) client.PublishOption {
|
||||
return client.SetPublishOption(publishKey{}, key)
|
||||
}
|
||||
|
||||
type statsIntervalKey struct{}
|
||||
|
||||
func StatsInterval(td time.Duration) broker.Option {
|
||||
return broker.SetOption(statsIntervalKey{}, td)
|
||||
}
|
||||
|
||||
type writerCompletionFunc struct{}
|
||||
|
||||
func WriterCompletionFunc(fn func([]kafka.Message, error)) broker.Option {
|
||||
return broker.SetOption(writerCompletionFunc{}, fn)
|
||||
}
|
||||
|
||||
type clientIDKey struct{}
|
||||
|
||||
func ClientID(id string) broker.Option {
|
||||
return broker.SetOption(clientIDKey{}, id)
|
||||
}
|
||||
|
||||
type commitIntervalKey struct{}
|
||||
|
||||
func CommitInterval(td time.Duration) broker.Option {
|
||||
return broker.SetOption(commitIntervalKey{}, td)
|
||||
}
|
||||
|
||||
883
segmentio.go
883
segmentio.go
File diff suppressed because it is too large
Load Diff
@ -7,14 +7,17 @@ import (
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
segmentio "github.com/unistack-org/micro-broker-segmentio/v3"
|
||||
"github.com/unistack-org/micro/v3/broker"
|
||||
"github.com/unistack-org/micro/v3/logger"
|
||||
segmentio "go.unistack.org/micro-broker-segmentio/v3"
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
)
|
||||
|
||||
func TestSegmentioSubscribe(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
logger.DefaultLogger.Init(logger.WithLevel(logger.TraceLevel))
|
||||
if err := logger.DefaultLogger.Init(logger.WithLevel(logger.TraceLevel)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
|
||||
t.Skip()
|
||||
}
|
||||
@ -178,7 +181,7 @@ func BenchmarkSegmentioCodecJsonSubscribe(b *testing.B) {
|
||||
return
|
||||
}
|
||||
if err := brk.Publish(ctx, "test_topic", bm); err != nil {
|
||||
b.Fatal(err)
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
131
stats.go
Normal file
131
stats.go
Normal file
@ -0,0 +1,131 @@
|
||||
package segmentio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
kafka "github.com/segmentio/kafka-go"
|
||||
"go.unistack.org/micro/v3/meter"
|
||||
)
|
||||
|
||||
func readerStats(ctx context.Context, r *kafka.Reader, td time.Duration, m meter.Meter) {
|
||||
ticker := time.NewTicker(td)
|
||||
var once sync.Once
|
||||
|
||||
onceLabels := make([]string, 0, 4)
|
||||
|
||||
defer func() {
|
||||
ticker.Stop()
|
||||
m.Counter("broker_reader_count", onceLabels...).Add(int(-1))
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
rstats := r.Stats()
|
||||
labels := []string{"topic", rstats.Topic, "partition", rstats.Partition, "client_id", rstats.ClientID}
|
||||
once.Do(func() {
|
||||
onceLabels = []string{"topic", rstats.Topic, "client_id", rstats.ClientID}
|
||||
m.Counter("broker_reader_count", onceLabels...).Add(int(1))
|
||||
})
|
||||
m.Counter("broker_reader_dial_count", labels...).Add(int(rstats.Dials))
|
||||
m.Counter("broker_reader_fetch_count", labels...).Add(int(rstats.Fetches))
|
||||
m.Counter("broker_reader_message_count", labels...).Add(int(rstats.Messages))
|
||||
m.Counter("broker_reader_message_bytes", labels...).Add(int(rstats.Bytes))
|
||||
m.Counter("broker_reader_rebalance_count", labels...).Add(int(rstats.Rebalances))
|
||||
m.Counter("broker_reader_timeout_count", labels...).Add(int(rstats.Timeouts))
|
||||
m.Counter("broker_reader_error", labels...).Add(int(rstats.Errors))
|
||||
|
||||
/*
|
||||
m.Counter("broker_reader_dial_seconds_avg", labels...).Add(uint64(rstats.DialTime.Avg))
|
||||
m.Counter("broker_reader_dial_seconds_min", labels...).Add(uint64(rstats.DialTime.Min))
|
||||
m.Counter("broker_reader_dial_seconds_max", labels...).Add(uint64(rstats.DialTime.Max))
|
||||
m.Counter("broker_reader_read_seconds_avg", labels...).Add(uint64(rstats.ReadTime.Avg))
|
||||
m.Counter("broker_reader_read_seconds_min", labels...).Add(uint64(rstats.ReadTime.Min))
|
||||
m.Counter("broker_reader_read_seconds_max", labels...).Add(uint64(rstats.ReadTime.Max))
|
||||
m.Counter("broker_reader_wait_seconds_avg", labels...).Add(uint64(rstats.WaitTime.Avg))
|
||||
m.Counter("broker_reader_wait_seconds_min", labels...).Add(uint64(rstats.WaitTime.Min))
|
||||
m.Counter("broker_reader_wait_seconds_max", labels...).Add(uint64(rstats.WaitTime.Max))
|
||||
*/
|
||||
/*
|
||||
m.Counter("broker_reader_fetch_size_avg", labels...).Add(uint64(rstats.FetchSize.Avg))
|
||||
m.Counter("broker_reader_fetch_size_min", labels...).Set(uint64(rstats.FetchSize.Min))
|
||||
m.Counter("broker_reader_fetch_size_max", labels...).Set(uint64(rstats.FetchSize.Max))
|
||||
m.Counter("broker_reader_fetch_bytes_avg", labels...).Set(uint64(rstats.FetchBytes.Avg))
|
||||
m.Counter("broker_reader_fetch_bytes_min", labels...).Set(uint64(rstats.FetchBytes.Min))
|
||||
m.Counter("broker_reader_fetch_bytes_max", labels...).Set(uint64(rstats.FetchBytes.Max))
|
||||
*/
|
||||
|
||||
m.Counter("broker_reader_offset", labels...).Set(uint64(rstats.Offset))
|
||||
m.Counter("broker_reader_lag", labels...).Set(uint64(rstats.Lag))
|
||||
m.Counter("broker_reader_fetch_bytes_min", labels...).Set(uint64(rstats.MinBytes))
|
||||
m.Counter("broker_reader_fetch_bytes_max", labels...).Set(uint64(rstats.MaxBytes))
|
||||
m.Counter("broker_reader_fetch_wait_max", labels...).Set(uint64(rstats.MaxWait))
|
||||
m.Counter("broker_reader_queue_length", labels...).Set(uint64(rstats.QueueLength))
|
||||
m.Counter("broker_reader_queue_capacity", labels...).Set(uint64(rstats.QueueCapacity))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func writerStats(ctx context.Context, w *kafka.Writer, td time.Duration, m meter.Meter) {
|
||||
ticker := time.NewTicker(td)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
wstats := w.Stats()
|
||||
labels := []string{}
|
||||
|
||||
m.Counter("broker_writer_write_count", labels...).Add(int(wstats.Writes))
|
||||
m.Counter("broker_writer_message_count", labels...).Add(int(wstats.Messages))
|
||||
m.Counter("broker_writer_message_bytes", labels...).Add(int(wstats.Bytes))
|
||||
m.Counter("broker_writer_error_count", labels...).Add(int(wstats.Errors))
|
||||
|
||||
/*
|
||||
m.Counter("broker_writer_batch_seconds_avg", labels...).Set(uint64(wstats.BatchTime.Avg))
|
||||
m.Counter("broker_writer_batch_seconds_min", labels...).Set(uint64(wstats.BatchTime.Min))
|
||||
m.Counter("broker_writer_batch_seconds_max", labels...).Set(uint64(wstats.BatchTime.Max))
|
||||
m.Counter("broker_writer_write_seconds_avg", labels...).Set(uint64(wstats.WriteTime.Avg))
|
||||
m.Counter("broker_writer_write_seconds_min", labels...).Set(uint64(wstats.WriteTime.Min))
|
||||
m.Counter("broker_writer_write_seconds_max", labels...).Set(uint64(wstats.WriteTime.Max))
|
||||
m.Counter("broker_writer_wait_seconds_avg", labels...).Set(uint64(wstats.WaitTime.Avg))
|
||||
m.Counter("broker_writer_wait_seconds_min", labels...).Set(uint64(wstats.WaitTime.Min))
|
||||
m.Counter("broker_writer_wait_seconds_max", labels...).Set(uint64(wstats.WaitTime.Max))
|
||||
|
||||
m.Counter("broker_writer_retries_count_avg", labels...).Set(uint64(wstats.Retries.Avg))
|
||||
m.Counter("broker_writer_retries_count_min", labels...).Set(uint64(wstats.Retries.Min))
|
||||
m.Counter("broker_writer_retries_count_max", labels...).Set(uint64(wstats.Retries.Max))
|
||||
m.Counter("broker_writer_batch_size_avg", labels...).Set(uint64(wstats.BatchSize.Avg))
|
||||
m.Counter("broker_writer_batch_size_min", labels...).Set(uint64(wstats.BatchSize.Min))
|
||||
m.Counter("broker_writer_batch_size_max", labels...).Set(uint64(wstats.BatchSize.Max))
|
||||
m.Counter("broker_writer_batch_bytes_avg", labels...).Set(uint64(wstats.BatchBytes.Avg))
|
||||
m.Counter("broker_writer_batch_bytes_min", labels...).Set(uint64(wstats.BatchBytes.Min))
|
||||
m.Counter("broker_writer_batch_bytes_max", labels...).Set(uint64(wstats.BatchBytes.Max))
|
||||
*/
|
||||
|
||||
m.Counter("broker_writer_attempts_max", labels...).Set(uint64(wstats.MaxAttempts))
|
||||
m.Counter("broker_writer_batch_max", labels...).Set(uint64(wstats.MaxBatchSize))
|
||||
m.Counter("broker_writer_batch_timeout", labels...).Set(uint64(wstats.BatchTimeout))
|
||||
m.Counter("broker_writer_read_timeout", labels...).Set(uint64(wstats.ReadTimeout))
|
||||
m.Counter("broker_writer_write_timeout", labels...).Set(uint64(wstats.WriteTimeout))
|
||||
m.Counter("broker_writer_acks_required", labels...).Set(uint64(wstats.RequiredAcks))
|
||||
if wstats.Async {
|
||||
m.Counter("broker_writer_async", labels...).Set(1)
|
||||
} else {
|
||||
m.Counter("broker_writer_async", labels...).Set(0)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user