Merge pull request 'rwfix' (#127 ) from rwfix into master

Reviewed-on: #127
fixup panic
2023-12-20 22:57:32 +03:00 · 2023-12-20 22:56:28 +03:00
4 changed files with 10 additions and 9 deletions
--- a/go.mod
+++ b/go.mod
@@ -4,11 +4,11 @@ go 1.19

 require (
 	github.com/twmb/franz-go v1.11.5
-	github.com/twmb/franz-go/pkg/kmsg v1.3.0
-	go.unistack.org/micro/v4 v4.0.2
+	go.unistack.org/micro/v4 v4.0.1
 )

 require (
 	github.com/klauspost/compress v1.15.9 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.3.0 // indirect
 )
--- a/go.sum
+++ b/go.sum
@@ -6,8 +6,8 @@ github.com/twmb/franz-go v1.11.5 h1:TTv5lVJd+87XkmP9dWN9Jgpf7IUUr7a7jee+byR8LBE=
 github.com/twmb/franz-go v1.11.5/go.mod h1:FvaHNlpT6woVYIl6LAuIeL7yHol1Fp6Gv2Dn21AvH78=
 github.com/twmb/franz-go/pkg/kmsg v1.3.0 h1:ouBETB7nTqRxiO5E8/pySoFZtVEW2VWw55z3/bsUzTw=
 github.com/twmb/franz-go/pkg/kmsg v1.3.0/go.mod h1:SxG/xJKhgPu25SamAq0rrucfp7lbzCpEXOC+vH/ELrY=
-go.unistack.org/micro/v4 v4.0.2 h1:2LeG6jslE50c72f1XwJhfTiidy67xklIC3saptLoUys=
-go.unistack.org/micro/v4 v4.0.2/go.mod h1:+wBa98rSf+mRXb/MuSVFPXtDrqN0k8rzPQiC8wRCwCo=
+go.unistack.org/micro/v4 v4.0.1 h1:xo1IxbVfgh8i0eY0VeYa3cbb13u5n/Mxnp3FOgWD4Jo=
+go.unistack.org/micro/v4 v4.0.1/go.mod h1:p/J5UcSJjfHsWGT31uKoghQ5rUQZzQJBAFy+Z4+ZVMs=
 golang.org/x/crypto v0.0.0-20220817201139-bc19a97f63c8/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
--- a/kgo.go
+++ b/kgo.go
@@ -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 {
 	k.RLock()
-	if !k.connected {
-		k.RUnlock()
+	ok := k.connected
+	k.RUnlock()
+
+	if !ok {
 		k.Lock()
 		c, err := k.connect(ctx, k.kopts...)
 		if err != nil {
@@ -208,7 +210,6 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
 		k.connected = true
 		k.Unlock()
 	}
-	k.RUnlock()

 	options := broker.NewPublishOptions(opts...)
 	records := make([]*kgo.Record, 0, len(msgs))
--- a/subscriber.go
+++ b/subscriber.go
@@ -22,7 +22,7 @@ type consumer struct {
 	partition int32
 	opts      broker.SubscribeOptions
 	kopts     broker.Options
-	handler   interface{}
+	handler   broker.Handler
 	quit      chan struct{}
 	done      chan struct{}
 	recs      chan kgo.FetchTopicPartition
@@ -33,7 +33,7 @@ type subscriber struct {
 	topic     string
 	opts      broker.SubscribeOptions
 	kopts     broker.Options
-	handler   interface{}
+	handler   broker.Handler
 	closed    bool
 	done      chan struct{}
 	consumers map[tp]*consumer
Author	SHA1	Message	Date
Vasiliy Tolstov	98da69fbe8	Merge pull request 'rwfix' (#127 ) from rwfix into master Some checks failed build / test (push) Failing after 1m27s Details build / lint (push) Failing after 2m45s Details codeql / analyze (go) (push) Failing after 2m41s Details Reviewed-on: #127	2023-12-20 22:57:32 +03:00
Vasiliy Tolstov	d6d2483d8d	fixup panic Some checks failed codeql / analyze (go) (pull_request) Failing after 2m41s Details prbuild / test (pull_request) Failing after 1m28s Details prbuild / lint (pull_request) Failing after 2m37s Details autoapprove / autoapprove (pull_request) Failing after 1m25s Details automerge / automerge (pull_request) Failing after 4s Details dependabot-automerge / automerge (pull_request) Has been skipped Details Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>	2023-12-20 22:56:28 +03:00