sender

Documentation

Index

• type MessageBuffer
  • func NewMessageBuffer(batchSizeLimit int, contentSizeLimit int) *MessageBuffer
  • func (p *MessageBuffer) AddMessage(message *message.Message) bool
  • func (p *MessageBuffer) Clear()
  • func (p *MessageBuffer) GetMessages() []*message.Message
  • func (p *MessageBuffer) IsEmpty() bool
  • func (p *MessageBuffer) IsFull() bool
• type Sender
  • func NewSender(inputChan chan *message.Message, outputChan chan *message.Message, ...) *Sender
  • func (s *Sender) Start()
  • func (s *Sender) Stop()
• type Serializer
• type Strategy
  • func NewBatchStrategy(serializer Serializer, batchWait time.Duration) Strategy

Types

type MessageBuffer

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

MessageBuffer accumulates messages to a buffer until the max capacity is reached.

func NewMessageBuffer

func NewMessageBuffer(batchSizeLimit int, contentSizeLimit int) *MessageBuffer

NewMessageBuffer returns a new MessageBuffer.

func (*MessageBuffer) AddMessage

func (p *MessageBuffer) AddMessage(message *message.Message) bool

AddMessage adds a message to the buffer if there is still some free space, returns true if the message was added.

func (*MessageBuffer) Clear

func (p *MessageBuffer) Clear()

Clear reinitializes the buffer.

func (*MessageBuffer) GetMessages

func (p *MessageBuffer) GetMessages() []*message.Message

GetMessages returns the messages stored in the buffer.

func (*MessageBuffer) IsEmpty

func (p *MessageBuffer) IsEmpty() bool

IsEmpty returns true if the buffer is empty.

func (*MessageBuffer) IsFull

func (p *MessageBuffer) IsFull() bool

IsFull returns true if the buffer is full.

type Sender

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

Sender sends logs to different destinations.

func NewSender

func NewSender(inputChan chan *message.Message, outputChan chan *message.Message, destinations *client.Destinations, strategy Strategy) *Sender

NewSender returns a new sender.

func (*Sender) Start

func (s *Sender) Start()

Start starts the sender.

func (*Sender) Stop

func (s *Sender) Stop()

Stop stops the sender, this call blocks until inputChan is flushed

type Serializer

type Serializer interface {
	Serialize(messages []*message.Message) []byte
}

Serializer transforms a batch of messages into a payload.

var (
	// LineSerializer is a shared line serializer.
	LineSerializer Serializer = &lineSerializer{}
	// ArraySerializer is a shared line serializer.
	ArraySerializer Serializer = &arraySerializer{}
)

type Strategy

type Strategy interface {
	Send(inputChan chan *message.Message, outputChan chan *message.Message, send func([]byte) error)
}

Strategy should contain all logic to send logs to a remote destination and forward them the next stage of the pipeline.

var StreamStrategy Strategy = &streamStrategy{}

StreamStrategy is a shared stream strategy.

func NewBatchStrategy

func NewBatchStrategy(serializer Serializer, batchWait time.Duration) Strategy

NewBatchStrategy returns a new batchStrategy.