代码拉取完成,页面将自动刷新
package publisher
import (
"sync"
"github.com/elastic/beats/filebeat/input"
"github.com/elastic/beats/libbeat/logp"
"github.com/elastic/beats/libbeat/publisher"
)
type syncLogPublisher struct {
pub publisher.Publisher
client publisher.Client
in chan []*input.Event
out SuccessLogger
done chan struct{}
wg sync.WaitGroup
}
func newSyncLogPublisher(
in chan []*input.Event,
out SuccessLogger,
pub publisher.Publisher,
) *syncLogPublisher {
return &syncLogPublisher{
in: in,
out: out,
pub: pub,
done: make(chan struct{}),
}
}
func (p *syncLogPublisher) Start() {
p.client = p.pub.Connect()
p.wg.Add(1)
go func() {
defer p.wg.Done()
logp.Info("Start sending events to output")
defer logp.Debug("publisher", "Shutting down sync publisher")
for {
err := p.Publish()
if err != nil {
return
}
}
}()
}
func (p *syncLogPublisher) Publish() error {
var events []*input.Event
select {
case <-p.done:
return sigPublisherStop
case events = <-p.in:
}
dataEvents, meta := getDataEvents(events)
ok := p.client.PublishEvents(dataEvents, publisher.Sync, publisher.Guaranteed,
publisher.MetadataBatch(meta))
if !ok {
// PublishEvents will only returns false, if p.client has been closed.
return sigPublisherStop
}
// TODO: move counter into logger?
logp.Debug("publish", "Events sent: %d", len(events))
eventsSent.Add(int64(len(events)))
// Tell the logger that we've successfully sent these events
ok = p.out.Published(events)
if !ok {
// stop publisher if successfully send events can not be logged anymore.
return sigPublisherStop
}
return nil
}
func (p *syncLogPublisher) Stop() {
p.client.Close()
close(p.done)
p.wg.Wait()
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。