processor

package
v1.3.4 Latest Latest
Warning

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

Go to latest
Published: Dec 5, 2022 License: AGPL-3.0 Imports: 42 Imported by: 0

Documentation

Index

Constants

View Source
const (
	METRICKEYDELIMITER  = "!<<#>>!"
	USER_TRANSFORMATION = "USER_TRANSFORMATION"
	DEST_TRANSFORMATION = "DEST_TRANSFORMATION"
	EVENT_FILTER        = "EVENT_FILTER"
)

Variables

View Source
var (
	GWCustomVal string
)

Functions

func ConvertToFilteredTransformerResponse added in v0.1.10

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

func Init added in v0.1.10

func Init()

func RegisterAdminHandlers added in v0.1.10

func RegisterAdminHandlers(readonlyProcErrorDB jobsdb.ReadonlyJobsDB)

func SetFeaturesRetryAttempts added in v0.1.10

func SetFeaturesRetryAttempts(overrideAttempts int)

Types

type DestStatT

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

type HandleT

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

HandleT is a handle to this object used in main.go

func (*HandleT) Setup

func (proc *HandleT) 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,
)

Setup initializes the module

func (*HandleT) Shutdown added in v0.1.10

func (proc *HandleT) Shutdown()

func (*HandleT) Start added in v0.1.10

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

Start starts this processor's main loops.

func (*HandleT) Status added in v0.1.10

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

func (*HandleT) Store added in v0.1.10

func (proc *HandleT) Store(in *storeMessage)

type LifecycleManager added in v0.1.10

type LifecycleManager struct {
	HandleT *HandleT

	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 added in v0.1.10

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,
) *LifecycleManager

New creates a new Processor instance

func (*LifecycleManager) Start added in v0.1.10

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 added in v0.1.10

func (proc *LifecycleManager) Stop()

Stop stops the processor, this is a blocking call.

type MetricMetadata added in v0.1.10

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

type ParametersT added in v0.1.10

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"`
	SourceBatchID           string      `json:"source_batch_id"`
	SourceTaskID            string      `json:"source_task_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 added in v0.1.10

type SourceIDT string

type TrackingPlanStatT added in v0.1.10

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

type TransformRequestT added in v0.1.10

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

type WriteKeyT added in v0.1.10

type WriteKeyT string

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL