processor

Documentation

Index

• Constants
• func ConvertToFilteredTransformerResponse(events []transformer.TransformerEventT, filter bool) transformer.ResponseT
• func RegisterAdminHandlers(readonlyProcErrorDB jobsdb.ReadonlyJobsDB)
• func WithFeaturesRetryMaxAttempts(maxAttempts int) func(l *LifecycleManager)
• type DestStatT
• type Handle
  • func NewHandle(transformer transformer.Transformer) *Handle
  • func (proc *Handle) Setup(backendConfig backendconfig.BackendConfig, gatewayDB, routerDB jobsdb.JobsDB, ...)
  • func (proc *Handle) Shutdown()
  • func (proc *Handle) Start(ctx context.Context) error
  • func (proc *Handle) Status() interface{}
  • func (proc *Handle) Store(partition string, in *storeMessage)
• type LifecycleManager
  • func New(ctx context.Context, clearDb *bool, gwDb, rtDb, brtDb, errDb *jobsdb.HandleT, ...) *LifecycleManager
  • func (proc *LifecycleManager) Start() error
  • func (proc *LifecycleManager) Stop()
• type MetricMetadata
• type Opts
  • func WithAdaptiveLimit(adaptiveLimitFunction func(int64) int64) Opts
• type ParametersT
• type SourceIDT
• type TrackingPlanStatT
• type TransformRequestT
• type WriteKeyT

Constants

const (
	MetricKeyDelimiter = "!<<#>>!"
	UserTransformation = "USER_TRANSFORMATION"
	DestTransformation = "DEST_TRANSFORMATION"
	EventFilter        = "EVENT_FILTER"
)

Functions

func ConvertToFilteredTransformerResponse

func ConvertToFilteredTransformerResponse(events []transformer.TransformerEventT, filter bool) transformer.ResponseT

func RegisterAdminHandlers

func RegisterAdminHandlers(readonlyProcErrorDB jobsdb.ReadonlyJobsDB)

func WithFeaturesRetryMaxAttempts

func WithFeaturesRetryMaxAttempts(maxAttempts int) func(l *LifecycleManager)

Types

type DestStatT

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

type Handle

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

Handle is a handle to the processor module

func NewHandle

func NewHandle(transformer transformer.Transformer) *Handle

func (*Handle) Setup

func (proc *Handle) Setup(
	backendConfig backendconfig.BackendConfig, gatewayDB, routerDB jobsdb.JobsDB,
	batchRouterDB, errorDB jobsdb.JobsDB, clearDB *bool, reporting types.ReportingI,
	multiTenantStat multitenant.MultiTenantI, transientSources transientsource.Service,
	fileuploader fileuploader.Provider, rsourcesService rsources.JobService, destDebugger destinationdebugger.DestinationDebugger, transDebugger transformationdebugger.TransformationDebugger,
)

Setup initializes the module

func (*Handle) Shutdown

func (proc *Handle) Shutdown()

func (*Handle) Start

func (proc *Handle) Start(ctx context.Context) error

Start starts this processor's main loops.

func (*Handle) Status

func (proc *Handle) Status() interface{}

func (*Handle) Store

func (proc *Handle) Store(partition string, in *storeMessage)

type LifecycleManager

type LifecycleManager struct {
	Handle *Handle

	MultitenantStats multitenant.MultiTenantI // need not initialize again
	ReportingI       types.ReportingI         // need not initialize again
	BackendConfig    backendconfig.BackendConfig
	Transformer      transformer.Transformer
	// contains filtered or unexported fields
}

func New

func New(ctx context.Context, clearDb *bool, gwDb, rtDb, brtDb, errDb *jobsdb.HandleT,
	tenantDB multitenant.MultiTenantI, reporting types.ReportingI, transientSources transientsource.Service, fileuploader fileuploader.Provider,
	rsourcesService rsources.JobService, destDebugger destinationdebugger.DestinationDebugger, transDebugger transformationdebugger.TransformationDebugger,
	opts ...Opts,
) *LifecycleManager

New creates a new Processor instance

func (*LifecycleManager) Start

func (proc *LifecycleManager) Start() error

Start starts a processor, this is not a blocking call. If the processor is not completely started and the data started coming then also it will not be problematic as we are assuming that the DBs will be up.

func (*LifecycleManager) Stop

func (proc *LifecycleManager) Stop()

Stop stops the processor, this is a blocking call.

type MetricMetadata

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

type Opts

type Opts func(l *LifecycleManager)

func WithAdaptiveLimit

func WithAdaptiveLimit(adaptiveLimitFunction func(int64) int64) Opts

type ParametersT

type ParametersT struct {
	SourceID                string      `json:"source_id"`
	DestinationID           string      `json:"destination_id"`
	ReceivedAt              string      `json:"received_at"`
	TransformAt             string      `json:"transform_at"`
	MessageID               string      `json:"message_id"`
	GatewayJobID            int64       `json:"gateway_job_id"`
	SourceTaskRunID         string      `json:"source_task_run_id"`
	SourceJobID             string      `json:"source_job_id"`
	SourceJobRunID          string      `json:"source_job_run_id"`
	EventName               string      `json:"event_name"`
	EventType               string      `json:"event_type"`
	SourceDefinitionID      string      `json:"source_definition_id"`
	DestinationDefinitionID string      `json:"destination_definition_id"`
	SourceCategory          string      `json:"source_category"`
	RecordID                interface{} `json:"record_id"`
	WorkspaceId             string      `json:"workspaceId"`
}

type SourceIDT

type SourceIDT string

type TrackingPlanStatT

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

type TransformRequestT

type TransformRequestT struct {
	Event          []transformer.TransformerEventT
	Stage          string
	ProcessingTime float64
	Index          int
}

type WriteKeyT

type WriteKeyT string