[v4] hide access to internal mutex (#185)

* changed embedded mutex to private field

* update ci

* fix tests
This commit is contained in:
2025-05-23 01:45:39 +05:00
committed by Vasiliy Tolstov
parent 44e2d5f9a4
commit b538ef82b5
2 changed files with 17 additions and 17 deletions

22
kgo.go
View File

@@ -74,7 +74,7 @@ type Broker struct {
opts broker.Options
sync.RWMutex
mu sync.RWMutex
init bool
}
@@ -141,9 +141,9 @@ func (b *Broker) newCodec(ct string) (codec.Codec, error) {
if idx := strings.IndexRune(ct, ';'); idx >= 0 {
ct = ct[:idx]
}
b.RLock()
b.mu.RLock()
c, ok := b.opts.Codecs[ct]
b.RUnlock()
b.mu.RUnlock()
if ok {
return c, nil
}
@@ -238,10 +238,10 @@ func (k *Broker) Connect(ctx context.Context) error {
return err
}
k.Lock()
k.mu.Lock()
k.c = c
k.connected.Store(1)
k.Unlock()
k.mu.Unlock()
return nil
}
@@ -259,8 +259,8 @@ func (k *Broker) Disconnect(ctx context.Context) error {
ctx, span = k.opts.Tracer.Start(ctx, "Disconnect")
defer span.Finish()
k.Lock()
defer k.Unlock()
k.mu.Lock()
defer k.mu.Unlock()
select {
case <-nctx.Done():
return nctx.Err()
@@ -284,8 +284,8 @@ func (k *Broker) Disconnect(ctx context.Context) error {
}
func (k *Broker) Init(opts ...broker.Option) error {
k.Lock()
defer k.Unlock()
k.mu.Lock()
defer k.mu.Unlock()
if len(opts) == 0 && k.init {
return nil
@@ -538,9 +538,9 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
go sub.poll(ctx)
b.Lock()
b.mu.Lock()
b.subs = append(b.subs, sub)
b.Unlock()
b.mu.Unlock()
return sub, nil
}