producer

Documentation

Index

• Constants
• func GenerateLog(logTime int64, addLogMap map[string]string) *pb.Log
• func GetLogListSize(logList []*pb.Log) int
• func GetLogSize(log *pb.Log) int
• func GetTimeMs(t int64) int64
• func WithRecover(fn func())
• type Attempt
• type Batch
• type BatchKey
• type BatchLog
• type CallBack
• type Config
  • func GetDefaultProducerConfig() *Config
• type Dispatcher
  • func (dispatcher *Dispatcher) IsShutDown() bool
  • func (dispatcher *Dispatcher) RetryQueueElegantQuit()
• type Producer
  • func NewProducer(config *Config) Producer
• type Result
• type RetryQueue
  • func (q *RetryQueue) Len() int
  • func (q *RetryQueue) Less(i, j int) bool
  • func (q *RetryQueue) Pop() interface{}
  • func (q *RetryQueue) Push(x interface{})
  • func (q *RetryQueue) Swap(i, j int)
• type Sender
  • func (sender *Sender) FailedCallback(batch *Batch)
  • func (sender *Sender) IsShutDown() bool
• type ThreadPool

Constants

const (
	TimeoutException = "TimeoutException"
)

Functions

func GenerateLog

func GenerateLog(logTime int64, addLogMap map[string]string) *pb.Log

func GetLogListSize

func GetLogListSize(logList []*pb.Log) int

func GetLogSize

func GetLogSize(log *pb.Log) int

func GetTimeMs

func GetTimeMs(t int64) int64

func WithRecover

func WithRecover(fn func())

Types

type Attempt

type Attempt struct {
	SuccessFlag  bool
	RequestId    string
	ErrorCode    string
	ErrorMessage string
	TimestampMs  int64
}

type Batch

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

type BatchKey

type BatchKey struct {
	Topic       string
	Source      string
	ShardHash   string
	FileName    string
	CallBackFun CallBack
}

type BatchLog

type BatchLog struct {
	Key BatchKey
	Log *pb.Log
}

type CallBack

type CallBack interface {
	Success(result *Result)
	Fail(result *Result)
}

type Config

type Config struct {
	TotalSizeLnBytes      int64
	MaxSenderCount        int64
	MaxBlockSec           int
	MaxBatchSize          int64
	MaxBatchCount         int
	LingerTime            time.Duration
	Retries               int
	MaxReservedAttempts   int
	BaseRetryBackoffMs    int64
	MaxRetryBackoffMs     int64
	AdjustShardHashFlag   bool
	ShardCount            int
	NoRetryStatusCodeList []int

	common.LoggerConfig
	common.ClientConfig
}

func GetDefaultProducerConfig

func GetDefaultProducerConfig() *Config

type Dispatcher

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

func (*Dispatcher) IsShutDown

func (dispatcher *Dispatcher) IsShutDown() bool

func (*Dispatcher) RetryQueueElegantQuit

func (dispatcher *Dispatcher) RetryQueueElegantQuit()

type Producer

type Producer interface {
	SendLog(shardHash, topic, source, filename string, log *pb.Log, callBack CallBack) error
	SendLogs(shardHash, topic, source, filename string, logs *pb.LogGroup, callBack CallBack) error
	ResetAccessKeyToken(accessKeyID, accessKeySecret, securityToken string)
	Start()
	Close()
	ForceClose()
}

func NewProducer

func NewProducer(config *Config) Producer

type Result

type Result struct {
	Attempts    []*Attempt
	SuccessFlag bool
}

type RetryQueue

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

func (*RetryQueue) Len

func (q *RetryQueue) Len() int

func (*RetryQueue) Less

func (q *RetryQueue) Less(i, j int) bool

func (*RetryQueue) Pop

func (q *RetryQueue) Pop() interface{}

func (*RetryQueue) Push

func (q *RetryQueue) Push(x interface{})

func (*RetryQueue) Swap

func (q *RetryQueue) Swap(i, j int)

type Sender

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

func (*Sender) FailedCallback

func (sender *Sender) FailedCallback(batch *Batch)

func (*Sender) IsShutDown

func (sender *Sender) IsShutDown() bool

type ThreadPool

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