common

package
v0.0.0-...-11310a4 Latest Latest
Warning

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

 Go to latest Published: Nov 2, 2021 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

Constants

View Source 
const (
	HeaderXAppAuthToken   = "X-SS-App-Auth-Token"
	HeaderXAppComponentId = "X-SS-App-Component-Id"
	HeaderXUserAuthToken  = "X-SS-User-Auth-Token"
	HeaderXRestCall       = "X-Requested-By"
	HeaderContentType     = "Content-Type"
	ApplicationJson       = "application/json"
	HeaderXRestCallValue  = "true"
	HttpPost              = "POST"
)
View Source 
const (
	ROOT = "ROOT"
	MAP  = "MAP"
	LIST = "LIST"

	InvalidFieldPath       = "Invalid fieldPath '%s' at char '%d'"
	InvalidFieldPathReason = "Invalid fieldPath '%s' at char position '%d' (%s)"
	ReasonEmptyFieldName   = "field name can't be empty"
	ReasonInvalidStart     = "field path needs to start with '[' or '/'"
	ReasonNotANumber       = "only numbers and '*' allowed between '[' and ']'"
	ReasonQuotes           = "quotes are not properly closed"
	InvalidFieldPathNumber = "Invalid fieldPath '%s' at char '%d' ('%s' needs to be a number or '*')"
)
View Source 
const (
	PipelineConfigSchemaVersion = 6
	PipelineConfigVersion       = 10
	ErrorRecordPolicyOriginal   = "ORIGINAL_RECORD"
	ErrorRecordPolicyStage      = "STAGE_RECORD"
	FragmentSourceStageName     = "com_streamsets_pipeline_stage_origin_fragment_FragmentSource"
	FragmentProcessorStageName  = "com_streamsets_pipeline_stage_processor_fragment_FragmentProcessor"
	FragmentTargetStageName     = "com_streamsets_pipeline_stage_destination_fragment_FragmentTarget"
	ConfFragmentId              = "conf.fragmentId"
	ConfFragmentInstanceId      = "conf.fragmentInstanceId"
)
View Source 
const (
	EDITED        = "EDITED"        // pipeline job has been create/modified, didn't run since the creation/modification
	STARTING      = "STARTING"      // pipeline job starting (initialization)
	START_ERROR   = "START_ERROR"   // pipeline job failed while start (during initialization)
	RUNNING       = "RUNNING"       // pipeline job running
	RUNNING_ERROR = "RUNNING_ERROR" // pipeline job failed while running (calling destroy on pipeline)
	RUN_ERROR     = "RUN_ERROR"     // pipeline job failed while running (done)
	FINISHING     = "FINISHING"     // pipeline job finishing (source reached end, returning NULL offset) (calling destroy on pipeline)
	FINISHED      = "FINISHED"      // pipeline job finished
	RETRY         = "RETRY"         // pipeline job retrying
	STOPPING      = "STOPPING"      // pipeline job has been manually stopped (calling destroy on pipeline)
	STOPPED       = "STOPPED"       // pipeline job has been manually stopped (done)
)
View Source 
const (
	CurrentOffsetVersion = 2
	PollSourceOffsetKey  = "$com.streamsets.sdc2go.pollsource.offset$"
)
View Source 
const (
	ConfigDefTagName      = "ConfigDef"
	ConfigDefBeanTagName  = "ConfigDefBean"
	ListBeanModelTagName  = "ListBeanModel"
	PredicateModelTagName = "PredicateModel"
	EvaluationExplicit    = "EXPLICIT"
	EvaluationImplicit    = "IMPLICIT"
)
View Source 
const (
	EdgeIdFile = "/data/edge.id"
)
View Source 
const StageConfig = "STAGE_CONFIG"

Variables

View Source 
var (
	BuiltBy           string
	BuiltDate         string
	BuiltRepoSha      string
	SourceMd5Checksum string
	Version           string
)
View Source 
var RootPathElement = &PathElement{
	Type: ROOT,
	Name: "",
	Idx:  0,
}

Functions

func AddStageToStagePath 

func AddStageToStagePath(header *HeaderImpl, stageInstanceName string)

func CreateRecordId 

func CreateRecordId(prefix string, counter int) string

func CreateTrackingId 

