@ -2,15 +2,16 @@ package kafka
import (
import (
"context"
"context"
"log"
"time"
"time"
logger "git.slaventius.ru/test3k/umate/pkg/logger"
"github.com/segmentio/kafka-go"
"github.com/segmentio/kafka-go"
)
)
type KafkaWriter struct {
type KafkaWriter struct {
ctx context . Context
ctx context . Context
writer * kafka . Writer
writer * kafka . Writer
logger * logger . Logger
first string
first string
topic string
topic string
}
}
@ -82,7 +83,7 @@ func (s *KafkaWriter) checkTopic() error {
return era
return era
}
}
log . Printf ( "create topic %q\n" , s . topic )
s . logger . Printf ( "create topic %q\n" , s . topic )
return era
return era
}
}
@ -90,7 +91,7 @@ func (s *KafkaWriter) checkTopic() error {
return nil
return nil
}
}
func NewWriter ( ctx context . Context , topic string , address ... string ) * KafkaWriter {
func NewWriter ( ctx context . Context , logger * logger . Logger , topic string , address ... string ) * KafkaWriter {
s := & KafkaWriter {
s := & KafkaWriter {
ctx : ctx ,
ctx : ctx ,
writer : & kafka . Writer {
writer : & kafka . Writer {
@ -100,6 +101,7 @@ func NewWriter(ctx context.Context, topic string, address ...string) *KafkaWrite
BatchTimeout : time . Millisecond * 100 ,
BatchTimeout : time . Millisecond * 100 ,
Addr : kafka . TCP ( address ... ) ,
Addr : kafka . TCP ( address ... ) ,
} ,
} ,
logger : logger ,
first : address [ 0 ] ,
first : address [ 0 ] ,
topic : topic ,
topic : topic ,
}
}
@ -107,7 +109,7 @@ func NewWriter(ctx context.Context, topic string, address ...string) *KafkaWrite
// Проверим и при необходимости создадим топик
// Проверим и при необходимости создадим топик
era := s . checkTopic ( )
era := s . checkTopic ( )
if era != nil {
if era != nil {
log . Fatal ( era )
logger . Fatal ( era )
}
}
return s
return s