pipeline

Mar 16, 2023

Changes in this version

type TaskRunResults

+ func (trrs *TaskRunResults) GetNextTaskOf(task TaskRunResult) *TaskRunResult

Feb 14, 2023

Changes in this version

type BridgeTask

+ Headers string

+ type ORMConfig interface

+ JobPipelineMaxSuccessfulRuns func() uint64

Dec 8, 2022

Changes in this version

type BridgeTask

+ CacheTTL string

type Config

+ BridgeCacheTTL func() time.Duration

+ type ETHGetBlockTask struct

+ EVMChainID string

+ func (t *ETHGetBlockTask) Run(ctx context.Context, lggr logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *ETHGetBlockTask) Type() TaskType

type GetterFunc

+ func ValidDurationInSeconds(s string) GetterFunc

+ type LookupTask struct

+ Key string

+ func (t *LookupTask) Run(ctx context.Context, lggr logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *LookupTask) Type() TaskType

type TaskType

+ const TaskTypeETHGetBlock

+ const TaskTypeLookup

Sep 30, 2022

Changes in this version

type ETHCallTask

+ GasUnlimited string

+ type LengthTask struct

+ Input string

+ func (t *LengthTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *LengthTask) Type() TaskType

+ type LessThanTask struct

+ Left string

+ Right string

+ func (t *LessThanTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *LessThanTask) Type() TaskType

type TaskType

+ const TaskTypeLength

+ const TaskTypeLessThan

Sep 1, 2022

Changes in this version

+ type Base64EncodeTask struct

+ Input string

+ func (t *Base64EncodeTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *Base64EncodeTask) Type() TaskType

+ type HexEncodeTask struct

+ Input string

+ func (t *HexEncodeTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *HexEncodeTask) Type() TaskType

type Keypath

+ Parts []string

type Spec

+ ForwardingAllowed bool

type TaskType

+ const TaskTypeBase64Encode

+ const TaskTypeHexEncode

Aug 8, 2022

Changes in this version

+ const BlockhashStoreJobType

+ const BootstrapJobType

+ const CronJobType

+ const DirectRequestJobType

+ const FluxMonitorJobType

+ const KeeperJobType

+ const OffchainReporting2JobType

+ const OffchainReportingJobType

+ const VRFJobType

+ const WebhookJobType

+ func SelectGasLimit(cfg config.ChainScopedConfig, jobType string, specGasLimit *uint32) uint64

+ type Base64DecodeTask struct

+ Input string

+ func (t *Base64DecodeTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *Base64DecodeTask) Type() TaskType

+ type HexDecodeTask struct

+ Input string

+ func (t *HexDecodeTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *HexDecodeTask) Type() TaskType

type Spec

+ GasLimit *uint32

type TaskType

+ const TaskTypeBase64Decode

+ const TaskTypeHexDecode

Jul 20, 2022

Changes in this version

+ const KeepersObservationSource

type Spec

+ JobType string

Jun 21, 2022

Changes in this version

+ var ErrDivideByZero = errors.New("divide by zero")

+ var ErrDivisionOverlow = errors.New("division overflow")

+ var ErrInvalidMultiplier = errors.New("Invalid multiplier")

+ var ErrMultiplyOverlow = errors.New("multiply overflow")

+ type ConditionalTask struct

+ Data string

+ func (t *ConditionalTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *ConditionalTask) Type() TaskType

type ETHCallTask

+ From string

type ETHTxTask

+ FailOnRevert string

type HTTPTask

+ Headers string

type Run

+ FailSilently bool

type StringParam

+ func (s *StringParam) String() string

type TaskType

+ const TaskTypeConditional

May 2, 2022

Changes in this version

+ const KeypathSeparator

+ var ErrIndexOutOfRange = errors.New("index out of range")

+ var ErrVarsSetNested = errors.New("cannot set a nested key of a pipeline.Vars")

+ var ErrWrongKeypath = errors.New("wrong keypath format")

type JSONParseTask

+ Separator string

type JSONPathParam

+ func NewJSONPathParam(sep string) JSONPathParam

type Keypath

+ Part0 string

+ Part1 string

+ func NewKeypathFromString(keypathStr string) (Keypath, error)

type MaybeBigIntParam

+ func NewMaybeBigIntParam(n *big.Int) MaybeBigIntParam

type MaybeInt32Param

+ func NewMaybeInt32Param(n int32, isSet bool) MaybeInt32Param

type MaybeUint64Param

+ func NewMaybeUint64Param(n uint64, isSet bool) MaybeUint64Param

type ORM

+ InsertFinishedRuns func(run []*Run, saveSuccessfulTaskRuns bool, qopts ...pg.QOpt) (err error)

type Runner