func CreateTrackingId(header *HeaderImpl)

Types

type BaseStage 

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

func (*BaseStage) Destroy 

func (b *BaseStage) Destroy() error

func (*BaseStage) GetDataGeneratorService 

func (b *BaseStage) GetDataGeneratorService() (dataformats.DataFormatGeneratorService, error)

func (*BaseStage) GetDataParserService 

func (b *BaseStage) GetDataParserService() (dataformats.DataFormatParserService, error)

func (*BaseStage) GetStageConfig 

func (b *BaseStage) GetStageConfig() *StageConfiguration

func (*BaseStage) GetStageContext 

func (b *BaseStage) GetStageContext() api.StageContext

func (*BaseStage) Init 

func (b *BaseStage) Init(stageContext api.StageContext) []validation.Issue

type BuildInfo 

type BuildInfo struct {
	BuiltBy           string `json:"builtBy"`
	BuiltDate         string `json:"builtDate"`
	BuiltRepoSha      string `json:"builtRepoSha"`
	SourceMd5Checksum string `json:"sourceMd5Checksum"`
	Version           string `json:"version"`
}

func NewBuildInfo 

func NewBuildInfo() (*BuildInfo, error)

type Config 

type Config struct {
	Name  string      `json:"name"`
	Value interface{} `json:"value"`
}

type ConfigDefinition 

type ConfigDefinition struct {
	Name       string
	Type       string
	Required   bool
	FieldName  string
	Evaluation string
	Model      ModelDefinition
}

type ErrorSink 

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

func NewErrorSink 

func NewErrorSink() *ErrorSink

func (*ErrorSink) ClearErrorRecordsAndMessages 

func (e *ErrorSink) ClearErrorRecordsAndMessages()

After each batch call this function to clear current batch error messages/records

func (*ErrorSink) GetErrorMessages 

func (e *ErrorSink) GetErrorMessages() map[string][]api.ErrorMessage

func (*ErrorSink) GetErrorRecords 

func (e *ErrorSink) GetErrorRecords() map[string][]api.Record

func (*ErrorSink) GetStageErrorMessages 

func (e *ErrorSink) GetStageErrorMessages(stageIns string) []api.ErrorMessage

func (*ErrorSink) GetStageErrorRecords 

func (e *ErrorSink) GetStageErrorRecords(stageIns string) []api.Record

func (*ErrorSink) GetTotalErrorMessages 

func (e *ErrorSink) GetTotalErrorMessages() int64

func (*ErrorSink) GetTotalErrorRecords 

func (e *ErrorSink) GetTotalErrorRecords() int64

func (*ErrorSink) ReportError 

func (e *ErrorSink) ReportError(stageIns string, errorMessage api.ErrorMessage)

func (*ErrorSink) ToError 

func (e *ErrorSink) ToError(stageIns string, record api.Record)

type EventSink 

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

func NewEventSink 

func NewEventSink() *EventSink

func (*EventSink) AddEvent 

func (e *EventSink) AddEvent(stageIns string, record api.Record)

func (*EventSink) ClearEventRecords 

func (e *EventSink) ClearEventRecords()

func (*EventSink) GetStageEvents 

func (e *EventSink) GetStageEvents(stageIns string) []api.Record

type HeaderImpl 

type HeaderImpl struct {
	StageCreator         string                 `json:"stageCreator"`
	SourceId             string                 `json:"sourceId"`
	StagesPath           string                 `json:"stagesPath"`
	TrackingId           string                 `json:"trackingId"`
	PreviousTrackingId   string                 `json:"previousTrackingId"`
	ErrorDataCollectorId string                 `json:"errorDataCollectorId"`
	ErrorPipelineName    string                 `json:"errorPipelineName"`
	ErrorStageInstance   string                 `json:"errorStage"`
	ErrorMessage         string                 `json:"errorMessage"`
	ErrorTimestamp       int64                  `json:"errorTimestamp"`
	Attributes           map[string]interface{} `json:"values"`
	// contains filtered or unexported fields
}

func (*HeaderImpl) GetAttribute 

func (h *HeaderImpl) GetAttribute(name string) interface{}

func (*HeaderImpl) GetAttributeNames 

func (h *HeaderImpl) GetAttributeNames() []string

func (*HeaderImpl) GetAttributes 

