Compare commits

...

4 Commits

Author SHA1 Message Date
f4f8793686 fixup connected status
All checks were successful
test / test (push) Successful in 2m25s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-03-07 11:45:36 +03:00
ec4922ad8b fixup connected status
Some checks failed
test / test (push) Has been cancelled
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-03-07 11:44:56 +03:00
eaea14e5a8 fixup connected status
All checks were successful
test / test (push) Successful in 3m3s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-03-07 11:26:34 +03:00
f6f7139d2f fixup panic
All checks were successful
test / test (push) Successful in 2m4s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-03-03 10:43:41 +03:00

4
kgo.go
View File

@@ -142,8 +142,9 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, *ho
} }
return nil, nil, err return nil, nil, err
} }
k.connected.Store(1)
return c, htracer, nil
} }
return c, htracer, nil
} }
func (k *Broker) Connect(ctx context.Context) error { func (k *Broker) Connect(ctx context.Context) error {
@@ -404,6 +405,7 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
consumers: make(map[tp]*consumer), consumers: make(map[tp]*consumer),
done: make(chan struct{}), done: make(chan struct{}),
fatalOnError: fatalOnError, fatalOnError: fatalOnError,
connected: k.connected,
} }
kopts := append(k.kopts, kopts := append(k.kopts,