sender

Documentation

Index

• Constants
• func ConvertDatas(ins []map[string]interface{}) []Data
• func ConvertDatasBack(ins []Data) []map[string]interface{}
• func RegisterConstructor(typ string, c Constructor)
• func SplitData(data string) (valArray []string)
• func SplitDataWithSplitSize(data string, splitSize int64) (valArray []string)
• type Constructor
• type FtOption
• type FtSender
  • func NewFtSender(innerSender Sender, conf conf.MapConf, ftSaveLogPath string) (*FtSender, error)
  • func (ft *FtSender) Close() error
  • func (ft *FtSender) Name() string
  • func (ft *FtSender) RawSend(datas []string) error
  • func (ft *FtSender) Reset() error
  • func (ft *FtSender) Restore(info *StatsInfo)
  • func (ft *FtSender) Send(datas []Data) error
  • func (ft *FtSender) SkipDeepCopy() bool
  • func (ft *FtSender) Stats() StatsInfo
  • func (ft *FtSender) TokenRefresh(mapConf conf.MapConf) (err error)
• type RawSender
• type Registry
  • func NewRegistry() *Registry
  • func (r *Registry) NewSender(conf conf.MapConf, ftSaveLogPath string) (sender Sender, err error)
  • func (r *Registry) RegisterSender(senderType string, constructor func(conf.MapConf) (Sender, error)) error
• type Sender
• type SkipDeepCopySender
• type StatsSender
• type TokenRefreshable

Constants

const (
	DefaultSplitSize = 64 * 1024 // 默认分割为 64 kb
	// TypeMarshalError 表示marshal出错
	TypeMarshalError = reqerr.SendErrorType("Data Marshal failed")
)

Functions

func ConvertDatas

func ConvertDatas(ins []map[string]interface{}) []Data

func ConvertDatasBack

func ConvertDatasBack(ins []Data) []map[string]interface{}

func RegisterConstructor

func RegisterConstructor(typ string, c Constructor)

RegisterConstructor adds a new constructor for a given type of reader.

func SplitData

func SplitData(data string) (valArray []string)

优先使用'\n'对数据进行切分，切分后单个分片仍大于batchsize再按指定大小进行切分

func SplitDataWithSplitSize

func SplitDataWithSplitSize(data string, splitSize int64) (valArray []string)

Types

type Constructor

type Constructor func(conf.MapConf) (Sender, error)

type FtOption

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

type FtSender

type FtSender struct {
	BackupQueue queue.BackendQueue
	// contains filtered or unexported fields
}

FtSender fault tolerance sender wrapper

func NewFtSender

func NewFtSender(innerSender Sender, conf conf.MapConf, ftSaveLogPath string) (*FtSender, error)

NewFtSender Fault tolerant sender constructor

func (*FtSender) Close

func (ft *FtSender) Close() error

func (*FtSender) Name

func (ft *FtSender) Name() string

func (*FtSender) RawSend

func (ft *FtSender) RawSend(datas []string) error

func (*FtSender) Reset

func (ft *FtSender) Reset() error

func (*FtSender) Restore

func (ft *FtSender) Restore(info *StatsInfo)

func (*FtSender) Send

func (ft *FtSender) Send(datas []Data) error

func (*FtSender) SkipDeepCopy

func (ft *FtSender) SkipDeepCopy() bool

func (*FtSender) Stats

func (ft *FtSender) Stats() StatsInfo

func (*FtSender) TokenRefresh

func (ft *FtSender) TokenRefresh(mapConf conf.MapConf) (err error)

type RawSender

type RawSender interface {
	RawSend([]string) error
}

RawSender used for optimize runner, without parse data

type Registry

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

SenderRegistry sender 的工厂类。可以注册自定义sender

func NewRegistry

func NewRegistry() *Registry

func (*Registry) NewSender

func (r *Registry) NewSender(conf conf.MapConf, ftSaveLogPath string) (sender Sender, err error)

func (*Registry) RegisterSender

func (r *Registry) RegisterSender(senderType string, constructor func(conf.MapConf) (Sender, error)) error

type Sender

type Sender interface {
	Name() string
	// send data, error if failed
	Send([]Data) error
	Close() error
}

Sender send data to pandora, prometheus such different destinations

type SkipDeepCopySender

type SkipDeepCopySender interface {
	// SkipDeepCopy 需要返回值是因为如果一个 sender 封装了其它 sender，需要根据实际封装的类型返回是否忽略深度拷贝
	SkipDeepCopy() bool
}

SkipDeepCopySender 表示该 sender 不会对传入数据进行污染，凡是有次保证的 sender 需要实现该接口提升发送效率

type StatsSender

type StatsSender interface {
	Name() string
	// send data, error if failed
	Send([]Data) error
	Close() error
	Stats() StatsInfo
	// 恢复 sender 停止之前的状态
	Restore(*StatsInfo)
}

type TokenRefreshable

type TokenRefreshable interface {
	TokenRefresh(conf.MapConf) error
}