analysis

Documentation

Overview

Package analysis implements the UI-independent topic scan + aggregation engine (TP-29/TP-30). The Aggregator is a pure, allocation-bounded accumulator that a Runner feeds sampled messages; the Registry owns the background lifecycle.

Index

• type Aggregator
  • func NewAggregator(topic string) *Aggregator
  • func (a *Aggregator) Add(msg api.Message)
  • func (a *Aggregator) Count() int64
  • func (a *Aggregator) Result() api.TopicAnalysisResult
• type ConsumeFunc
• type Registry
  • func NewRegistry(consume ConsumeFunc) *Registry
  • func (reg *Registry) Cancel(topic string) error
  • func (reg *Registry) Get(topic string) (*api.TopicAnalysis, error)
  • func (reg *Registry) Start(ctx context.Context, topic string, totalOffsets int64) error

Types

type Aggregator

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

Aggregator accumulates message statistics for a topic scan. It is not safe for concurrent use; the Runner owns a single instance per scan.

func NewAggregator

func NewAggregator(topic string) *Aggregator

NewAggregator returns a ready Aggregator for the named topic.

func (*Aggregator) Add

func (a *Aggregator) Add(msg api.Message)

Add folds one message into the aggregation.

func (*Aggregator) Count

func (a *Aggregator) Count() int64

Count returns the number of messages folded so far.

func (*Aggregator) Result

func (a *Aggregator) Result() api.TopicAnalysisResult

Result materialises the aggregation into an api.TopicAnalysisResult.

type ConsumeFunc

type ConsumeFunc func(ctx context.Context, topic string, flags api.ConsumeFlags, handle api.MessageHandlerFunc, onError func(err any)) error

ConsumeFunc matches api.KafkaDataSource.ConsumeTopic and is the only coupling between the engine and the datasource.

type Registry

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

Registry owns per-topic analysis runs (single result retained per topic).

func NewRegistry

func NewRegistry(consume ConsumeFunc) *Registry

NewRegistry builds a Registry driven by the given ConsumeFunc.

func (*Registry) Cancel

func (reg *Registry) Cancel(topic string) error

Cancel stops a running analysis and discards its run (records nothing).

func (*Registry) Get

func (reg *Registry) Get(topic string) (*api.TopicAnalysis, error)

Get returns a snapshot of the latest analysis for topic, or (nil, nil) when none has ever been started.

func (*Registry) Start

func (reg *Registry) Start(ctx context.Context, topic string, totalOffsets int64) error

Start begins a background analysis of topic. totalOffsets is the message total used for the progress percentage (0 when unknown). It returns an AnalysisAlreadyRunningError when one is already in progress.