kafka

Documentation

Index

• type KafkaDataType
• type KafkaOutput
  • func (out *KafkaOutput) Close()
  • func (out *KafkaOutput) Connect() error
  • func (out *KafkaOutput) GetNameByIP(ip string) string
  • func (out *KafkaOutput) Init(config outputs.MothershipConfig, topology_expire int) error
  • func (out *KafkaOutput) PublishEvent(ts time.Time, event common.MapStr) error
  • func (out *KafkaOutput) PublishIPs(name string, localAddrs []string) error
  • func (out *KafkaOutput) Reconnect()
  • func (out *KafkaOutput) SendMessagesGoroutine()
• type KafkaQueueMsg
  • func (qmsg *KafkaQueueMsg) Encode() ([]byte, error)
  • func (qmsg *KafkaQueueMsg) Length() int

Types

type KafkaDataType

type KafkaDataType uint16

const (
	KafkaListType KafkaDataType = iota
	KafkaChannelType
)

type KafkaOutput

type KafkaOutput struct {
	ReconnectInterval time.Duration
	BrokerList        []string
	Topic             string
	Timeout           time.Duration
	FlushInterval     time.Duration
	Producer          sarama.AsyncProducer
	// contains filtered or unexported fields
}

func (*KafkaOutput) Close

func (out *KafkaOutput) Close()

func (*KafkaOutput) Connect

func (out *KafkaOutput) Connect() error

func (*KafkaOutput) GetNameByIP

func (out *KafkaOutput) GetNameByIP(ip string) string

func (*KafkaOutput) Init

func (out *KafkaOutput) Init(config outputs.MothershipConfig, topology_expire int) error

func (*KafkaOutput) PublishEvent

func (out *KafkaOutput) PublishEvent(ts time.Time, event common.MapStr) error

func (*KafkaOutput) PublishIPs

func (out *KafkaOutput) PublishIPs(name string, localAddrs []string) error

func (*KafkaOutput) Reconnect

func (out *KafkaOutput) Reconnect()

func (*KafkaOutput) SendMessagesGoroutine

func (out *KafkaOutput) SendMessagesGoroutine()

type KafkaQueueMsg

type KafkaQueueMsg struct {
	// contains filtered or unexported fields
}

func (*KafkaQueueMsg) Encode

func (qmsg *KafkaQueueMsg) Encode() ([]byte, error)

func (*KafkaQueueMsg) Length

func (qmsg *KafkaQueueMsg) Length() int