topo

package
v2.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 19, 2026 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	InitState  int32 = 0
	OpenState  int32 = 1
	CloseState int32 = -1
)

Variables

This section is empty.

Functions

func CloseSubTopo added in v2.4.0 

func CloseSubTopo(ctx api.StreamContext, s *SrcSubTopo)

CloseSubTopo decrements the reference count for ctx's rule on s and, when the last reference is removed, atomically cancels the subtopo and evicts it from the pool. Both locks (pool first, then subtopo) are held for the entire operation, so the destroy decision and the pool delete are one atomic step — no zombie window is possible. This mirrors GetOrCreateSubTopo: the pool file is the sole owner of all mutations to lock and subTopoPool.

func GetSubTopoPoolSize added in v2.4.0 

func GetSubTopoPoolSize() int

GetSubTopoPoolSize returns the number of entries in the subtopo pool. It is intended for use in tests to verify cleanup after planning errors.

func RemoveSubTopo 

func RemoveSubTopo(name string)

Types

type SrcSubTopo 

type SrcSubTopo struct {

	// runtime state
	// Ref state, affect the pool. Update when rule created or stopped
	syncx.RWMutex
	// contains filtered or unexported fields
}

SrcSubTopo Implements node.SourceNode

func GetOrCreateSubTopo added in v2.0.1 

func GetOrCreateSubTopo(ctx api.StreamContext, name string, isSliceMode bool, init func(*SrcSubTopo) error) (*SrcSubTopo, error)

func (*SrcSubTopo) AddOperator 

func (s *SrcSubTopo) AddOperator(inputs []node.Emitter, operator node.OperatorNode) *SrcSubTopo

AddOperator adds an internal operator to the subtopo.

func (*SrcSubTopo) AddOutput 

func (s *SrcSubTopo) AddOutput(output chan interface{}, name string) error

func (*SrcSubTopo) AddSrc 

func (s *SrcSubTopo) AddSrc(src node.DataSourceNode) *SrcSubTopo

func (*SrcSubTopo) Close 

func (s *SrcSubTopo) Close(ctx api.StreamContext)

Close satisfies the node.DataSourceNode interface. All pool-lifecycle logic (reference counting, pool eviction, chained close) lives in CloseSubTopo so that subtopo_pool.go is the single owner of the pool lock and map.

func (*SrcSubTopo) EnableCheckpoint 

func (s *SrcSubTopo) EnableCheckpoint(sources *[]checkpoint.StreamTask, ops *[]checkpoint.NonSourceTask)

func (*SrcSubTopo) GetMetrics 

func (s *SrcSubTopo) GetMetrics() []any

func (*SrcSubTopo) GetName 

func (s *SrcSubTopo) GetName() string

func (*SrcSubTopo) GetSource 

func (s *SrcSubTopo) GetSource() node.DataSourceNode

func (*SrcSubTopo) Init added in v2.3.0 

func (s *SrcSubTopo) Init(ctx api.StreamContext)

func (*SrcSubTopo) IsSliceMode added in v2.3.0 

func (s *SrcSubTopo) IsSliceMode() bool

IsSliceMode this is a constant set when creating new subtopo

func (*SrcSubTopo) LinkTopo 

func (s *SrcSubTopo) LinkTopo(parentTopo *def.PrintableTopo, parentJointName string)

func (*SrcSubTopo) MergeSrc 

func (s *SrcSubTopo) MergeSrc(parentTopo *def.PrintableTopo)

func (*SrcSubTopo) Open 

func (s *SrcSubTopo) Open(ctx api.StreamContext, parentErrCh chan<- error)

Open is different from main topo because this will run multiple times. Each new ref rule will run open subtopo

func (*SrcSubTopo) OpsCount 

func (s *SrcSubTopo) OpsCount() int

func (*SrcSubTopo) RefCount added in v2.4.0 

func (s *SrcSubTopo) RefCount() int

RefCount returns the number of rules currently referencing this sub-topology. It is intended for use in tests to verify cleanup after planning errors.

func (*SrcSubTopo) RemoveMetrics 

func (s *SrcSubTopo) RemoveMetrics(ruleId string)

RemoveMetrics is called when the rule is deleted

func (*SrcSubTopo) RemoveOutput 

func (s *SrcSubTopo) RemoveOutput(name string) error

func (*SrcSubTopo) StoreSchema 

func (s *SrcSubTopo) StoreSchema(ruleID, dataSource string, schema map[string]*ast.JsonStreamField, isWildCard bool)

func (*SrcSubTopo) SubMetrics 

func (s *SrcSubTopo) SubMetrics() (keys []string, values []any)

type State added in v2.3.0 

type State int

State represents the lifecycle state 

const (
	StateInitialized State = iota
	StateOpened
	StateClosed
)

type Topo 

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

Topo is the runtime DAG for a rule It only runs once. If the rule restarts, another topo is created.

func NewWithNameAndOptions 

func NewWithNameAndOptions(name string, options *def.RuleOption) (*Topo, error)

func (*Topo) AddOperator 

func (s *Topo) AddOperator(inputs []node.Emitter, operator node.OperatorNode) *Topo

func (*Topo) AddSink 

func (s *Topo) AddSink(inputs []node.Emitter, snk node.DataSinkNode) *Topo

func (*Topo) AddSinkAlterOperator 

func (s *Topo) AddSinkAlterOperator(sink *node.SinkNode, operator node.OperatorNode) *Topo

func (*Topo) AddSrc 

func (s *Topo) AddSrc(src node.DataSourceNode) *Topo

func (*Topo) Cancel 

func (s *Topo) Cancel()

Cancel may be called multiple times so must be idempotent

func (*Topo) EnableTracer 

func (s *Topo) EnableTracer(isEnabled bool, strategy kctx.TraceStrategy)

func (*Topo) GetContext 

func (s *Topo) GetContext() api.StreamContext

func (*Topo) GetCoordinator 

func (s *Topo) GetCoordinator() *checkpoint.Coordinator

func (*Topo) GetMetrics 

func (s *Topo) GetMetrics() (keys []string, values []any)

func (*Topo) GetMetricsV2 

func (s *Topo) GetMetricsV2() map[string]map[string]any

func (*Topo) GetName 

func (s *Topo) GetName() string

func (*Topo) GetRunId added in v2.3.0 

func (s *Topo) GetRunId() int

func (*Topo) GetSinkSchema added in v2.3.0 

func (s *Topo) GetSinkSchema() map[string]*ast.JsonStreamField

func (*Topo) GetSourceNodes added in v2.0.7 

func (s *Topo) GetSourceNodes() []node.DataSourceNode

GetSourceNodes only for test

func (*Topo) GetStreams 

func (s *Topo) GetStreams() []string

func (*Topo) GetTopo 

func (s *Topo) GetTopo() *def.PrintableTopo

func (*Topo) GracefulStop added in v2.3.0 

func (s *Topo) GracefulStop(timeout time.Duration) error

func (*Topo) IsClosed added in v2.3.0 

func (s *Topo) IsClosed() bool

func (*Topo) IsTraceEnabled 

func (s *Topo) IsTraceEnabled() bool

func (*Topo) Open 

func (s *Topo) Open() <-chan error

func (*Topo) RemoveMetrics 

func (s *Topo) RemoveMetrics()

func (*Topo) ResetStreamOffset 

func (s *Topo) ResetStreamOffset(name string, input map[string]interface{}) error

func (*Topo) SetSinkSchema added in v2.3.0 

func (s *Topo) SetSinkSchema(sinkSchema map[string]*ast.JsonStreamField)

func (*Topo) SetStreams 

func (s *Topo) SetStreams(streams []string)

Source Files

View all Source files

Directories

Path Synopsis
cache
cache
conf
metric
tracenode
machine
mockclock
mocknode

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.