func (h *HeaderImpl) GetAttributes() map[string]string

func (*HeaderImpl) GetErrorDataCollectorId 

func (h *HeaderImpl) GetErrorDataCollectorId() string

func (*HeaderImpl) GetErrorMessage 

func (h *HeaderImpl) GetErrorMessage() string

func (*HeaderImpl) GetErrorPipelineName 

func (h *HeaderImpl) GetErrorPipelineName() string

func (*HeaderImpl) GetErrorStage 

func (h *HeaderImpl) GetErrorStage() string

func (*HeaderImpl) GetErrorTimestamp 

func (h *HeaderImpl) GetErrorTimestamp() int64

func (*HeaderImpl) GetPreviousTrackingId 

func (h *HeaderImpl) GetPreviousTrackingId() string

func (*HeaderImpl) GetSourceId 

func (h *HeaderImpl) GetSourceId() string

func (*HeaderImpl) GetSourceRecord 

func (h *HeaderImpl) GetSourceRecord() api.Record

func (*HeaderImpl) GetStageCreator 

func (h *HeaderImpl) GetStageCreator() string

func (*HeaderImpl) GetStagesPath 

func (h *HeaderImpl) GetStagesPath() string

func (*HeaderImpl) GetTrackingId 

func (h *HeaderImpl) GetTrackingId() string

func (*HeaderImpl) SetAttribute 

func (h *HeaderImpl) SetAttribute(name string, value string)

func (*HeaderImpl) SetErrorDataCollectorId 

func (h *HeaderImpl) SetErrorDataCollectorId(errorDataCollectorId string)

func (*HeaderImpl) SetErrorMessage 

func (h *HeaderImpl) SetErrorMessage(errorMessage string)

func (*HeaderImpl) SetErrorPipelineName 

func (h *HeaderImpl) SetErrorPipelineName(errorPipelineName string)

func (*HeaderImpl) SetErrorStageInstance 

func (h *HeaderImpl) SetErrorStageInstance(errorStageInstance string)

func (*HeaderImpl) SetErrorTimeStamp 

func (h *HeaderImpl) SetErrorTimeStamp(timeStamp int64)

func (*HeaderImpl) SetPreviousTrackingId 

func (h *HeaderImpl) SetPreviousTrackingId(previousTrackingId string)

func (*HeaderImpl) SetSourceId 

func (h *HeaderImpl) SetSourceId(sourceId string)

func (*HeaderImpl) SetSourceRecord 

func (h *HeaderImpl) SetSourceRecord(sourceRecord api.Record)

func (*HeaderImpl) SetStageCreator 

func (h *HeaderImpl) SetStageCreator(stageCreator string)

func (*HeaderImpl) SetStagesPath 

func (h *HeaderImpl) SetStagesPath(stagesPath string)

func (*HeaderImpl) SetTrackingId 

func (h *HeaderImpl) SetTrackingId(trackingId string)

type ModelDefinition 

type ModelDefinition struct {
	ConfigDefinitionsMap map[string]*ConfigDefinition
}

type PathElement 

type PathElement struct {
	Type string
	Name string
	Idx  int
}

func CreateListElement 

func CreateListElement(idx int) PathElement

func CreateMapElement 

func CreateMapElement(name string) PathElement

func ParseFieldPath 

func ParseFieldPath(fieldPath string, isSingleQuoteEscaped bool) ([]PathElement, error)

type PipelineConfiguration 

type PipelineConfiguration struct {
	SchemaVersion        int                              `json:"schemaVersion"`
	Version              int                              `json:"version"`
	PipelineId           string                           `json:"pipelineId"`
	Title                string                           `json:"title"`
	Description          string                           `json:"description"`
	UUID                 string                           `json:"uuid"`
	Configuration        []Config                         `json:"configuration"`
	UiInfo               map[string]interface{}           `json:"uiInfo"`
	Stages               []*StageConfiguration            `json:"stages"`
	ErrorStage           *StageConfiguration              `json:"errorStage"`
	TestOriginStage      *StageConfiguration              `json:"testOriginStage"`
	StatsAggregatorStage *StageConfiguration              `json:"statsAggregatorStage"`
	Previewable          bool                             `json:"previewable"`
	Info                 PipelineInfo                     `json:"info"`
	Metadata             map[string]interface{}           `json:"metadata"`
	Fragments            []*PipelineFragmentConfiguration `json:"fragments"`
}

