datastreams

Mar 1, 2024

Changes in this version

type Processor

+ func (p *Processor) TrackKafkaHighWatermarkOffset(_ string, topic string, partition int32, offset int64)

Oct 18, 2023

Changes in this version

type StatsPayload

+ Version string

Sep 15, 2023

Changes in this version

+ const PropagationKeyBase64

+ func ContextWithPathway(ctx context.Context, p Pathway) context.Context

+ type Backlog struct

+ Tags []string

+ Value int64

+ func (z *Backlog) DecodeMsg(dc *msgp.Reader) (err error)

+ func (z *Backlog) EncodeMsg(en *msgp.Writer) (err error)

+ func (z *Backlog) Msgsize() (s int)

+ type CommitOffset struct

+ ConsumerGroup string

+ Offset int64

+ Partition int32

+ Topic string

+ func (z *CommitOffset) DecodeMsg(dc *msgp.Reader) (err error)

+ func (z *CommitOffset) EncodeMsg(en *msgp.Writer) (err error)

+ func (z *CommitOffset) Msgsize() (s int)

+ type Pathway struct

+ func Decode(ctx context.Context, data []byte) (p Pathway, outCtx context.Context, err error)

+ func DecodeBase64(ctx context.Context, str string) (p Pathway, outCtx context.Context, err error)

+ func Merge(pathways []Pathway) Pathway

+ func PathwayFromContext(ctx context.Context) (p Pathway, ok bool)

+ func (p Pathway) EdgeStart() time.Time

+ func (p Pathway) Encode() []byte

+ func (p Pathway) EncodeBase64() string

+ func (p Pathway) GetHash() uint64

+ func (p Pathway) PathwayStart() time.Time

+ type Processor struct

+ func NewProcessor(statsd internal.StatsdClient, env, service string, agentURL *url.URL, ...) *Processor

+ func (p *Processor) Flush()

+ func (p *Processor) SetCheckpoint(ctx context.Context, edgeTags ...string) context.Context

+ func (p *Processor) SetCheckpointWithParams(ctx context.Context, params options.CheckpointParams, edgeTags ...string) context.Context

+ func (p *Processor) Start()

+ func (p *Processor) Stop()

+ func (p *Processor) TrackKafkaCommitOffset(group string, topic string, partition int32, offset int64)

+ func (p *Processor) TrackKafkaProduceOffset(topic string, partition int32, offset int64)

+ type ProduceOffset struct

+ Offset int64

+ Partition int32

+ Topic string

+ func (z *ProduceOffset) DecodeMsg(dc *msgp.Reader) (err error)

+ func (z ProduceOffset) EncodeMsg(en *msgp.Writer) (err error)

+ func (z ProduceOffset) Msgsize() (s int)

+ type StatsBucket struct

+ Backlogs []Backlog

+ Duration uint64

+ Start uint64

+ Stats []StatsPoint

+ func (z *StatsBucket) DecodeMsg(dc *msgp.Reader) (err error)

+ func (z *StatsBucket) EncodeMsg(en *msgp.Writer) (err error)

+ func (z *StatsBucket) Msgsize() (s int)

+ type StatsPayload struct

+ Env string

+ Lang string

+ Service string

+ Stats []StatsBucket

+ TracerVersion string

+ func (z *StatsPayload) DecodeMsg(dc *msgp.Reader) (err error)

+ func (z *StatsPayload) EncodeMsg(en *msgp.Writer) (err error)

+ func (z *StatsPayload) Msgsize() (s int)

+ type StatsPoint struct

+ EdgeLatency []byte

+ EdgeTags []string

+ Hash uint64

+ ParentHash uint64

+ PathwayLatency []byte

+ PayloadSize []byte

+ Service string

+ TimestampType TimestampType

+ func (z *StatsPoint) DecodeMsg(dc *msgp.Reader) (err error)

+ func (z *StatsPoint) EncodeMsg(en *msgp.Writer) (err error)

+ func (z *StatsPoint) Msgsize() (s int)

+ type TimestampType string

+ const TimestampTypeCurrent

+ const TimestampTypeOrigin

+ func (z *TimestampType) DecodeMsg(dc *msgp.Reader) (err error)

+ func (z TimestampType) EncodeMsg(en *msgp.Writer) (err error)

+ func (z TimestampType) Msgsize() (s int)