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

rwfix #127

Merged
vtolstov merged 2 commits from rwfix into master 2023-12-20 22:57:33 +03:00
Conversation 0 Commits 2 Files Changed 2 +4 -3
1 changed files with 4 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit d6d2483d8d - Show all commits

7
kgo.go
View File

@ -196,8 +196,10 @@ func (k *Broker) Publish(ctx context.Context, topic string, msg *broker.Message,
func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error { func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
k.RLock() k.RLock()
if !k.connected { ok := k.connected
k.RUnlock() k.RUnlock()
if !ok {
k.Lock() k.Lock()
c, err := k.connect(ctx, k.kopts...) c, err := k.connect(ctx, k.kopts...)
if err != nil { if err != nil {
@ -208,7 +210,6 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
k.connected = true k.connected = true
k.Unlock() k.Unlock()
} }
k.RUnlock()
options := broker.NewPublishOptions(opts...) options := broker.NewPublishOptions(opts...)
records := make([]*kgo.Record, 0, len(msgs)) records := make([]*kgo.Record, 0, len(msgs))