+ InsertFinishedRuns func(runs []*Run, saveSuccessfulTaskRuns bool, qopts ...pg.QOpt) error

Mar 3, 2022

Changes in this version

+ var Client *http.Client

+ var ErrDisallowedIP = errors.New("disallowed IP")

+ var UnrestrictedClient *http.Client

type ETHTxTask

+ TransmitChecker string

type Graph

+ func (g *Graph) AddImplicitDependenciesAsEdges()

+ func (g *Graph) IsImplicitEdge(uid, vid int64) bool

+ func (g *Graph) NewEdge(from, to graph.Node) graph.Edge

+ type GraphEdge struct

+ func (e *GraphEdge) IsImplicit() bool

+ func (e *GraphEdge) SetIsImplicit(isImplicit bool)

+ type HTTPRequest struct

+ Config HTTPRequestConfig

+ Logger logger.Logger

+ Request *http.Request

+ func (h *HTTPRequest) SendRequest() (responseBody []byte, statusCode int, headers http.Header, err error)

+ type HTTPRequestConfig struct

+ AllowUnrestrictedNetworkAccess bool

+ SizeLimit int64

+ type LowercaseTask struct

+ Input string

+ func (t *LowercaseTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *LowercaseTask) Type() TaskType

type RunErrors

+ func (re RunErrors) ToError() error

+ type TaskDependency struct

+ InputTask Task

+ PropagateResult bool

type TaskType

+ const TaskTypeLowercase

+ const TaskTypeUppercase

+ type UppercaseTask struct

+ Input string

+ func (t *UppercaseTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *UppercaseTask) Type() TaskType

Jan 25, 2022

Changes in this version

+ func ParseETHABIArgsString(theABI []byte, isLog bool) (args abi.Arguments, indexedArgs abi.Arguments, _ error)

type CBORParseTask

+ Mode string

+ type ETHABIEncodeTask2 struct

+ ABI string

+ Data string

+ func (t *ETHABIEncodeTask2) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (Result, RunInfo)

+ func (t *ETHABIEncodeTask2) Type() TaskType

type ETHCallTask

+ EVMChainID string

+ ExtractRevertReason bool

+ Gas string

+ GasFeeCap string

+ GasPrice string

+ GasTipCap string

type ETHTxTask

+ EVMChainID string

+ Simulate string

type EstimateGasLimitTask

+ EVMChainID string

+ From string

+ type FailTask struct

+ Msg string

+ func (t *FailTask) Run(_ context.Context, _ logger.Logger, vars Vars, _ []Result) (Result, RunInfo)

+ func (t *FailTask) Type() TaskType

type FinalResult

+ AllErrors []error

+ FatalErrors []error

+ func (result FinalResult) HasFatalErrors() bool

type JSONSerializable

+ Valid bool

type MapParam

+ func (m MapParam) Map() map[string]interface{}

+ type MaybeBigIntParam struct

+ func (p *MaybeBigIntParam) UnmarshalPipelineParam(val interface{}) error

+ func (p MaybeBigIntParam) BigInt() *big.Int

+ type MemoTask struct

+ Value string

