base

Documentation

Index

• Variables
• type Node
  • func NewEmptyNode(id string) *Node
  • func (n *Node) AddOutput(node *Node)
  • func (n *Node) AddTaskNode(nextNode *Node, taskNode *TaskNode)
  • func (n *Node) RemoveOutput(node *Node)
  • func (n *Node) RemoveTaskNode(nextNode *Node, taskNode *TaskNode)
  • func (n *Node) Run()
  • func (n *Node) SetInput(input chan interface{})
  • func (n *Node) String() string
  • func (n *Node) WaitUntilGameOver()
• type NodeI
• type TaskNode

Variables

var (
	CrawlerSendTotal        = bkmonitoring.NewInt("crawler_send_total")         // 兼容指标，之前有使用这个作为发送的数量
	CrawlerReceived         = bkmonitoring.NewInt("crawler_received")           // 兼容指标，接收到的所有事件数
	CrawlerState            = bkmonitoring.NewInt("crawler_state")              // 兼容指标，接收到的所有事件中状态事件数量
	CrawlerDropped          = bkmonitoring.NewInt("crawler_dropped")            // 兼容指标，丢弃数量
	CrawlerPackageSendTotal = bkmonitoring.NewInt("crawler_package_send_total") // 打包发送的数量
)

Types

type Node

type Node struct {
	ID         string
	ParentNode *Node

	Outs map[string]chan interface{}
	In   chan interface{}

	CloseOnce sync.Once
	End       chan struct{}

	GameOver chan struct{} // 用该信号代表Run函数已经完整退出

	TaskNodeList map[string]map[string]*TaskNode
}

func NewEmptyNode

func NewEmptyNode(id string) *Node

func (*Node) AddOutput

func (n *Node) AddOutput(node *Node)

func (*Node) AddTaskNode

func (n *Node) AddTaskNode(nextNode *Node, taskNode *TaskNode)

func (*Node) RemoveOutput

func (n *Node) RemoveOutput(node *Node)

func (*Node) RemoveTaskNode

func (n *Node) RemoveTaskNode(nextNode *Node, taskNode *TaskNode)

func (*Node) Run

func (n *Node) Run()

func (*Node) SetInput

func (n *Node) SetInput(input chan interface{})

func (*Node) String

func (n *Node) String() string

func (*Node) WaitUntilGameOver

func (n *Node) WaitUntilGameOver()

type NodeI

type NodeI interface {
	SetInput(input chan interface{})
	AddOutput(node *Node)
	RemoveOutput(node *Node)
	Run()
}

type TaskNode

type TaskNode struct {
	*Node

	CrawlerReceived  *monitoring.Int //state事件
	CrawlerState     *monitoring.Int //state事件
	CrawlerSendTotal *monitoring.Int //正常事件总数
	CrawlerDropped   *monitoring.Int //过滤掉的事件总数

	SenderReceive   *monitoring.Int // 接收的事件数
	SenderSendTotal *monitoring.Int // 发送到pipeline的数量
	SenderState     *monitoring.Int // 仅需要更新采集状态的事件数(event.Field为空)
}