stream

Documentation

Index

• type Manager
  • func NewStreamManager(nameGenerator schema.NameGenerator, typeManager pgtypes.TypeManager, ...) (Manager, error)
• type Stream
  • func NewMessageStream(nameGenerator schema.NameGenerator, sinkManager sink.Manager) Stream
  • func NewTableStream(nameGenerator schema.NameGenerator, typeManager pgtypes.TypeManager, ...) Stream

Types

type Manager

type Manager interface {
	Start() error
	Stop() error
	GetStream(
		table schema.TableAlike,
	) (stream Stream, present bool)
	GetOrCreateStream(
		table schema.TableAlike,
	) Stream
}

func NewStreamManager

func NewStreamManager(
	nameGenerator schema.NameGenerator, typeManager pgtypes.TypeManager, sinkManager sink.Manager,
) (Manager, error)

type Stream

type Stream interface {
	KeySchema() schema.Struct
	Key(
		values map[string]any,
	) (schema.Struct, error)
	PayloadSchema() schema.Struct
	Emit(
		key, envelope schema.Struct,
	) error
}

func NewMessageStream

func NewMessageStream(
	nameGenerator schema.NameGenerator, sinkManager sink.Manager,
) Stream

func NewTableStream

func NewTableStream(
	nameGenerator schema.NameGenerator, typeManager pgtypes.TypeManager,
	sinkManager sink.Manager, tableDefinition schema.TableAlike,
) Stream