+ func (t *MemoTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (Result, RunInfo)

+ func (t *MemoTask) Type() TaskType

+ type MergeTask struct

+ Left string

+ Right string

+ func (t *MergeTask) Run(_ context.Context, _ logger.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *MergeTask) Type() TaskType

+ type Method struct

+ Inputs abi.Arguments

+ Name string

type ORM

+ GetQ func() pg.Q

+ InsertRun func(run *Run, qopts ...pg.QOpt) error

+ type ObjectParam struct

+ BoolValue BoolParam

+ DecimalValue DecimalParam

+ MapValue MapParam

+ SliceValue SliceParam

+ StringValue StringParam

+ Type ObjectType

+ func MustNewObjectParam(val interface{}) *ObjectParam

+ func (o *ObjectParam) UnmarshalPipelineParam(val interface{}) error

+ func (o ObjectParam) Marshal() (string, error)

+ func (o ObjectParam) MarshalJSON() ([]byte, error)

+ func (o ObjectParam) String() string

+ type ObjectType int

+ const BoolType

+ const DecimalType

+ const MapType

+ const NilType

+ const SliceType

+ const StringType

+ type ResumeRequest struct

+ Error null.String

+ Value json.RawMessage

+ func (rr ResumeRequest) ToResult() (Result, error)

type Run

+ AllErrors RunErrors

+ FatalErrors RunErrors

+ func (r *Run) StringAllErrors() []*string

+ func (r *Run) StringFatalErrors() []*string

+ func (r *Run) StringOutputs() ([]*string, error)

+ func (r Run) HasFatalErrors() bool

+ type RunInfo struct

+ IsPending bool

+ IsRetryable bool

type TaskType

+ const TaskTypeETHABIEncode2

+ const TaskTypeFail

+ const TaskTypeMemo

+ const TaskTypeMerge

Oct 19, 2021

Changes in this version

+ var ErrCancelled = errors.New("task run cancelled (fail early)")

type BaseTask

+ MaxBackoff time.Duration

+ MinBackoff time.Duration

+ Retries null.Uint32

+ func (t BaseTask) TaskMaxBackoff() time.Duration

+ func (t BaseTask) TaskMinBackoff() time.Duration

+ func (t BaseTask) TaskRetries() uint32

type Config

+ EvmGasLimitDefault func() uint64

+ EvmMaxQueuedTransactions func() uint64

+ MinRequiredOutgoingConfirmations func() uint64

type ETHTxTask

+ MinConfirmations string

+ type EstimateGasLimitTask struct

+ Data string

+ EvmGasLimit uint64

+ GasEstimator GasEstimator

+ Input string

+ Multiplier string

+ To string

+ func (t *EstimateGasLimitTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *EstimateGasLimitTask) Type() TaskType

+ type GasEstimator interface

+ EstimateGas func(ctx context.Context, call ethereum.CallMsg) (uint64, error)

type ORM

+ DeleteRun func(id int64) error

type Pipeline

+ func (p *Pipeline) RequiresPreInsert() bool

type Runner

+ OnRunFinished func(func(*Run))

+ ResumeRun func(taskID uuid.UUID, result interface{}) error

type Task

+ TaskMaxBackoff func() time.Duration

+ TaskMinBackoff func() time.Duration

+ TaskRetries func() uint32

type TaskRunResult

+ Attempts uint

type TaskType

+ const TaskTypeEstimateGasLimit

+ const TaskTypeVRFV2

+ type VRFTaskV2 struct

+ PublicKey string

+ RequestBlockHash string

+ RequestBlockNumber string

+ Topics string

+ func (t *VRFTaskV2) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *VRFTaskV2) Type() TaskType

Aug 9, 2021

Changes in this version

+ var ErrInputTaskErrored = errors.New("input task errored")

+ var ErrOverflow = errors.New("overflow")

type BaseTask

+ FailEarly string

+ type ETHKeyStore interface

+ GetRoundRobinAddress func(addrs ...common.Address) (common.Address, error)

type Run

+ FailEarly bool

+ type VRFKeyStore interface

+ GenerateProof func(k secp256k1.PublicKey, seed *big.Int) (vrfkey.Proof, error)

type VRFTask

+ PublicKey string

+ RequestBlockHash string

+ RequestBlockNumber string

+ Topics string

Jul 26, 2021

Changes in this version

+ var ErrPending = errors.New("pending")

type BridgeTask

+ Async string

type JSONSerializable

+ func (js *JSONSerializable) Empty() bool

type ORM

+ GetUnfinishedRuns func(now time.Time, fn func(run Run) error) error

+ StoreRun func(db *sql.DB, run *Run) (restart bool, err error)

+ UpdateTaskRunResult func(db *sql.DB, taskID uuid.UUID, result interface{}) (run Run, start bool, err error)

type Pipeline

+ func (p *Pipeline) ByDotID(id string) Task

+ func (p *Pipeline) HasAsync() bool

type Run

+ Async bool

+ Inputs JSONSerializable

+ Pending bool

+ State RunStatus

+ func (r *Run) ByDotID(id string) *TaskRun

type RunStatus

+ const RunStatusRunning

+ const RunStatusSuspended

type Runner

+ Run func(ctx context.Context, run *Run, l logger.Logger, saveSuccessfulTaskRuns bool) (incomplete bool, err error)

type TaskRun

+ func (tr *TaskRun) IsPending() bool

type TaskRunResult

+ func (result *TaskRunResult) IsPending() bool

Jul 6, 2021

Changes in this version

+ var ErrTaskRunFailed = errors.New("task run failed")

+ var ErrTimeout = errors.New("timeout")

+ type AddressParam common.Address

+ func (a *AddressParam) UnmarshalPipelineParam(val interface{}) error

+ type AddressSliceParam []common.Address

+ func (s *AddressSliceParam) UnmarshalPipelineParam(val interface{}) error

type BaseTask

+ func (t *BaseTask) Base() *BaseTask

+ func (t BaseTask) ID() int

+ func (t BaseTask) Inputs() []Task

+ func (t BaseTask) Outputs() []Task

+ type CBORParseTask struct

+ Data string

+ func (t *CBORParseTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *CBORParseTask) Type() TaskType

type Config

+ EthGasLimitDefault func() uint64

+ EthMaxQueuedTransactions func() uint64

