overflow_cloud/external-go
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-05-11 18:56:33 +09:00
parent 1742b7fab6
commit 4c55033eae
1 changed files with 8 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
kafka/kafka.go
View File

@ -19,7 +19,7 @@ func InitPackage(config *occe.Kafka) {
return return
} }
if nil != config.Producers || 0 < len(config.Producers) { if nil != config.Producers && 0 < len(config.Producers) {
kafkaWriters = make(map[string]*kafka.Writer) kafkaWriters = make(map[string]*kafka.Writer)
for n, c := range config.Producers { for n, c := range config.Producers {
wc := kafka.WriterConfig{ wc := kafka.WriterConfig{
@ -40,7 +40,7 @@ func InitPackage(config *occe.Kafka) {
} }
} }
if nil != config.Consumers || 0 < len(config.Consumers) { if nil != config.Consumers && 0 < len(config.Consumers) {
kafkaReaders = make(map[string]*kafka.Reader) kafkaReaders = make(map[string]*kafka.Reader)
for n, c := range config.Consumers { for n, c := range config.Consumers {
rc := kafka.ReaderConfig{ rc := kafka.ReaderConfig{
@ -87,6 +87,12 @@ func DestroyPackage(config *occe.Kafka) {
logging.Logger().Errorf("%v", err) logging.Logger().Errorf("%v", err)
} }
} }
for _, r := range kafkaReaders {
if err := r.Close(); nil != err {
logging.Logger().Errorf("%v", err)
}
}
} }
func Writer(name string) *kafka.Writer { func Writer(name string) *kafka.Writer {