代码拉取完成,页面将自动刷新
package channel
import (
"sync"
"sync/atomic"
"github.com/elastic/beats/filebeat/input"
)
// Outlet struct is used to be passed to an object which needs an outlet
//
// The difference between signal and done channel is as following:
// - signal channel can be added through SetSignal and is used to
// interrupt events sent through OnEventSignal-
// - done channel is used to close and stop the outlet
//
// If SetSignal is used, it must be ensure that there is only one event producer.
type Outlet struct {
wg *sync.WaitGroup // Use for counting active events
done <-chan struct{}
signal <-chan struct{}
channel chan *input.Event
isOpen int32 // atomic indicator
}
func NewOutlet(
done <-chan struct{},
c chan *input.Event,
wg *sync.WaitGroup,
) *Outlet {
return &Outlet{
done: done,
channel: c,
wg: wg,
isOpen: 1,
}
}
// SetSignal sets the signal channel for OnEventSignal
// If SetSignal is used, it must be ensure that only one producer exists.
func (o *Outlet) SetSignal(signal <-chan struct{}) {
o.signal = signal
}
func (o *Outlet) OnEvent(event *input.Event) bool {
open := atomic.LoadInt32(&o.isOpen) == 1
if !open {
return false
}
if o.wg != nil {
o.wg.Add(1)
}
select {
case <-o.done:
if o.wg != nil {
o.wg.Done()
}
atomic.StoreInt32(&o.isOpen, 0)
return false
case o.channel <- event:
return true
}
}
// OnEventSignal can be stopped by the signal that is set with SetSignal
// This does not close the outlet. Only OnEvent does close the outlet.
// If OnEventSignal is used, it must be ensured that only one producer is used.
func (o *Outlet) OnEventSignal(event *input.Event) bool {
open := atomic.LoadInt32(&o.isOpen) == 1
if !open {
return false
}
if o.wg != nil {
o.wg.Add(1)
}
select {
case <-o.signal:
if o.wg != nil {
o.wg.Done()
}
o.signal = nil
return false
case o.channel <- event:
return true
}
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。