Compare commits
4 Commits
Author | SHA1 | Date | |
---|---|---|---|
f4f8793686 | |||
ec4922ad8b | |||
eaea14e5a8 | |||
f6f7139d2f |
4
kgo.go
4
kgo.go
@@ -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,
|
||||||
|
Reference in New Issue
Block a user