+ type DivideTask struct

+ Divisor string

+ Input string

+ Precision string

+ func (t *DivideTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *DivideTask) Type() TaskType

+ type ETHABIDecodeLogTask struct

+ ABI string

+ Data string

+ Topics string

+ func (t *ETHABIDecodeLogTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *ETHABIDecodeLogTask) Type() TaskType

+ type ETHABIDecodeTask struct

+ ABI string

+ Data string

+ func (t *ETHABIDecodeTask) Run(_ context.Context, vars Vars, inputs []Result) Result

+ func (t *ETHABIDecodeTask) Type() TaskType

+ type ETHABIEncodeTask struct

+ ABI string

+ Data string

+ func (t *ETHABIEncodeTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *ETHABIEncodeTask) Type() TaskType

+ type ETHCallTask struct

+ Contract string

+ Data string

+ func (t *ETHCallTask) Run(ctx context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *ETHCallTask) Type() TaskType

+ type ETHTxTask struct

+ Data string

+ From string

+ GasLimit string

+ To string

+ TxMeta string

+ func (t *ETHTxTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *ETHTxTask) Type() TaskType

type ErrRunPanicked

+ func (err ErrRunPanicked) Error() string

+ type Graph struct

+ func NewGraph() *Graph

+ func (g *Graph) NewNode() graph.Node

+ func (g *Graph) UnmarshalText(bs []byte) (err error)

+ type GraphNode struct

+ func NewGraphNode(n graph.Node, dotID string, attrs map[string]string) *GraphNode

+ func (n *GraphNode) Attributes() []encoding.Attribute

+ func (n *GraphNode) DOTID() string

+ func (n *GraphNode) SetAttribute(attr encoding.Attribute) error

+ func (n *GraphNode) SetDOTID(id string)

+ func (n *GraphNode) String() string

+ type HashSliceParam []common.Hash

+ func (s *HashSliceParam) UnmarshalPipelineParam(val interface{}) error

+ type JSONPathParam []string

+ func (p *JSONPathParam) UnmarshalPipelineParam(val interface{}) error

+ type KeyStore interface

+ GetRoundRobinAddress func(addrs ...common.Address) (common.Address, error)

+ type MaybeInt32Param struct

+ func (p *MaybeInt32Param) UnmarshalPipelineParam(val interface{}) error

+ func (p MaybeInt32Param) Int32() (int32, bool)

+ type MeanTask struct

+ AllowedFaults string

+ Precision string

+ Values string

+ func (t *MeanTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *MeanTask) Type() TaskType

+ type ModeTask struct

+ AllowedFaults string

+ Values string

+ func (t *ModeTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *ModeTask) Type() TaskType

+ type Pipeline struct

+ Source string

+ Tasks []Task

+ func Parse(text string) (*Pipeline, error)

+ func (p *Pipeline) MinTimeout() (time.Duration, bool, error)

+ func (p *Pipeline) UnmarshalText(bs []byte) (err error)

type Spec

+ func (s Spec) Pipeline() (*Pipeline, error)

+ type SumTask struct

+ AllowedFaults string

+ Values string

+ func (t *SumTask) Run(_ context.Context, vars Vars, inputs []Result) (result Result)

+ func (t *SumTask) Type() TaskType

type Task

+ Base func() *BaseTask

+ ID func() int

+ Inputs func() []Task

+ Outputs func() []Task

type TaskType

+ const TaskTypeCBORParse

+ const TaskTypeDivide

+ const TaskTypeETHABIDecode

+ const TaskTypeETHABIDecodeLog

+ const TaskTypeETHABIEncode

+ const TaskTypeETHCall

+ const TaskTypeETHTx

+ const TaskTypeMean

+ const TaskTypeMode

+ const TaskTypeSum

+ type TxManager interface

+ CreateEthTransaction func(db *gorm.DB, fromAddress, toAddress common.Address, payload []byte, ...) (etx models.EthTx, err error)

type Vars

+ func (vars Vars) Copy() Vars

Jun 21, 2021

Changes in this version

+ const InputTaskKey

+ var ErrKeypathNotFound = errors.New("keypath not found")

+ var ErrKeypathTooDeep = errors.New("keypath too deep (maximum 2 keys)")

+ var ErrParameterEmpty = errors.New("parameter is empty")

+ var ErrTooManyErrors = errors.New("too many errors")

+ var ErrVarsRoot = errors.New("cannot get/set the root of a pipeline.Vars")

+ var PromPipelineRunErrors = promauto.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... })

+ var PromPipelineRunTotalTimeToCompletion = promauto.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... })

+ var PromPipelineTaskExecutionTime = promauto.NewGaugeVec(prometheus.GaugeOpts{ ... }, []string{ ... })

