topology

type Builder interface {
	NewKSubTopologyBuilder(kind SubTopologyKind) SubTopologyBuilder
	RemoveSubTopology(builder SubTopologyBuilder)
	SubTopologies() SubTopologyBuilders
	Build(ctx BuilderContext) (Topology, error)
	Describe() string
	Reset(ctx BuilderContext) error
}

type BuilderContext interface {
	StoreRegistry() stores.Registry
	ProducerBuilder() kafka.ProducerBuilder
	Admin() kafka.Admin
	ApplicationId() string
	Logger() log.Logger
	MetricsReporter() metrics.Reporter
}

type ChangeLogger interface {
	Log(ctx context.Context, key, value []byte) error
}

type ChangelogSyncer interface {
	Sync(ctx context.Context, synced chan struct{}) error
	Stop() error
}

type ChangelogSyncerBuilder interface {
	// Setup setups the changelog by creating changelog topics and offset stores
	Setup(ctx SubTopologyBuilderContext) error
	Build(ctx SubTopologyContext, store stores.Store) (ChangelogSyncer, error)
	BuildLogger(ctx SubTopologyContext, store string) (ChangeLogger, error)
	Internal() bool
	Topic() string
}

type CloseableNode interface {
	Node
	// Close is called once the message processing stops
	// Please refer SubTopology.Close()
	Close() error
}

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

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

type InitableNode interface {
	Node
	// Init is called once the Node build is completed and before message processing starts.
	// Please refer SubTopology.Init()
	Init(ctx NodeContext) error
}

type KeyValPair struct {
	Key   interface{}
	Value interface{}
}

type LoggableStateStore interface {
	StateStore
	ChangeLogger
}

type LoggableStoreBuilder interface {
	Name() string
	NameFormatter(ctx SubTopologyContext) StateStoreNameFunc
	KeyEncoder() encoding.Encoder
	ValEncoder() encoding.Encoder
	Build(ctx SubTopologyContext) (StateStore, error)
	Changelog() ChangelogSyncerBuilder
}

type Node interface {
	NodeInfo
	Run(ctx base.Context, kIn, vIn interface{}) (kOut, vOut interface{}, cont bool, err error)
	AddEdge(node Node)
	Edges() []Node
}

type NodeBuilder interface {
	NodeInfo
	SetId(id NodeId)
	// Setup is called once when stream app starting (SubTopologyBuilder.Setup())
	// Eg usage: create changelog topics for node
	Setup(ctx SubTopologySetupContext) error
	// Build calls with every Consumer PartitionAssignEvent and this method is responsible to create a new instance of
	// the node
	Build(ctx SubTopologyContext) (Node, error)
}

type NodeContext interface {
	SubTopologyContext
	Store(name string) StateStore
}

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

type NodeInfo interface {
	// Id returns a unique node id within the topology
	Id() NodeId
	// Type returns type of the node eg: Joiner, Flatmap
	Type() Type
	ReadsFrom() []string
	WritesAt() []string
	NameAs(name string)
	NodeName() string
}

type RecodeContext interface {
	context.Context
}

type Sink interface {
	Node
	Encoder() SinkEncoder
	Topic() string

	// Close closes the source buffers
	Close() error
}

type SinkBuilder interface {
	NodeBuilder
	Topic() string
	AutoCreate() bool
}

type SinkEncoder struct {
	Key, Value encoding.Encoder
}

type Source interface {
	Node
	NodeBuilder
	Encoder() SourceEncoder
	Topic() string
	ShouldCoPartitionedWith(source Source)
	TopicConfigs() kafka.TopicConfig
	CoPartitionedWith() Source
	RePartitionedAs() Source
	AutoCreate() bool
	Internal() bool
	InitialOffset() kafka.Offset
}

type SourceEncoder struct {
	Key, Value encoding.Encoder
}

type State struct {
	Type  StateType
	Store LoggableStateStore
}

type StateBuilder struct {
	Type  StateType
	Store LoggableStateStore
}

type StateStore interface {
	stores.Store
	// Flush flashes the records in buffer to stores.Store
	Flush() error
	ResetCache()
	ChangelogSyncer
}

type StateStoreNameFunc func(store string) string

type StateStoreProvider interface {
	Store(ctx SubTopologyContext) LoggableStateStore
}

type StateType string

type SubTopology interface {
	Id() SubTopologyId
	Sinks() []Sink
	Source(topic string) Source
	Nodes() []Node
	Init(ctx SubTopologyContext) error
	Close() error
	Store(name string) StateStore
	StateStores() map[string]StateStore
}

type SubTopologyBuilder interface {
	Id() SubTopologyId
	SetId(SubTopologyId)
	AddNode(builder NodeBuilder)
	AddNodeWithEdge(node, edge NodeBuilder)
	Parent(node NodeBuilder) NodeBuilder
	NodeSource(node NodeBuilder) Source
	Setup(ctx SubTopologySetupContext) error
	Build(ctx SubTopologyContext) (SubTopology, error)
	AddEdge(parent, node NodeBuilder)
	RemoveNode(node NodeBuilder)
	RemoveAll()
	Edges() []Edge
	Sources() []Source
	MergeSubTopology(subTp SubTopologyBuilder)
	Nodes() []NodeBuilder
	AddStore(builder LoggableStoreBuilder)
	StateStores() map[string]LoggableStoreBuilder
	AddSource(source Source)
	Kind() SubTopologyKind
}

type SubTopologyBuilderContext interface {
	BuilderContext
	MaxPartitionCount() int32
}

type SubTopologyBuilders []SubTopologyBuilder

type SubTopologyContext interface {
	BuilderContext
	context.Context
	PartitionConsumer() kafka.PartitionConsumer
	Partition() int32
	Producer() kafka.Producer
	Logger() log.Logger
	TopicMeta() map[string]*kafka.Topic
}

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

type SubTopologyKind string

type SubTopologySetupContext interface {
	BuilderContext
	MaxPartitionCount() int32
	TopicMeta() map[string]*kafka.TopicConfig
}

type Topology interface {
	SubTopologies() []SubTopologyBuilder
	SubTopology(id SubTopologyId) SubTopologyBuilder
	SubTopologyByTopic(topic string) SubTopologyBuilder
	StreamTopologies() SubTopologyBuilders
	SourceByTopic(topic string) Source
	GlobalTableTopologies() SubTopologyBuilders
	Describe() string
}

type Type struct {
	Name  string
	Attrs map[string]string
}

type Visualizer interface {
	AddTopology(node Builder)
	Visualize() (string, error)
}