Documentation
¶
Overview ¶
Package kafkalog defines Send method to stream your msgs to kafka. Any msg (must end with \n) sent to Stdout with fmt package commands will be redirected to kafka. Implement with the following code: old := os.Stdout r, w, err := os.Pipe()
if err != nil {
panic("Error running os.Pipe()")
}
os.Stdout = w lw := kafkalog.NewAsyncProducer("loggerID")
defer func() {
w.Close()
os.Stdout = old
lw.Close()
}()
go func() {
sc := bufio.NewScanner(r)
for sc.Scan() {
l := sc.Text()
lw.Send(l)
}
}()
Must have "KAFKA_BROKERS" env var set formated as "127.0.0.1:9092,127.0.0.1:9092".
Index ¶
Constants ¶
This section is empty.
Variables ¶
View Source
var ( // AP implements sarama.AsyncProducer. AP = AsyncProducer{} )
View Source
var GetSaramaProducer = func(p *AsyncProducer) error { var err error p.prodr, err = sarama.NewAsyncProducer(p.brokers, p.conf) if err != nil { return err } return nil }
GetSaramaProducer factores out sarama.NewAsyncProducer() for easy testing.
Functions ¶
This section is empty.
Types ¶
type AsyncProducer ¶
type AsyncProducer struct {
// contains filtered or unexported fields
}
AsyncProducer implements sarama.AsyncProducer
func (*AsyncProducer) Send ¶
func (p *AsyncProducer) Send(msg string)
Send sends kafka message to a topic.
type IAsyncProducer ¶
type IAsyncProducer interface {
Send(msg string)
Close()
}
IAsyncProducer exposes writing msg to kafka stream.
func NewAsyncProducer ¶
func NewAsyncProducer(loggerID string) IAsyncProducer
NewAsyncProducer return AsyncProducer as IAsyncProducer.
Source Files
Click to show internal directories.
Click to hide internal directories.