igeist

Documentation

Index

• type Executor
• type Registry
• type Stream
• type StreamBuilder
• type StreamEntityFactory
• type StreamRegistry
• type Transformer

Types

type Executor

type Executor interface {
	Stream() Stream
	StreamId() string
	Metrics() entity.Metrics
	Run(ctx context.Context, wg *sync.WaitGroup)

	// ProcessEvent returns the processed/persisted event's event/resource ID, together with error, retryable.
	// The implementation varies across different sink types, but the returned resource ID it will be the key
	// used for key lookups of the event (e.g. row-key for BigTable), and will be used as the resource ID when
	// using API as Source.
	ProcessEvent(ctx context.Context, events []entity.Event) entity.EventProcessingResult
	Shutdown(ctx context.Context)
}

Executor interface required for Stream ETL Executors

type Registry

type Registry interface {
	Fetch(ctx context.Context) error
	Put(ctx context.Context, id string, spec *entity.Spec) error
	Get(ctx context.Context, id string) (*entity.Spec, error)
	GetAll(ctx context.Context) (map[string]*entity.Spec, error)
	Delete(ctx context.Context, id string) error
	Exists(id string) bool
	ExistsWithSameOrHigherVersion(specData []byte) (bool, error)
	Validate(spec []byte) (*entity.Spec, error)
}

type Stream

type Stream interface {
	Spec() *entity.Spec
	Instance() string
	Extractor() entity.Extractor
	Transformer() Transformer
	Loader() entity.Loader
	Publish(ctx context.Context, event []byte) (string, error)
	ExtractFromSink(
		ctx context.Context,
		query entity.ExtractorQuery,
		result *[]*entity.Transformed) (error, bool)
}

type StreamBuilder

type StreamBuilder interface {
	Build(ctx context.Context, spec *entity.Spec) (Stream, error)
}

type StreamEntityFactory

type StreamEntityFactory interface {
	SetAdminLoader(loader entity.Loader)
	CreateExtractor(ctx context.Context, spec *entity.Spec, instanceId string) (entity.Extractor, error)
	CreateSinkExtractor(ctx context.Context, spec *entity.Spec, instanceId string) (entity.Extractor, error)
	CreateTransformer(ctx context.Context, spec *entity.Spec) (Transformer, error)
	CreateLoader(ctx context.Context, spec *entity.Spec, instanceId string) (entity.Loader, error)
}

type StreamRegistry

type StreamRegistry interface {
	Registry
	Executor
	SetAdminStream(stream Stream)
}

A StreamRegistry is a Registry also managing a stream as an Executor, receiving Specs to be added to registry in an arbitrary ETL stream, e.g. via Kafka or REST API into a db. It is also required to publish admin events when registry is modified (e.g. new stream registrations, etc).

type Transformer

type Transformer interface {
	// Based on input event data, and the transformation rules in the Spec, the Transform() function returns
	// a key-value map where keys are "id" fields from Transform spec, and values are the transformation results.
	// The output from Transform() can contain multiple new events, in case of applied
	// event-split transformations.
	// If Transform() succeeded, but the transformation resulted in no output, e.g. for non-applicable incoming events,
	// the return values are nil, nil (i.e., not regarded as an error).
	Transform(
		ctx context.Context,
		event []byte,
		retryable *bool) ([]*entity.Transformed, error)
}

Transformer interface required for transformer implementations