109 lines
2.4 KiB
Go
109 lines
2.4 KiB
Go
package kgo
|
|
|
|
import (
|
|
"context"
|
|
"net"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/twmb/franz-go/pkg/kgo"
|
|
"go.unistack.org/micro/v4/logger"
|
|
)
|
|
|
|
type hookEvent struct {
|
|
log logger.Logger
|
|
fatalOnError bool
|
|
connected *atomic.Uint32
|
|
}
|
|
|
|
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) {
|
|
switch {
|
|
case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
|
|
return
|
|
default:
|
|
ctx := context.TODO()
|
|
logMsg := "kgo.OnGroupManageError"
|
|
|
|
if m.fatalOnError {
|
|
m.log.Fatal(ctx, logMsg, err)
|
|
} else {
|
|
m.log.Error(ctx, logMsg, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
|
|
switch {
|
|
case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
|
|
return
|
|
default:
|
|
ctx := context.TODO()
|
|
logMsg := "kgo.OnBrokerConnect"
|
|
|
|
if m.fatalOnError {
|
|
m.log.Fatal(ctx, logMsg, err)
|
|
} else {
|
|
m.log.Error(ctx, logMsg, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) {}
|
|
|
|
func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
|
|
switch {
|
|
case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
|
|
return
|
|
default:
|
|
ctx := context.TODO()
|
|
logMsg := "kgo.OnBrokerWrite"
|
|
|
|
if m.fatalOnError {
|
|
m.log.Fatal(ctx, logMsg, err)
|
|
} else {
|
|
m.log.Error(ctx, logMsg, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
|
|
switch {
|
|
case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
|
|
return
|
|
default:
|
|
ctx := context.TODO()
|
|
logMsg := "kgo.OnBrokerRead"
|
|
|
|
if m.fatalOnError {
|
|
m.log.Fatal(ctx, logMsg, err)
|
|
} else {
|
|
m.log.Error(ctx, logMsg, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) {
|
|
switch {
|
|
case err == nil || isContextError(err) || kgo.IsRetryableBrokerErr(err):
|
|
return
|
|
default:
|
|
ctx := context.TODO()
|
|
logMsg := "kgo.OnProduceRecordUnbuffered"
|
|
|
|
if m.fatalOnError {
|
|
m.log.Fatal(ctx, logMsg, err)
|
|
} else {
|
|
m.log.Error(ctx, logMsg, err)
|
|
}
|
|
}
|
|
}
|