Compare commits
38 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bec11f7633 | |||
|
a6f7fb0e35 | ||
| 3bf3012b88 | |||
|
|
5d0f687a8b | ||
| 49e7bca9df | |||
|
|
9db3acc793 | ||
| 7312397abe | |||
| 7676631737 | |||
| 6c2cf494ca | |||
| 6e5e2e0338 | |||
| 6dd3b4548a | |||
| fff768dc2a | |||
| 755bd187ba | |||
| 141c7fb848 | |||
|
f347ca4e12 | ||
| c78bffdb8d | |||
| 49ba8880f2 | |||
|
|
bbd840b96e | ||
| 1ee739de80 | |||
|
|
7f0265b6d1 | ||
| 4dc19dc63f | |||
|
|
5eb2718cd4 | ||
| b4c98e207f | |||
|
|
20652894b3 | ||
| 27bb8c50d1 | |||
|
|
b90ef8cdf3 | ||
| 222370f96c | |||
|
|
1152be720d | ||
| f0b9665816 | |||
|
|
310e9a8ceb | ||
| cc155511e1 | |||
|
|
0a0c592b64 | ||
| 372c5c92f0 | |||
|
|
2601514578 | ||
| 8a1856e814 | |||
|
|
08e04cdb70 | ||
| 7286521472 | |||
|
|
b01e2b5563 |
29
.gitea/workflows/job_lint.yml
Normal file
29
.gitea/workflows/job_lint.yml
Normal file
@ -0,0 +1,29 @@
|
||||
name: lint
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
types: [opened, reopened, synchronize]
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
- v4
|
||||
|
||||
jobs:
|
||||
lint:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout code
|
||||
uses: actions/checkout@v4
|
||||
with:
|
||||
filter: 'blob:none'
|
||||
- name: setup go
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
cache-dependency-path: "**/*.sum"
|
||||
go-version: 'stable'
|
||||
- name: setup deps
|
||||
run: go get -v ./...
|
||||
- name: run lint
|
||||
uses: https://github.com/golangci/golangci-lint-action@v6
|
||||
with:
|
||||
version: 'latest'
|
||||
34
.gitea/workflows/job_test.yml
Normal file
34
.gitea/workflows/job_test.yml
Normal file
@ -0,0 +1,34 @@
|
||||
name: test
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
types: [opened, reopened, synchronize]
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
- v4
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
- v4
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout code
|
||||
uses: actions/checkout@v4
|
||||
with:
|
||||
filter: 'blob:none'
|
||||
- name: setup go
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
cache-dependency-path: "**/*.sum"
|
||||
go-version: 'stable'
|
||||
- name: setup deps
|
||||
run: go get -v ./...
|
||||
- name: run test
|
||||
env:
|
||||
INTEGRATION_TESTS: yes
|
||||
run: go test -mod readonly -v ./...
|
||||
53
.gitea/workflows/job_tests.yml
Normal file
53
.gitea/workflows/job_tests.yml
Normal file
@ -0,0 +1,53 @@
|
||||
name: test
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
types: [opened, reopened, synchronize]
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
- v4
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
- v4
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout code
|
||||
uses: actions/checkout@v4
|
||||
with:
|
||||
filter: 'blob:none'
|
||||
- name: checkout tests
|
||||
uses: actions/checkout@v4
|
||||
with:
|
||||
ref: master
|
||||
filter: 'blob:none'
|
||||
repository: unistack-org/micro-tests
|
||||
path: micro-tests
|
||||
- name: setup go
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
cache-dependency-path: "**/*.sum"
|
||||
go-version: 'stable'
|
||||
- name: setup go work
|
||||
env:
|
||||
GOWORK: /workspace/${{ github.repository_owner }}/go.work
|
||||
run: |
|
||||
go work init
|
||||
go work use .
|
||||
go work use micro-tests
|
||||
- name: setup deps
|
||||
env:
|
||||
GOWORK: /workspace/${{ github.repository_owner }}/go.work
|
||||
run: go get -v ./...
|
||||
- name: run tests
|
||||
env:
|
||||
INTEGRATION_TESTS: yes
|
||||
GOWORK: /workspace/${{ github.repository_owner }}/go.work
|
||||
run: |
|
||||
cd micro-tests
|
||||
go test -mod readonly -v ./... || true
|
||||
19
.github/dependabot.yml
vendored
19
.github/dependabot.yml
vendored
@ -1,19 +0,0 @@
|
||||
# To get started with Dependabot version updates, you'll need to specify which
|
||||
# package ecosystems to update and where the package manifests are located.
|
||||
# Please see the documentation for all configuration options:
|
||||
# https://help.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
|
||||
|
||||
version: 2
|
||||
updates:
|
||||
|
||||
# Maintain dependencies for GitHub Actions
|
||||
- package-ecosystem: "github-actions"
|
||||
directory: "/"
|
||||
schedule:
|
||||
interval: "daily"
|
||||
|
||||
# Maintain dependencies for Golang
|
||||
- package-ecosystem: "gomod"
|
||||
directory: "/"
|
||||
schedule:
|
||||
interval: "daily"
|
||||
20
.github/workflows/autoapprove.yml
vendored
20
.github/workflows/autoapprove.yml
vendored
@ -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
21
.github/workflows/automerge.yml
vendored
@ -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}}
|
||||
47
.github/workflows/build.yml
vendored
47
.github/workflows/build.yml
vendored
@ -1,47 +0,0 @@
|
||||
name: build
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
jobs:
|
||||
test:
|
||||
name: test
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: setup
|
||||
uses: actions/setup-go@v3
|
||||
with:
|
||||
go-version: 1.17
|
||||
- name: checkout
|
||||
uses: actions/checkout@v3
|
||||
- name: cache
|
||||
uses: actions/cache@v3
|
||||
with:
|
||||
path: ~/go/pkg/mod
|
||||
key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
|
||||
restore-keys: ${{ runner.os }}-go-
|
||||
- name: deps
|
||||
run: go get -v -t -d ./...
|
||||
- name: test
|
||||
env:
|
||||
INTEGRATION_TESTS: yes
|
||||
run: go test -mod readonly -v ./...
|
||||
lint:
|
||||
name: lint
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout
|
||||
uses: actions/checkout@v3
|
||||
- name: lint
|
||||
uses: golangci/golangci-lint-action@v3.4.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.
|
||||
version: v1.30
|
||||
# Optional: working directory, useful for monorepos
|
||||
# working-directory: somedir
|
||||
# Optional: golangci-lint command line arguments.
|
||||
# args: --issues-exit-code=0
|
||||
# Optional: show only new issues if it's a pull request. The default value is `false`.
|
||||
# only-new-issues: true
|
||||
78
.github/workflows/codeql-analysis.yml
vendored
78
.github/workflows/codeql-analysis.yml
vendored
@ -1,78 +0,0 @@
|
||||
# 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@v3
|
||||
with:
|
||||
go-version: 1.17
|
||||
# Initializes the CodeQL tools for scanning.
|
||||
- name: init
|
||||
uses: github/codeql-action/init@v2
|
||||
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@v2
|
||||
|
||||
# ℹ️ 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@v2
|
||||
27
.github/workflows/dependabot-automerge.yml
vendored
27
.github/workflows/dependabot-automerge.yml
vendored
@ -1,27 +0,0 @@
|
||||
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.6
|
||||
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}}
|
||||
47
.github/workflows/pr.yml
vendored
47
.github/workflows/pr.yml
vendored
@ -1,47 +0,0 @@
|
||||
name: prbuild
|
||||
on:
|
||||
pull_request:
|
||||
branches:
|
||||
- master
|
||||
- v3
|
||||
jobs:
|
||||
test:
|
||||
name: test
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: setup
|
||||
uses: actions/setup-go@v3
|
||||
with:
|
||||
go-version: 1.17
|
||||
- name: checkout
|
||||
uses: actions/checkout@v3
|
||||
- name: cache
|
||||
uses: actions/cache@v3
|
||||
with:
|
||||
path: ~/go/pkg/mod
|
||||
key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
|
||||
restore-keys: ${{ runner.os }}-go-
|
||||
- name: deps
|
||||
run: go get -v -t -d ./...
|
||||
- name: test
|
||||
env:
|
||||
INTEGRATION_TESTS: yes
|
||||
run: go test -mod readonly -v ./...
|
||||
lint:
|
||||
name: lint
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: checkout
|
||||
uses: actions/checkout@v3
|
||||
- name: lint
|
||||
uses: golangci/golangci-lint-action@v3.4.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.
|
||||
version: v1.30
|
||||
# Optional: working directory, useful for monorepos
|
||||
# working-directory: somedir
|
||||
# Optional: golangci-lint command line arguments.
|
||||
# args: --issues-exit-code=0
|
||||
# Optional: show only new issues if it's a pull request. The default value is `false`.
|
||||
# only-new-issues: true
|
||||
5
.golangci.yml
Normal file
5
.golangci.yml
Normal file
@ -0,0 +1,5 @@
|
||||
run:
|
||||
concurrency: 8
|
||||
deadline: 5m
|
||||
issues-exit-code: 1
|
||||
tests: true
|
||||
2
event.go
2
event.go
@ -3,7 +3,7 @@ package kgo
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v4/broker"
|
||||
)
|
||||
|
||||
type event struct {
|
||||
|
||||
6
go.mod
6
go.mod
@ -1,11 +1,11 @@
|
||||
module go.unistack.org/micro-broker-kgo/v3
|
||||
module go.unistack.org/micro-broker-kgo/v4
|
||||
|
||||
go 1.17
|
||||
go 1.19
|
||||
|
||||
require (
|
||||
github.com/twmb/franz-go v1.11.5
|
||||
github.com/twmb/franz-go/pkg/kmsg v1.3.0
|
||||
go.unistack.org/micro/v3 v3.10.14
|
||||
go.unistack.org/micro/v4 v4.0.2
|
||||
)
|
||||
|
||||
require (
|
||||
|
||||
10
go.sum
10
go.sum
@ -1,16 +1,13 @@
|
||||
github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
|
||||
github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
|
||||
github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
|
||||
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
|
||||
github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
|
||||
github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
|
||||
github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
|
||||
github.com/twmb/franz-go v1.11.5 h1:TTv5lVJd+87XkmP9dWN9Jgpf7IUUr7a7jee+byR8LBE=
|
||||
github.com/twmb/franz-go v1.11.5/go.mod h1:FvaHNlpT6woVYIl6LAuIeL7yHol1Fp6Gv2Dn21AvH78=
|
||||
github.com/twmb/franz-go/pkg/kmsg v1.3.0 h1:ouBETB7nTqRxiO5E8/pySoFZtVEW2VWw55z3/bsUzTw=
|
||||
github.com/twmb/franz-go/pkg/kmsg v1.3.0/go.mod h1:SxG/xJKhgPu25SamAq0rrucfp7lbzCpEXOC+vH/ELrY=
|
||||
go.unistack.org/micro/v3 v3.10.14 h1:7fgLpwGlCN67twhwtngJDEQvrMkUBDSA5vzZqxIDqNE=
|
||||
go.unistack.org/micro/v3 v3.10.14/go.mod h1:uMAc0U/x7dmtICCrblGf0ZLgYegu3VwQAquu+OFCw1Q=
|
||||
go.unistack.org/micro/v4 v4.0.2 h1:2LeG6jslE50c72f1XwJhfTiidy67xklIC3saptLoUys=
|
||||
go.unistack.org/micro/v4 v4.0.2/go.mod h1:+wBa98rSf+mRXb/MuSVFPXtDrqN0k8rzPQiC8wRCwCo=
|
||||
golang.org/x/crypto v0.0.0-20220817201139-bc19a97f63c8/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
|
||||
golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
@ -19,6 +16,3 @@ golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBc
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
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=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
||||
61
kgo.go
61
kgo.go
@ -1,5 +1,5 @@
|
||||
// Package kgo provides a kafka broker using kgo
|
||||
package kgo // import "go.unistack.org/micro-broker-kgo/v3"
|
||||
package kgo // import "go.unistack.org/micro-broker-kgo/v4"
|
||||
|
||||
import (
|
||||
"context"
|
||||
@ -11,13 +11,14 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/twmb/franz-go/pkg/kgo"
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/metadata"
|
||||
id "go.unistack.org/micro/v3/util/id"
|
||||
mrand "go.unistack.org/micro/v3/util/rand"
|
||||
"github.com/twmb/franz-go/pkg/kmsg"
|
||||
"go.unistack.org/micro/v4/broker"
|
||||
"go.unistack.org/micro/v4/metadata"
|
||||
id "go.unistack.org/micro/v4/util/id"
|
||||
mrand "go.unistack.org/micro/v4/util/rand"
|
||||
)
|
||||
|
||||
var _ broker.Broker = &Broker{}
|
||||
var _ broker.Broker = (*Broker)(nil)
|
||||
|
||||
var ErrLostMessage = errors.New("message not marked for offsets commit and will be lost in next iteration")
|
||||
|
||||
@ -55,7 +56,6 @@ type Broker struct {
|
||||
c *kgo.Client
|
||||
kopts []kgo.Opt
|
||||
connected bool
|
||||
init bool
|
||||
sync.RWMutex
|
||||
opts broker.Options
|
||||
subs []*subscriber
|
||||
@ -134,6 +134,9 @@ func (k *Broker) Disconnect(ctx context.Context) error {
|
||||
return nctx.Err()
|
||||
default:
|
||||
for _, sub := range k.subs {
|
||||
if sub.closed {
|
||||
continue
|
||||
}
|
||||
if err := sub.Unsubscribe(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
@ -212,16 +215,21 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
|
||||
var errs []string
|
||||
var err error
|
||||
var key []byte
|
||||
var promise func(*kgo.Record, error)
|
||||
|
||||
if options.Context != nil {
|
||||
if k, ok := options.Context.Value(publishKey{}).([]byte); ok && k != nil {
|
||||
key = k
|
||||
}
|
||||
if p, ok := options.Context.Value(publishPromiseKey{}).(func(*kgo.Record, error)); ok && p != nil {
|
||||
promise = p
|
||||
}
|
||||
}
|
||||
|
||||
for _, msg := range msgs {
|
||||
rec := &kgo.Record{Context: ctx, Key: key}
|
||||
rec.Topic, _ = msg.Header.Get(metadata.HeaderTopic)
|
||||
k.opts.Meter.Counter(broker.PublishMessageInflight, "endpoint", rec.Topic).Inc()
|
||||
if options.BodyOnly {
|
||||
rec.Value = msg.Body
|
||||
} else if k.opts.Codec.String() == "noop" {
|
||||
@ -238,10 +246,36 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
|
||||
records = append(records, rec)
|
||||
}
|
||||
|
||||
if promise != nil {
|
||||
ts := time.Now()
|
||||
for _, rec := range records {
|
||||
k.c.Produce(ctx, rec, func(r *kgo.Record, err error) {
|
||||
te := time.Since(ts)
|
||||
k.opts.Meter.Counter(broker.PublishMessageInflight, "endpoint", rec.Topic).Dec()
|
||||
k.opts.Meter.Summary(broker.PublishMessageLatencyMicroseconds, "endpoint", r.Topic).Update(te.Seconds())
|
||||
k.opts.Meter.Histogram(broker.PublishMessageDurationSeconds, "endpoint", r.Topic).Update(te.Seconds())
|
||||
if err != nil {
|
||||
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", r.Topic, "status", "failure").Inc()
|
||||
} else {
|
||||
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", r.Topic, "status", "success").Inc()
|
||||
}
|
||||
promise(r, err)
|
||||
})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
ts := time.Now()
|
||||
results := k.c.ProduceSync(ctx, records...)
|
||||
te := time.Since(ts)
|
||||
for _, result := range results {
|
||||
k.opts.Meter.Summary(broker.PublishMessageLatencyMicroseconds, "endpoint", result.Record.Topic).Update(te.Seconds())
|
||||
k.opts.Meter.Histogram(broker.PublishMessageDurationSeconds, "endpoint", result.Record.Topic).Update(te.Seconds())
|
||||
k.opts.Meter.Counter(broker.PublishMessageInflight, "endpoint", result.Record.Topic).Dec()
|
||||
if result.Err != nil {
|
||||
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", result.Record.Topic, "status", "failure").Inc()
|
||||
errs = append(errs, result.Err.Error())
|
||||
} else {
|
||||
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", result.Record.Topic, "status", "success").Inc()
|
||||
}
|
||||
}
|
||||
|
||||
@ -306,6 +340,18 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
|
||||
return nil, err
|
||||
}
|
||||
|
||||
mdreq := kmsg.NewMetadataRequest()
|
||||
mdreq.Topics = []kmsg.MetadataRequestTopic{
|
||||
{Topic: &topic},
|
||||
}
|
||||
|
||||
mdrsp, err := mdreq.RequestWith(ctx, c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if mdrsp.Topics[0].ErrorCode != 0 {
|
||||
return nil, fmt.Errorf("topic %s not exists or permission error", topic)
|
||||
}
|
||||
|
||||
sub.c = c
|
||||
go sub.poll(ctx)
|
||||
|
||||
@ -320,7 +366,6 @@ func (k *Broker) String() string {
|
||||
}
|
||||
|
||||
func NewBroker(opts ...broker.Option) *Broker {
|
||||
rand.Seed(time.Now().Unix())
|
||||
options := broker.NewOptions(opts...)
|
||||
|
||||
kaddrs := options.Addrs
|
||||
|
||||
@ -10,10 +10,10 @@ import (
|
||||
"time"
|
||||
|
||||
kg "github.com/twmb/franz-go/pkg/kgo"
|
||||
kgo "go.unistack.org/micro-broker-kgo/v3"
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v3/metadata"
|
||||
kgo "go.unistack.org/micro-broker-kgo/v4"
|
||||
"go.unistack.org/micro/v4/broker"
|
||||
"go.unistack.org/micro/v4/logger"
|
||||
"go.unistack.org/micro/v4/metadata"
|
||||
)
|
||||
|
||||
var (
|
||||
|
||||
@ -5,7 +5,7 @@ import (
|
||||
"fmt"
|
||||
|
||||
"github.com/twmb/franz-go/pkg/kgo"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v4/logger"
|
||||
)
|
||||
|
||||
type mlogger struct {
|
||||
|
||||
37
metrics.go
37
metrics.go
@ -6,9 +6,44 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/twmb/franz-go/pkg/kgo"
|
||||
"go.unistack.org/micro/v3/meter"
|
||||
"go.unistack.org/micro/v4/meter"
|
||||
)
|
||||
|
||||
/*
|
||||
func NewServerSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
|
||||
handler := &wrapper{
|
||||
opts: NewOptions(opts...),
|
||||
}
|
||||
return handler.SubscriberFunc
|
||||
}
|
||||
|
||||
func (w *wrapper) SubscriberFunc(fn server.SubscriberFunc) server.SubscriberFunc {
|
||||
return func(ctx context.Context, msg server.Message) error {
|
||||
endpoint := msg.Topic()
|
||||
|
||||
labels := make([]string, 0, 4)
|
||||
labels = append(labels, labelEndpoint, endpoint)
|
||||
|
||||
w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Inc()
|
||||
ts := time.Now()
|
||||
err := fn(ctx, msg)
|
||||
te := time.Since(ts)
|
||||
w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Dec()
|
||||
|
||||
w.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, labels...).Update(te.Seconds())
|
||||
w.opts.Meter.Histogram(SubscribeMessageDurationSeconds, labels...).Update(te.Seconds())
|
||||
|
||||
if err == nil {
|
||||
labels = append(labels, labelStatus, labelSuccess)
|
||||
} else {
|
||||
labels = append(labels, labelStatus, labelFailure)
|
||||
}
|
||||
w.opts.Meter.Counter(SubscribeMessageTotal, labels...).Inc()
|
||||
|
||||
return err
|
||||
}
|
||||
}
|
||||
*/
|
||||
type metrics struct {
|
||||
meter meter.Meter
|
||||
}
|
||||
|
||||
16
options.go
16
options.go
@ -5,8 +5,8 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/twmb/franz-go/pkg/kgo"
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/client"
|
||||
"go.unistack.org/micro/v4/broker"
|
||||
"go.unistack.org/micro/v4/client"
|
||||
)
|
||||
|
||||
// DefaultCommitInterval specifies how fast send commit offsets to kafka
|
||||
@ -78,3 +78,15 @@ type subscribeMaxInflightKey struct{}
|
||||
func SubscribeMaxInFlight(n int) broker.SubscribeOption {
|
||||
return broker.SetSubscribeOption(subscribeMaxInflightKey{}, n)
|
||||
}
|
||||
|
||||
type publishPromiseKey struct{}
|
||||
|
||||
// PublishPromise set the kafka promise func for Produce
|
||||
func PublishPromise(fn func(*kgo.Record, error)) broker.PublishOption {
|
||||
return broker.SetPublishOption(publishPromiseKey{}, fn)
|
||||
}
|
||||
|
||||
// ClientPublishKey set the kafka message key (client option)
|
||||
func ClientPublishPromise(fn func(*kgo.Record, error)) client.PublishOption {
|
||||
return client.SetPublishOption(publishPromiseKey{}, fn)
|
||||
}
|
||||
|
||||
@ -3,11 +3,12 @@ package kgo
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/twmb/franz-go/pkg/kgo"
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v3/metadata"
|
||||
"go.unistack.org/micro/v4/broker"
|
||||
"go.unistack.org/micro/v4/logger"
|
||||
"go.unistack.org/micro/v4/metadata"
|
||||
)
|
||||
|
||||
type tp struct {
|
||||
@ -21,7 +22,7 @@ type consumer struct {
|
||||
partition int32
|
||||
opts broker.SubscribeOptions
|
||||
kopts broker.Options
|
||||
handler broker.Handler
|
||||
handler interface{}
|
||||
quit chan struct{}
|
||||
done chan struct{}
|
||||
recs chan kgo.FetchTopicPartition
|
||||
@ -32,7 +33,7 @@ type subscriber struct {
|
||||
topic string
|
||||
opts broker.SubscribeOptions
|
||||
kopts broker.Options
|
||||
handler broker.Handler
|
||||
handler interface{}
|
||||
closed bool
|
||||
done chan struct{}
|
||||
consumers map[tp]*consumer
|
||||
@ -55,6 +56,12 @@ func (s *subscriber) Unsubscribe(ctx context.Context) error {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
s.c.PauseFetchTopics(s.topic)
|
||||
kc := make(map[string][]int32)
|
||||
for ctp := range s.consumers {
|
||||
kc[ctp.t] = append(kc[ctp.t], ctp.p)
|
||||
}
|
||||
s.killConsumers(ctx, kc)
|
||||
close(s.done)
|
||||
s.closed = true
|
||||
}
|
||||
@ -71,15 +78,14 @@ func (s *subscriber) poll(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
s.c.Close()
|
||||
s.c.CloseAllowingRebalance()
|
||||
return
|
||||
case <-s.done:
|
||||
s.c.Close()
|
||||
s.c.CloseAllowingRebalance()
|
||||
return
|
||||
default:
|
||||
fetches := s.c.PollRecords(ctx, maxInflight)
|
||||
if fetches.IsClientClosed() {
|
||||
s.kopts.Logger.Errorf(ctx, "[kgo] client closed")
|
||||
if !s.closed && fetches.IsClientClosed() {
|
||||
s.closed = true
|
||||
return
|
||||
}
|
||||
@ -163,6 +169,8 @@ func (pc *consumer) consume() {
|
||||
return
|
||||
case p := <-pc.recs:
|
||||
for _, record := range p.Records {
|
||||
ts := time.Now()
|
||||
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Inc()
|
||||
p := eventPool.Get().(*event)
|
||||
p.msg.Header = nil
|
||||
p.msg.Body = nil
|
||||
@ -179,30 +187,45 @@ func (pc *consumer) consume() {
|
||||
p.msg.Body = record.Value
|
||||
} else {
|
||||
if err := pc.kopts.Codec.Unmarshal(record.Value, p.msg); err != nil {
|
||||
pc.kopts.Meter.Counter(broker.SubscribeMessageTotal, "endpoint", record.Topic, "status", "failure").Inc()
|
||||
p.err = err
|
||||
p.msg.Body = record.Value
|
||||
if eh != nil {
|
||||
_ = eh(p)
|
||||
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Dec()
|
||||
if p.ack {
|
||||
pc.c.MarkCommitRecords(record)
|
||||
} else {
|
||||
eventPool.Put(p)
|
||||
pc.kopts.Logger.Infof(pc.kopts.Context, "[kgo] ErrLostMessage wtf?")
|
||||
pc.kopts.Logger.Fatalf(pc.kopts.Context, "[kgo] ErrLostMessage wtf?")
|
||||
return
|
||||
}
|
||||
eventPool.Put(p)
|
||||
te := time.Since(ts)
|
||||
pc.kopts.Meter.Summary(broker.SubscribeMessageLatencyMicroseconds, "endpoint", record.Topic).Update(te.Seconds())
|
||||
pc.kopts.Meter.Histogram(broker.SubscribeMessageDurationSeconds, "endpoint", record.Topic).Update(te.Seconds())
|
||||
continue
|
||||
} else {
|
||||
if pc.kopts.Logger.V(logger.ErrorLevel) {
|
||||
pc.kopts.Logger.Errorf(pc.kopts.Context, "[kgo]: failed to unmarshal: %v", err)
|
||||
}
|
||||
}
|
||||
te := time.Since(ts)
|
||||
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Dec()
|
||||
pc.kopts.Meter.Summary(broker.SubscribeMessageLatencyMicroseconds, "endpoint", record.Topic).Update(te.Seconds())
|
||||
pc.kopts.Meter.Histogram(broker.SubscribeMessageDurationSeconds, "endpoint", record.Topic).Update(te.Seconds())
|
||||
eventPool.Put(p)
|
||||
pc.kopts.Logger.Infof(pc.kopts.Context, "[kgo] Unmarshal err not handled wtf?")
|
||||
pc.kopts.Logger.Fatalf(pc.kopts.Context, "[kgo] Unmarshal err not handled wtf?")
|
||||
return
|
||||
}
|
||||
}
|
||||
err := pc.handler(p)
|
||||
if err == nil {
|
||||
pc.kopts.Meter.Counter(broker.SubscribeMessageTotal, "endpoint", record.Topic, "status", "success").Inc()
|
||||
} else {
|
||||
pc.kopts.Meter.Counter(broker.SubscribeMessageTotal, "endpoint", record.Topic, "status", "failure").Inc()
|
||||
}
|
||||
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Dec()
|
||||
if err == nil && pc.opts.AutoAck {
|
||||
p.ack = true
|
||||
} else if err != nil {
|
||||
@ -215,6 +238,9 @@ func (pc *consumer) consume() {
|
||||
}
|
||||
}
|
||||
}
|
||||
te := time.Since(ts)
|
||||
pc.kopts.Meter.Summary(broker.SubscribeMessageLatencyMicroseconds, "endpoint", record.Topic).Update(te.Seconds())
|
||||
pc.kopts.Meter.Histogram(broker.SubscribeMessageDurationSeconds, "endpoint", record.Topic).Update(te.Seconds())
|
||||
if p.ack {
|
||||
eventPool.Put(p)
|
||||
pc.c.MarkCommitRecords(record)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user