+ var PromPipelineTasksTotalFinished = promauto.NewCounterVec(prometheus.CounterOpts{ ... }, []string{ ... })

+ func CheckInputs(inputs []Result, minLen, maxLen, maxErrors int) ([]interface{}, error)

+ func ResolveParam(out PipelineParamUnmarshaler, getters []GetterFunc) error

type BaseTask

+ func (t BaseTask) NumPredecessors() int

+ type BoolParam bool

+ func (b *BoolParam) UnmarshalPipelineParam(val interface{}) error

+ type BytesParam string

+ func (b *BytesParam) UnmarshalPipelineParam(val interface{}) error

+ type DecimalParam decimal.Decimal

+ func (d *DecimalParam) UnmarshalPipelineParam(val interface{}) error

+ func (d DecimalParam) Decimal() decimal.Decimal

+ type DecimalSliceParam []decimal.Decimal

+ func (s *DecimalSliceParam) UnmarshalPipelineParam(val interface{}) error

+ type GetterFunc func() (interface{}, error)

+ func From(getters ...interface{}) []GetterFunc

+ func Input(inputs []Result, index int) GetterFunc

+ func Inputs(inputs []Result) GetterFunc

+ func JSONWithVarExprs(s string, vars Vars, allowErrors bool) GetterFunc

+ func NonemptyString(s string) GetterFunc

+ func VarExpr(s string, vars Vars) GetterFunc

type JSONParseTask

+ Data string

+ type Keypath [2][]byte

+ func (keypath Keypath) NumParts() int

+ func (keypath Keypath) String() string

+ type MapParam map[string]interface

+ func (m *MapParam) UnmarshalPipelineParam(val interface{}) error

+ type MaybeUint64Param struct

+ func (u *MaybeUint64Param) UnmarshalPipelineParam(val interface{}) error

+ func (u MaybeUint64Param) Uint64() (uint64, bool)

type MedianTask

+ Values string

type MultiplyTask

+ Input string

type ORM

+ GetAllRuns func() ([]Run, error)

+ type PipelineParamUnmarshaler interface

+ UnmarshalPipelineParam func(val interface{}) error

type Runner

+ TestInsertFinishedRun func(db *gorm.DB, jobID int32, jobName string, jobType string, specID int32) (int64, error)

+ type SliceParam []interface

+ func (s *SliceParam) UnmarshalPipelineParam(val interface{}) error

+ func (s SliceParam) FilterErrors() (SliceParam, int)

+ type StringParam string

+ func (s *StringParam) UnmarshalPipelineParam(val interface{}) error

+ type StringSliceParam []string

+ func (p *StringSliceParam) UnmarshalPipelineParam(val interface{}) error

type Task

+ NumPredecessors func() int

type TaskType

+ const TaskTypeVRF

+ type URLParam url.URL

+ func (u *URLParam) String() string

+ func (u *URLParam) UnmarshalPipelineParam(val interface{}) error

+ type Uint64Param uint64

+ func (u *Uint64Param) UnmarshalPipelineParam(val interface{}) error

+ type VRFTask struct

+ func (t *VRFTask) Run(_ context.Context, vars Vars, _ JSONSerializable, inputs []Result) (result Result)

+ func (t *VRFTask) Type() TaskType

+ type Vars struct

+ func NewVarsFrom(m map[string]interface{}) Vars

+ func (vars Vars) Get(keypathStr string) (interface{}, error)

+ func (vars Vars) Set(dotID string, value interface{})

May 10, 2021

Changes in this version

+ type TaskDAGNode struct

+ func NewTaskDAGNode(n graph.Node, dotID string, attrs map[string]string) *TaskDAGNode

+ func (n *TaskDAGNode) Attributes() []encoding.Attribute

+ func (n *TaskDAGNode) DOTID() string

+ func (n *TaskDAGNode) SetAttribute(attr encoding.Attribute) error

+ func (n *TaskDAGNode) SetDAG(g *TaskDAG)

+ func (n *TaskDAGNode) SetDOTID(id string)

+ func (n *TaskDAGNode) String() string

type TaskType

+ func (t TaskType) String() string

Apr 26, 2021

Changes in this version

type BridgeTask

+ IncludeInputAtKey string

type ORM

+ InsertFinishedRun func(ctx context.Context, run Run, trrs []TaskRunResult, ...) (runID int64, err error)

type Runner

+ ExecuteAndInsertFinishedRun func(ctx context.Context, spec Spec, meta JSONSerializable, l logger.Logger, ...) (runID int64, finalResult FinalResult, err error)

+ InsertFinishedRun func(ctx context.Context, run Run, trrs TaskRunResults, saveSuccessfulTaskRuns bool) (int64, error)