func (*PipelineConfiguration) ProcessFragmentStages 

func (p *PipelineConfiguration) ProcessFragmentStages()

type PipelineEnvelope 

type PipelineEnvelope struct {
	PipelineConfig     PipelineConfiguration  `json:"pipelineConfig"`
	PipelineRules      map[string]interface{} `json:"pipelineRules"`
	LibraryDefinitions map[string]interface{} `json:"libraryDefinitions"`
}

type PipelineFragmentConfiguration 

type PipelineFragmentConfiguration struct {
	SchemaVersion      int                              `json:"schemaVersion"`
	Version            int                              `json:"version"`
	PipelineId         string                           `json:"fragmentId"`
	FragmentInstanceId string                           `json:"fragmentInstanceId"`
	Title              string                           `json:"title"`
	Description        string                           `json:"description"`
	UUID               string                           `json:"uuid"`
	Configuration      []Config                         `json:"configuration"`
	UiInfo             map[string]interface{}           `json:"uiInfo"`
	Stages             []*StageConfiguration            `json:"stages"`
	TestOriginStage    *StageConfiguration              `json:"testOriginStage"`
	Previewable        bool                             `json:"previewable"`
	Info               PipelineInfo                     `json:"info"`
	Metadata           map[string]interface{}           `json:"metadata"`
	Fragments          []*PipelineFragmentConfiguration `json:"fragments"`
}

type PipelineInfo 

type PipelineInfo struct {
	PipelineId   string                 `json:"pipelineId"`
	Title        string                 `json:"title"`
	Description  string                 `json:"description"`
	Created      int64                  `json:"created"`
	LastModified int64                  `json:"lastModified"`
	Creator      string                 `json:"creator"`
	LastModifier string                 `json:"lastModifier"`
	LastRev      string                 `json:"lastRev"`
	UUID         string                 `json:"uuid"`
	Valid        bool                   `json:"valid"`
	Metadata     map[string]interface{} `json:"metadata"`
	Name         string                 `json:"name"`
	SdcVersion   string                 `json:"sdcVersion"`
	SdcId        string                 `json:"sdcId"`
}

type PipelineState 

type PipelineState struct {
	PipelineId string                 `json:"pipelineId"`
	Status     string                 `json:"status"`
	Message    string                 `json:"message"`
	TimeStamp  int64                  `json:"timeStamp"`
	Attributes map[string]interface{} `json:"attributes"`
	Metrics    string                 `json:"metrics"`
}

type RecordImpl 

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

func (*RecordImpl) Clone 

func (r *RecordImpl) Clone() api.Record

func (*RecordImpl) Delete 

func (r *RecordImpl) Delete(fieldPath string) (*api.Field, error)

func (*RecordImpl) Get 

func (r *RecordImpl) Get(fieldPath ...string) (*api.Field, error)

func (*RecordImpl) GetFieldPaths 

func (r *RecordImpl) GetFieldPaths() map[string]bool

func (*RecordImpl) GetHeader 

func (r *RecordImpl) GetHeader() api.Header

func (*RecordImpl) Set 

func (r *RecordImpl) Set(field *api.Field) *api.Field

func (*RecordImpl) SetField 

func (r *RecordImpl) SetField(fieldPath string, field *api.Field) (*api.Field, error)

type RuntimeInfo 

type RuntimeInfo struct {
	ID           string
	BaseDir      string
	HttpUrl      string
	DPMEnabled   bool
	AppAuthToken string
}

func NewRuntimeInfo 

func NewRuntimeInfo(httpUrl string, baseDir string) (*RuntimeInfo, error)

type ServiceConfiguration 

type ServiceConfiguration struct {
	Service        string      `json:"service"`
	ServiceVersion interface{} `json:"serviceVersion"` // interface{} is for backward compatible - SDC-10526
	Configuration  []Config    `json:"configuration"`
}

func (ServiceConfiguration) GetConfigurationMap 

func (s ServiceConfiguration) GetConfigurationMap() map[string]Config

type ServiceDefinition 

type ServiceDefinition struct {
	Name                 string
	Version              string
	ConfigDefinitionsMap map[string]*ConfigDefinition
}

