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

export Subscriber
Some checks failed
build / test (push) Failing after 1m38s
Details
codeql / analyze (go) (push) Failing after 1m59s
Details
build / lint (push) Successful in 9m15s
Details

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2024-05-22 18:28:51 +03:00
parent 88777a29ad
commit d404fa31ab
2 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
kgo.go
View File

@ -62,7 +62,7 @@ type Broker struct {
connected bool connected bool
sync.RWMutex sync.RWMutex
opts broker.Options opts broker.Options
subs []*subscriber subs []*Subscriber
} }
func (k *Broker) Address() string { func (k *Broker) Address() string {
@ -364,7 +364,7 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
} }
} }
sub := &subscriber{ sub := &Subscriber{
topic: topic, topic: topic,
opts: options, opts: options,
handler: handler, handler: handler,

20
subscriber.go
View File

@ -33,7 +33,7 @@ type consumer struct {
recs chan kgo.FetchTopicPartition recs chan kgo.FetchTopicPartition
} }
type subscriber struct { type Subscriber struct {
c *kgo.Client c *kgo.Client
topic string topic string
htracer *hookTracer htracer *hookTracer
@ -46,19 +46,19 @@ type subscriber struct {
sync.RWMutex sync.RWMutex
} }
func (s *subscriber) Client() *kgo.Client { func (s *Subscriber) Client() *kgo.Client {
return s.c return s.c
} }
func (s *subscriber) Options() broker.SubscribeOptions { func (s *Subscriber) Options() broker.SubscribeOptions {
return s.opts return s.opts
} }
func (s *subscriber) Topic() string { func (s *Subscriber) Topic() string {
return s.topic return s.topic
} }
func (s *subscriber) Unsubscribe(ctx context.Context) error { func (s *Subscriber) Unsubscribe(ctx context.Context) error {
if s.closed { if s.closed {
return nil return nil
} }
@ -80,7 +80,7 @@ func (s *subscriber) Unsubscribe(ctx context.Context) error {
return nil return nil
} }
func (s *subscriber) poll(ctx context.Context) { func (s *Subscriber) poll(ctx context.Context) {
maxInflight := DefaultSubscribeMaxInflight maxInflight := DefaultSubscribeMaxInflight
if s.opts.Context != nil { if s.opts.Context != nil {
if n, ok := s.opts.Context.Value(subscribeMaxInflightKey{}).(int); n > 0 && ok { if n, ok := s.opts.Context.Value(subscribeMaxInflightKey{}).(int); n > 0 && ok {
@ -148,7 +148,7 @@ func (s *subscriber) poll(ctx context.Context) {
} }
} }
func (s *subscriber) killConsumers(ctx context.Context, lost map[string][]int32) { func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32) {
var wg sync.WaitGroup var wg sync.WaitGroup
defer wg.Wait() defer wg.Wait()
@ -165,12 +165,12 @@ func (s *subscriber) killConsumers(ctx context.Context, lost map[string][]int32)
} }
} }
func (s *subscriber) lost(ctx context.Context, _ *kgo.Client, lost map[string][]int32) { func (s *Subscriber) lost(ctx context.Context, _ *kgo.Client, lost map[string][]int32) {
s.kopts.Logger.Debugf(ctx, "[kgo] lost %#+v", lost) s.kopts.Logger.Debugf(ctx, "[kgo] lost %#+v", lost)
s.killConsumers(ctx, lost) s.killConsumers(ctx, lost)
} }
func (s *subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[string][]int32) { func (s *Subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[string][]int32) {
s.kopts.Logger.Debugf(ctx, "[kgo] revoked %#+v", revoked) s.kopts.Logger.Debugf(ctx, "[kgo] revoked %#+v", revoked)
s.killConsumers(ctx, revoked) s.killConsumers(ctx, revoked)
if err := c.CommitMarkedOffsets(ctx); err != nil { if err := c.CommitMarkedOffsets(ctx); err != nil {
@ -178,7 +178,7 @@ func (s *subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[str
} }
} }
func (s *subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[string][]int32) { func (s *Subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[string][]int32) {
for topic, partitions := range assigned { for topic, partitions := range assigned {
for _, partition := range partitions { for _, partition := range partitions {
pc := &consumer{ pc := &consumer{