type Spec

+ JobID int32

+ JobName string

type TaskRunResult

+ CreatedAt time.Time

Apr 12, 2021

Changes in this version

+ var ErrRunPanicked = errors.New("pipeline run panicked")

type BaseTask

+ func (t BaseTask) NPreds() int

+ type PanicTask struct

+ Msg string

+ func (t *PanicTask) Run(_ context.Context, _ TaskRun, _ []Result) (result Result)

+ func (t *PanicTask) SetDefaults(_ map[string]string, _ TaskDAG, _ taskDAGNode) error

+ func (t *PanicTask) Type() TaskType

+ type RunErrors []null.String

+ func (re *RunErrors) Scan(value interface{}) error

+ func (re RunErrors) HasError() bool

+ func (re RunErrors) Value() (driver.Value, error)

+ type SafeTx struct

type Spec

+ func (s Spec) TasksInDependencyOrder() ([]Task, error)

type Task

+ NPreds func() int

type TaskRun

+ Index int32

+ func (tr TaskRun) GetDotID() string

type TaskRunResult

+ Task Task

+ TaskRun TaskRun

type TaskType

+ const TaskTypePanic

Mar 22, 2021

Changes in this version

+ var ErrNoSuchBridge = errors.New("no such bridge exists")

type ORM

+ FindRun func(id int64) (Run, error)

type Run

+ func NewRun(spec Spec, startedAt time.Time) (run Run, err error)

+ func (r *Run) Status() RunStatus

+ type RunStatus int

+ const RunStatusCompleted

+ const RunStatusErrored

+ const RunStatusInProgress

+ const RunStatusUnknown

+ func (s RunStatus) Completed() bool

+ func (s RunStatus) Errored() bool

+ func (s RunStatus) Finished() bool

+ type RunWithResults struct

+ Run Run

+ TaskRunResults TaskRunResults

type Runner

+ InsertFinishedRunWithResults func(ctx context.Context, run Run, trrs TaskRunResults) (int64, error)

type TaskSpec

+ Bridge models.BridgeType

+ BridgeName *string

Feb 24, 2021

Changes in this version

+ type AnyTask struct

+ func (t *AnyTask) Run(_ context.Context, taskRun TaskRun, inputs []Result) (result Result)

