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

disable connected state changes now

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2025-03-12 20:40:30 +03:00
parent 21f06fee6c
commit da7b4c4a9a
1 changed files with 6 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
broker.go
View File

@@ -27,7 +27,7 @@ var (
func (m *hookEvent) OnGroupManageError(err error) { func (m *hookEvent) OnGroupManageError(err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
if m.fatalOnError { if m.fatalOnError {
m.log.Fatal(context.TODO(), "kgo.OnGroupManageError", err) m.log.Fatal(context.TODO(), "kgo.OnGroupManageError", err)
} }
@@ -36,7 +36,7 @@ func (m *hookEvent) OnGroupManageError(err error) {
func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) { func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net.Conn, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
if m.fatalOnError { if m.fatalOnError {
m.log.Fatal(context.TODO(), "kgo.OnBrokerConnect", err) m.log.Fatal(context.TODO(), "kgo.OnBrokerConnect", err)
} }
@@ -44,12 +44,12 @@ func (m *hookEvent) OnBrokerConnect(_ kgo.BrokerMetadata, _ time.Duration, _ net
} }
func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) { func (m *hookEvent) OnBrokerDisconnect(_ kgo.BrokerMetadata, _ net.Conn) {
m.connected.Store(0) // m.connected.Store(0)
} }
func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) { func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
if m.fatalOnError { if m.fatalOnError {
m.log.Fatal(context.TODO(), "kgo.OnBrokerWrite", err) m.log.Fatal(context.TODO(), "kgo.OnBrokerWrite", err)
} }
@@ -58,12 +58,12 @@ func (m *hookEvent) OnBrokerWrite(_ kgo.BrokerMetadata, _ int16, _ int, _ time.D
func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) { func (m *hookEvent) OnBrokerRead(_ kgo.BrokerMetadata, _ int16, _ int, _ time.Duration, _ time.Duration, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
} }
} }
func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) { func (m *hookEvent) OnProduceRecordUnbuffered(_ *kgo.Record, err error) {
if err != nil { if err != nil {
m.connected.Store(0) // m.connected.Store(0)
} }
} }