type SourceOffset 

type SourceOffset struct {
	Version int
	Offset  map[string]*string
}

func GetDefaultOffset 

func GetDefaultOffset() SourceOffset

type StageConfiguration 

type StageConfiguration struct {
	InstanceName  string                  `json:"instanceName"`
	Library       string                  `json:"library"`
	StageName     string                  `json:"stageName"`
	StageVersion  string                  `json:"stageVersion"`
	Configuration []Config                `json:"configuration"`
	Services      []*ServiceConfiguration `json:"services"`
	UiInfo        map[string]interface{}  `json:"uiInfo"`
	InputLanes    []string                `json:"inputLanes"`
	OutputLanes   []string                `json:"outputLanes"`
	EventLanes    []string                `json:"eventLanes"`
}

func (StageConfiguration) GetConfigurationMap 

func (s StageConfiguration) GetConfigurationMap() map[string]Config

func (StageConfiguration) GetOutputAndEventLanes 

func (s StageConfiguration) GetOutputAndEventLanes() []string

type StageContextImpl 

type StageContextImpl struct {
	StageConfig       *StageConfiguration
	Parameters        map[string]interface{}
	Metrics           metrics.Registry
	ErrorSink         *ErrorSink
	EventSink         *EventSink
	ErrorStage        bool
	ErrorRecordPolicy string
	Services          map[string]api.Service
	ElContext         context.Context
	// contains filtered or unexported fields
}

func NewStageContext 

func NewStageContext(
	stageConfig *StageConfiguration,
	resolvedParameters map[string]interface{},
	metricRegistry metrics.Registry,
	errorSink *ErrorSink,
	errorStage bool,
	errorRecordPolicy string,
	services map[string]api.Service,
	elContext context.Context,
	eventSink *EventSink,
	isPreview bool,
) (*StageContextImpl, error)

func (*StageContextImpl) CreateConfigIssue 

func (s *StageContextImpl) CreateConfigIssue(error string, optional ...interface{}) validation.Issue

optional argument, first optional argument is configGroup, second optional argument- configName

func (*StageContextImpl) CreateEventRecord 

func (s *StageContextImpl) CreateEventRecord(
	recordSourceId string,
	value interface{},
	eventType string,
	eventVersion int,
) (api.Record, error)

func (*StageContextImpl) CreateRecord 

func (s *StageContextImpl) CreateRecord(recordSourceId string, value interface{}) (api.Record, error)

func (*StageContextImpl) Evaluate 

func (s *StageContextImpl) Evaluate(
	value string,
	configName string,
	ctx context.Context,
) (interface{}, error)

func (*StageContextImpl) GetMetrics 

func (s *StageContextImpl) GetMetrics() metrics.Registry

func (*StageContextImpl) GetOutputLanes 

func (s *StageContextImpl) GetOutputLanes() []string

func (*StageContextImpl) GetParameterValue 

func (s *StageContextImpl) GetParameterValue(paramName string) interface{}

func (*StageContextImpl) GetPipelineParameters 

func (s *StageContextImpl) GetPipelineParameters() map[string]interface{}

func (*StageContextImpl) GetResolvedValue 

func (s *StageContextImpl) GetResolvedValue(configValue interface{}) (interface{}, error)

func (*StageContextImpl) GetService 

func (s *StageContextImpl) GetService(serviceName string) (api.Service, error)

func (*StageContextImpl) IsErrorStage 

func (s *StageContextImpl) IsErrorStage() bool

func (*StageContextImpl) IsPreview 

func (s *StageContextImpl) IsPreview() bool

func (*StageContextImpl) IsStopped 

func (s *StageContextImpl) IsStopped() bool

func (*StageContextImpl) ReportError 

func (s *StageContextImpl) ReportError(err error)

func (*StageContextImpl) SetStop 

func (s *StageContextImpl) SetStop()

func (*StageContextImpl) ToError 

func (s *StageContextImpl) ToError(err error, record api.Record)

func (*StageContextImpl) ToEvent 

func (s *StageContextImpl) ToEvent(record api.Record)

type StageDefinition 

type StageDefinition struct {
	Name                 string
	Library              string
	Version              string
	ConfigDefinitionsMap map[string]*ConfigDefinition
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.