+ func (t *AnyTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

+ func (t *AnyTask) Type() TaskType

type TaskType

+ const TaskTypeAny

Feb 1, 2021

Changes in this version

type Config

+ JobPipelineMaxRunDuration func() time.Duration

+ type FinalResult struct

+ Errors []error

+ Values []interface{}

+ func (result FinalResult) ErrorsDB() JSONSerializable

+ func (result FinalResult) HasErrors() bool

+ func (result FinalResult) OutputsDB() JSONSerializable

+ func (result FinalResult) SingularResult() (Result, error)

type JSONSerializable

+ Null bool

type ORM

+ DB func() *gorm.DB

+ InsertFinishedRunWithResults func(ctx context.Context, run Run, trrs []TaskRunResult) (runID int64, err error)

+ ProcessNextUnfinishedRun func(ctx context.Context, fn ProcessRunFunc) (bool, error)

+ type ProcessRunFunc func(ctx context.Context, txdb *gorm.DB, pRun Run) (TaskRunResults, error)

type Result

+ func (result Result) ErrorDB() null.String

+ func (result Result) OutputDB() JSONSerializable

type Run

+ func (r Run) FinalErrors() (f FinalErrors)

+ func (r Run) HasErrors() bool

type Runner

+ ExecuteAndInsertNewRun func(ctx context.Context, spec Spec) (finalResult FinalResult, err error)

+ ExecuteRun func(ctx context.Context, run Run) (trrs TaskRunResults, err error)

type Spec

+ PipelineTaskSpecs []TaskSpec

type TaskDAG

+ func (g TaskDAG) MinTimeout() (time.Duration, bool, error)

+ type TaskRunResult struct

+ FinishedAt time.Time

+ ID int64

+ IsTerminal bool

+ Result Result

+ TaskSpecID int32

+ type TaskRunResults []TaskRunResult

+ func (trrs TaskRunResults) FinalResult() (result FinalResult)

type TaskSpec

+ PipelineSpec Spec

Jan 18, 2021

Changes in this version

+ const DotStr

type BaseTask

+ func NewBaseTask(dotID string, t Task, index int32) BaseTask

type BridgeTask

+ func (t *BridgeTask) HelperSetConfigAndTxDB(config Config, txdb *gorm.DB)

+ func (t *BridgeTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

+ func (t BridgeTask) ExportedEquals(otherTask Task) bool

type HTTPTask

+ func (t *HTTPTask) HelperSetConfig(config Config)

+ func (t *HTTPTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

+ func (t HTTPTask) ExportedEquals(otherTask Task) bool

type JSONParseTask

+ func (t *JSONParseTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

+ func (t JSONParseTask) ExportedEquals(otherTask Task) bool

type MedianTask

+ AllowedFaults uint64

+ func (t *MedianTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

+ func (t MedianTask) ExportedEquals(otherTask Task) bool

type MultiplyTask

+ func (t *MultiplyTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

+ func (t MultiplyTask) ExportedEquals(otherTask Task) bool

type ResultTask

+ func (t *ResultTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

+ func (t ResultTask) ExportedEquals(otherTask Task) bool

type Runner

+ Close func() error

type Spec

+ MaxTaskDuration models.Interval

type Task

+ SetDefaults func(inputValues map[string]string, g TaskDAG, self taskDAGNode) error

Dec 14, 2020

Changes in this version

+ const MaybeBoolFalse

+ const MaybeBoolNull

+ const MaybeBoolTrue

type BaseTask

+ Timeout time.Duration

+ func (t BaseTask) TaskTimeout() (time.Duration, bool)

type FinalErrors

+ func (fe FinalErrors) HasErrors() bool

type HTTPTask

+ AllowUnrestrictedNetworkAccess MaybeBool

+ type MaybeBool string

+ func MaybeBoolFromString(s string) (MaybeBool, error)

+ func (m MaybeBool) Bool() (b bool, isSet bool)

type Task

+ TaskTimeout func() (time.Duration, bool)

Nov 23, 2020

Changes in this version

type Config

+ TriggerFallbackDBPollInterval func() time.Duration

type Run

+ Errors *JSONSerializable

+ Outputs *JSONSerializable

+ PipelineSpec Spec

+ PipelineTaskRuns []TaskRun

+ func (r *Run) SetID(value string) error

+ func (r Run) GetID() string

type TaskRun

+ Type TaskType

+ func (tr *TaskRun) SetID(value string) error

+ func (tr TaskRun) GetID() string

Nov 2, 2020

Changes in this version

+ const ResultTaskDotID

+ var ErrBadInput = errors.New("bad input for task")

+ var ErrWrongInputCardinality = errors.New("wrong number of task inputs")

+ func FindBridge(db *gorm.DB, name models.TaskType) (models.BridgeType, error)

+ func NewORM(db *gorm.DB, config Config, eventBroadcaster postgres.EventBroadcaster) *orm

+ func NewRunner(orm ORM, config Config) *runner

+ func WrapResultIfError(result *Result, msg string, args ...interface{})

+ type BaseTask struct

+ Index int32

+ func (t *BaseTask) SetOutputTask(outputTask Task)

+ func (t BaseTask) DotID() string

+ func (t BaseTask) OutputIndex() int32

+ func (t BaseTask) OutputTask() Task

+ type BridgeTask struct

+ Name string

+ RequestData HttpRequestData

+ func (t *BridgeTask) Run(ctx context.Context, taskRun TaskRun, inputs []Result) (result Result)

+ func (t *BridgeTask) Type() TaskType

+ type Config interface

+ BridgeResponseURL func() *url.URL

+ DatabaseMaximumTxDuration func() time.Duration

+ DatabaseURL func() string

+ DefaultHTTPAllowUnrestrictedNetworkAccess func() bool

+ DefaultHTTPLimit func() int64

+ DefaultHTTPTimeout func() models.Duration

+ DefaultMaxHTTPAttempts func() uint

+ JobPipelineDBPollInterval func() time.Duration

+ JobPipelineMaxTaskDuration func() time.Duration

+ JobPipelineParallelism func() uint8

+ JobPipelineReaperInterval func() time.Duration

+ JobPipelineReaperThreshold func() time.Duration

+ type FinalErrors []null.String

+ func (fe *FinalErrors) Scan(value interface{}) error

+ func (fe FinalErrors) Error() string

+ func (fe FinalErrors) Value() (driver.Value, error)

+ type HTTPTask struct

+ Method string

+ RequestData HttpRequestData

+ URL models.WebURL

+ func (t *HTTPTask) Run(ctx context.Context, taskRun TaskRun, inputs []Result) Result

+ func (t *HTTPTask) Type() TaskType

+ type HttpRequestData map[string]interface

+ func (h *HttpRequestData) Scan(value interface{}) error

+ func (h HttpRequestData) AsMap() map[string]interface{}

+ func (h HttpRequestData) Value() (driver.Value, error)

+ type JSONParseTask struct

+ Lax bool

+ Path JSONPath

+ func (t *JSONParseTask) Run(_ context.Context, taskRun TaskRun, inputs []Result) (result Result)

+ func (t *JSONParseTask) Type() TaskType

+ type JSONPath []string

+ func (p *JSONPath) Scan(value interface{}) error

+ func (p *JSONPath) UnmarshalText(bs []byte) error

+ func (p JSONPath) Value() (driver.Value, error)

+ type JSONSerializable struct

+ Val interface{}

+ func (js *JSONSerializable) Scan(value interface{}) error

+ func (js *JSONSerializable) UnmarshalJSON(bs []byte) error

+ func (js JSONSerializable) MarshalJSON() ([]byte, error)

+ func (js JSONSerializable) Value() (driver.Value, error)

+ type MedianTask struct

+ func (t *MedianTask) Run(_ context.Context, taskRun TaskRun, inputs []Result) (result Result)

+ func (t *MedianTask) Type() TaskType

+ type MultiplyTask struct

+ Times decimal.Decimal

+ func (t *MultiplyTask) Run(_ context.Context, taskRun TaskRun, inputs []Result) (result Result)

+ func (t *MultiplyTask) Type() TaskType

+ type ORM interface

+ AwaitRun func(ctx context.Context, runID int64) error

+ CreateRun func(ctx context.Context, jobID int32, meta map[string]interface{}) (int64, error)

+ CreateSpec func(ctx context.Context, taskDAG TaskDAG) (int32, error)

+ DeleteRunsOlderThan func(threshold time.Duration) error

+ FindBridge func(name models.TaskType) (models.BridgeType, error)

+ ListenForNewRuns func() (postgres.Subscription, error)

+ ProcessNextUnclaimedTaskRun func(ctx context.Context, fn ProcessTaskRunFunc) (bool, error)

+ ResultsForRun func(ctx context.Context, runID int64) ([]Result, error)

+ RunFinished func(runID int64) (bool, error)

+ type PossibleErrorResponses struct

+ Error string

+ ErrorMessage string

+ type ProcessTaskRunFunc func(ctx context.Context, txdb *gorm.DB, jobID int32, ptRun TaskRun, ...) Result

+ type Result struct

+ Error error

+ Value interface{}

+ type ResultTask struct

+ func (t *ResultTask) Run(_ context.Context, taskRun TaskRun, inputs []Result) Result

+ func (t *ResultTask) Type() TaskType

+ type Run struct

+ CreatedAt time.Time

+ FinishedAt time.Time

+ ID int64

+ Meta JSONSerializable

+ PipelineSpecID int32

+ func (Run) TableName() string

+ type Runner interface

+ AwaitRun func(ctx context.Context, runID int64) error

+ CreateRun func(ctx context.Context, jobID int32, meta map[string]interface{}) (int64, error)

+ ResultsForRun func(ctx context.Context, runID int64) ([]Result, error)

+ Start func()

+ Stop func()

+ type Spec struct

+ CreatedAt time.Time

+ DotDagSource string

+ ID int32

+ func (Spec) TableName() string

+ type Task interface

+ DotID func() string

+ OutputIndex func() int32

+ OutputTask func() Task

+ Run func(ctx context.Context, taskRun TaskRun, inputs []Result) Result

+ SetOutputTask func(task Task)

+ Type func() TaskType

+ func UnmarshalTaskFromMap(taskType TaskType, taskMap interface{}, dotID string, config Config, ...) (_ Task, err error)

+ type TaskDAG struct

+ DOTSource string

+ func NewTaskDAG() *TaskDAG

+ func (g *TaskDAG) HasCycles() bool

+ func (g *TaskDAG) NewNode() graph.Node

+ func (g *TaskDAG) UnmarshalText(bs []byte) (err error)

+ func (g TaskDAG) TasksInDependencyOrder() ([]Task, error)

+ type TaskRun struct

+ CreatedAt time.Time

+ Error null.String

+ FinishedAt time.Time

+ ID int64

+ Output *JSONSerializable

+ PipelineRun Run

+ PipelineRunID int64

+ PipelineTaskSpec TaskSpec

+ PipelineTaskSpecID int32

+ func (TaskRun) TableName() string

+ func (r TaskRun) DotID() string

+ func (r TaskRun) Result() Result

+ type TaskSpec struct

+ CreatedAt time.Time

+ DotID string

+ ID int32

+ Index int32

+ JSON JSONSerializable

+ PipelineSpecID int32

+ SuccessorID null.Int

+ Type TaskType

+ func (TaskSpec) TableName() string

+ func (s TaskSpec) IsFinalPipelineOutput() bool

+ type TaskType string

+ const TaskTypeBridge

+ const TaskTypeHTTP

+ const TaskTypeJSONParse

+ const TaskTypeMedian

+ const TaskTypeMultiply

+ const TaskTypeResult