pipe

Documentation

Index

• type Pipe
  • func New(c *cmq.CMQ, s *streaming.Streaming, logger *zap.Logger, tracer trace.Tracer) *Pipe
  • func (p *Pipe) Pipe(topic string)
• type Piped
  • func NewPiped(name string) Piped

Types

type Pipe

type Pipe struct {
	CMQ       *cmq.CMQ
	Streaming *streaming.Streaming
	Tracer    trace.Tracer
	Logger    *zap.Logger
}

func New

func New(c *cmq.CMQ, s *streaming.Streaming, logger *zap.Logger, tracer trace.Tracer) *Pipe

New create pipe to pipe data from streaming to jetstream. you can use a single pipe instance for mutiple topics.

func (*Pipe) Pipe

func (p *Pipe) Pipe(topic string)

Pipe start piping messages from streaming to jetstream based on given topic. its subscription on streaming isn't durable and it always start from 1 second behind. the reason here is to reduce load on the streaming server as much as possible.

type Piped

type Piped struct {
	PipedMessages  prometheus.Counter
	FailedMessages prometheus.Counter
	TimeLag        prometheus.Histogram
}

Piped contains metrics to meter the number of piped messages.

func NewPiped

func NewPiped(name string) Piped

NewPiped creates piped metrics based on topic name with const labels. nolint: funlen