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

update deps && structs && hooks

Browse Source
This commit is contained in:
Денис Евстигнеев
2024-12-19 13:21:32 +03:00
committed by devstigneev
parent 07fa36d704
commit 7c5da60556
6 changed files with 71 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
kgo.go
View File

@@ -5,7 +5,7 @@ import (
"context"
"errors"
"fmt"
"math/rand"
"math/rand/v2"
"net/http"
"strings"
"sync"
@@ -29,7 +29,6 @@ var ErrLostMessage = errors.New("message not marked for offsets commit and will
var DefaultRetryBackoffFn = func() func(int) time.Duration {
var rngMu sync.Mutex
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
return func(fails int) time.Duration {
const (
min = 100 * time.Millisecond
@@ -45,7 +44,7 @@ var DefaultRetryBackoffFn = func() func(int) time.Duration {
backoff := min * time.Duration(1<<(fails-1))
rngMu.Lock()
jitter := 0.8 + 0.4*rng.Float64()
jitter := 0.8 + 0.4*rand.Float64()
rngMu.Unlock()
backoff = time.Duration(float64(backoff) * jitter)
@@ -58,13 +57,16 @@ var DefaultRetryBackoffFn = func() func(int) time.Duration {
}()
type Broker struct {
init bool
c *kgo.Client
kopts []kgo.Opt
c *kgo.Client
connected *atomic.Uint32
sync.RWMutex
kopts []kgo.Opt
subs []*subscriber
opts broker.Options
subs []*Subscriber
sync.RWMutex
init bool
}
func (r *Broker) Live() bool {
@@ -302,11 +304,11 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
k.opts.Meter.Counter(semconv.PublishMessageInflight, "endpoint", rec.Topic, "topic", rec.Topic).Dec()
k.opts.Meter.Summary(semconv.PublishMessageLatencyMicroseconds, "endpoint", rec.Topic, "topic", rec.Topic).Update(te.Seconds())
k.opts.Meter.Histogram(semconv.PublishMessageDurationSeconds, "endpoint", rec.Topic, "topic", rec.Topic).Update(te.Seconds())
if err != nil {
// if err != nil {
k.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", rec.Topic, "topic", rec.Topic, "status", "failure").Inc()
} else {
k.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", rec.Topic, "topic", rec.Topic, "status", "success").Inc()
}
// } else {
k.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", rec.Topic, "topic", rec.Topic, "status", "success").Inc()
// }
promise(r, err)
})
}
@@ -322,9 +324,9 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
if result.Err != nil {
k.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", result.Record.Topic, "topic", result.Record.Topic, "status", "failure").Inc()
errs = append(errs, result.Err.Error())
} else {
} // else {
k.opts.Meter.Counter(semconv.PublishMessageTotal, "endpoint", result.Record.Topic, "topic", result.Record.Topic, "status", "success").Inc()
}
// }
}
if len(errs) > 0 {
@@ -350,7 +352,7 @@ func (k *Broker) TopicExists(ctx context.Context, topic string) error {
return nil
}
func (k *Broker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
func (k *Broker) BatchSubscribe(_ context.Context, _ string, _ broker.BatchHandler, _ ...broker.SubscribeOption) (broker.Subscriber, error) {
return nil, nil
}