in

Documentation

Index

• func AcceptAndListenTCPBuf(l net.Listener, queue chan<- [][]byte, term <-chan struct{}, ...)
• func ListenGRPC(l net.Listener, queue chan []byte, stop <-chan struct{}, ...)
• func ListenUDP(conn net.PacketConn, queue chan []byte, stop <-chan struct{}, ...)
• func ListenUDPBuf(conn net.PacketConn, queue chan [][]byte, stop <-chan struct{}, ...)
• type BatchChan
  • func NewBatchChan(q chan<- [][]byte, bufSize int, periodSec int, ms *metrics.Prom) *BatchChan
  • func (qb *BatchChan) Close()
  • func (qb *BatchChan) Flush()
  • func (qb *BatchChan) Push(rec []byte)

Functions

func AcceptAndListenTCPBuf

func AcceptAndListenTCPBuf(l net.Listener, queue chan<- [][]byte, term <-chan struct{},
	rateLimiters []*ratelimiter.SlidingWindow, cfg *conf.Main, connWG *sync.WaitGroup, ms *metrics.Prom, lg *zap.Logger)

AcceptAndListenTCPBuf is batched version of AcceptAndListenTCP.

func ListenGRPC

func ListenGRPC(l net.Listener, queue chan []byte, stop <-chan struct{}, connWG *sync.WaitGroup, cfg *conf.Main, ms *metrics.Prom, lg *zap.Logger)

ListenGRPC listens for incoming gRPC connections. Blocking. Returns when server returns.

func ListenUDP

func ListenUDP(conn net.PacketConn, queue chan []byte, stop <-chan struct{}, connWG *sync.WaitGroup, ms *metrics.Prom, lg *zap.Logger)

ListenUDP listens for incoming UDP connections.

func ListenUDPBuf

func ListenUDPBuf(conn net.PacketConn, queue chan [][]byte, stop <-chan struct{}, rateLimiters []*ratelimiter.SlidingWindow,
	connWG *sync.WaitGroup, cfg *conf.Main, ms *metrics.Prom, lg *zap.Logger)

ListenUDPBuf is a buffered version of ListenUDP.

Types

type BatchChan

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

BatchChan represents a buffer to send records into a chan in batches. BatchChan accumulates records in a buffer. When buffer is full, it sends it as a batch to a chan. It also periodically flushes the buffer to prevent losing data when it's not enough to fill the buffer.

func NewBatchChan

func NewBatchChan(q chan<- [][]byte, bufSize int, periodSec int, ms *metrics.Prom) *BatchChan

NewBatchChan makes a new batched chan buffer. It also starts a flushing goroutine in the background if periodSec > 0.

func (*BatchChan) Close

func (qb *BatchChan) Close()

Close closes the batch channel. It does not close the target channel. Must be called exactly once for every new instance.

func (*BatchChan) Flush

func (qb *BatchChan) Flush()

Flush immediately sends buffered items to the target chan.

func (*BatchChan) Push

func (qb *BatchChan) Push(rec []byte)

Push pushes a single item to the batched channel.