unistack-org/micro-broker-kgo
4
0
Fork 2
Code Issues 4 Pull Requests 3 Actions 2 Packages Projects Releases 83 Wiki Activity

remove noisy logs from event hooks and add tests (#192)
Some checks failed
coverage / build (push) Failing after 2m7s
Details
test / test (push) Failing after 19m17s
Details

Browse Source
This commit is contained in:
Абильмансур Мансуров
2025-07-10 12:49:49 +05:00
committed by GitHub
parent aabfecfd3d
commit d5d1e26d7b
5 changed files with 416 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
broker.go
View File

@@ -1,71 +0,0 @@
package kgo
import (
"context"
"net"
"sync/atomic"
"time"
"github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v3/logger"
)
type hookEvent struct {
log logger.Logger
connected *atomic.Uint32
fatalOnError bool
}
var (
_ kgo.HookBrokerConnect = &hookEvent{}
_ kgo.HookBrokerDisconnect = &hookEvent{}
_ kgo.HookBrokerRead = &hookEvent{}
_ kgo.HookBrokerWrite = &hookEvent{}
_ kgo.HookGroupManageError = &hookEvent{}
_ kgo.HookProduceRecordUnbuffered = &hookEvent{}
)
func (m *hookEvent) OnGroupManageError(err error) {
if err != nil {
// m.connected.Store(0)
// if m.fatalOnError {
m.log.Error(context.TODO(), "kgo.OnGroupManageError", err)
//}
}
}
func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
if err != nil {
// m.connected.Store(0)
// if m.fatalOnError {
m.log.Error(context.TODO(), "kgo.OnBrokerConnect", err)
//}
}
}
func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) {
// m.connected.Store(0)
}
func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
if err != nil {
// m.connected.Store(0)
// if m.fatalOnError {
m.log.Error(context.TODO(), "kgo.OnBrokerWrite", err)
//}
}
}
func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
if err != nil {
// m.connected.Store(0)
m.log.Error(context.TODO(), "kgo.OnBrokerRead", err)
}
}
func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) {
if err != nil {
// m.connected.Store(0)
m.log.Error(context.TODO(), "kgo.OnProduceRecordUnbuffered", err)
}
}