pipeline

package
v2.0.0-rc3 Latest Latest
Warning

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

Go to latest
Published: Apr 25, 2023 License: MIT Imports: 62 Imported by: 0

Documentation

Index

Constants

View Source
const (
	CronJobType               string = "cron"
	DirectRequestJobType      string = "directrequest"
	FluxMonitorJobType        string = "fluxmonitor"
	OffchainReportingJobType  string = "offchainreporting"
	OffchainReporting2JobType string = "offchainreporting2"
	KeeperJobType             string = "keeper"
	VRFJobType                string = "vrf"
	BlockhashStoreJobType     string = "blockhashstore"
	BlockHeaderFeederJobType  string = "blockheaderfeeder"
	WebhookJobType            string = "webhook"
	BootstrapJobType          string = "bootstrap"
)
View Source
const (
	InputTaskKey = "input"
)
View Source
const KeepersObservationSource = `` /* 3338-byte string literal not displayed */

KeepersObservationSource is the same for all keeper jobs and it is not persisted in DB

View Source
const KeypathSeparator = "."

Variables

View Source
var (
	ErrWrongInputCardinality = errors.New("wrong number of task inputs")
	ErrBadInput              = errors.New("bad input for task")
	ErrInputTaskErrored      = errors.New("input task errored")
	ErrParameterEmpty        = errors.New("parameter is empty")
	ErrIndexOutOfRange       = errors.New("index out of range")
	ErrTooManyErrors         = errors.New("too many errors")
	ErrTimeout               = errors.New("timeout")
	ErrTaskRunFailed         = errors.New("task run failed")
	ErrCancelled             = errors.New("task run cancelled (fail early)")
)
View Source
var (
	// PromPipelineTaskExecutionTime reports how long each pipeline task took to execute
	// TODO: Make private again after
	// https://app.clubhouse.io/chainlinklabs/story/6065/hook-keeper-up-to-use-tasks-in-the-pipeline
	PromPipelineTaskExecutionTime = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Name: "pipeline_task_execution_time",
		Help: "How long each pipeline task took to execute",
	},
		[]string{"job_id", "job_name", "task_id", "task_type"},
	)
	PromPipelineRunErrors = promauto.NewCounterVec(prometheus.CounterOpts{
		Name: "pipeline_run_errors",
		Help: "Number of errors for each pipeline spec",
	},
		[]string{"job_id", "job_name"},
	)
	PromPipelineRunTotalTimeToCompletion = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Name: "pipeline_run_total_time_to_completion",
		Help: "How long each pipeline run took to finish (from the moment it was created)",
	},
		[]string{"job_id", "job_name"},
	)
	PromPipelineTasksTotalFinished = promauto.NewCounterVec(prometheus.CounterOpts{
		Name: "pipeline_tasks_total_finished",
		Help: "The total number of pipeline tasks which have finished",
	},
		[]string{"job_id", "job_name", "task_id", "task_type", "status"},
	)
)
View Source
var (
	ErrDivideByZero    = errors.New("divide by zero")
	ErrDivisionOverlow = errors.New("division overflow")
)
View Source
var (
	ErrKeypathNotFound = errors.New("keypath not found")
	ErrVarsRoot        = errors.New("cannot get/set the root of a pipeline.Vars")
	ErrVarsSetNested   = errors.New("cannot set a nested key of a pipeline.Vars")
)
View Source
var (
	ErrInvalidMultiplier = errors.New("Invalid multiplier")
)
View Source
var (
	ErrMultiplyOverlow = errors.New("multiply overflow")
)
View Source
var ErrOverflow = errors.New("overflow")
View Source
var (
	ErrWrongKeypath = errors.New("wrong keypath format")
)

Functions

func CheckInputs

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

func NewORM

func NewORM(db *sqlx.DB, lggr logger.Logger, cfg ORMConfig) *orm

func NewRunner

func NewRunner(orm ORM, btORM bridges.ORM, cfg Config, chainSet evm.ChainSet, ethks ETHKeyStore, vrfks VRFKeyStore, lggr logger.Logger, httpClient, unrestrictedHTTPClient *http.Client) *runner

func ParseETHABIArgsString

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

func ResolveParam

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

func SelectGasLimit

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

Types

type AddressParam

type AddressParam common.Address

func (*AddressParam) UnmarshalPipelineParam

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

type AddressSliceParam

type AddressSliceParam []common.Address

func (*AddressSliceParam) UnmarshalPipelineParam

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

type AnyTask

type AnyTask struct {
	BaseTask `mapstructure:",squash"`
}

AnyTask picks a value at random from the set of non-errored inputs. If there are zero non-errored inputs then it returns an error.

func (*AnyTask) Run

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

func (*AnyTask) Type

func (t *AnyTask) Type() TaskType

type Base64DecodeTask

type Base64DecodeTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
}

Return types:

bytes

func (*Base64DecodeTask) Run

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

func (*Base64DecodeTask) Type

func (t *Base64DecodeTask) Type() TaskType

type Base64EncodeTask

type Base64EncodeTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
}

Return types:

string

func (*Base64EncodeTask) Run

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

func (*Base64EncodeTask) Type

func (t *Base64EncodeTask) Type() TaskType

type BaseTask

type BaseTask struct {
	Index     int32          `mapstructure:"index" json:"-" `
	Timeout   *time.Duration `mapstructure:"timeout"`
	FailEarly bool           `mapstructure:"failEarly"`

	Retries    null.Uint32   `mapstructure:"retries"`
	MinBackoff time.Duration `mapstructure:"minBackoff"`
	MaxBackoff time.Duration `mapstructure:"maxBackoff"`
	// contains filtered or unexported fields
}

func NewBaseTask

func NewBaseTask(id int, dotID string, inputs []TaskDependency, outputs []Task, index int32) BaseTask

func (*BaseTask) Base

func (t *BaseTask) Base() *BaseTask

func (BaseTask) DotID

func (t BaseTask) DotID() string

func (BaseTask) ID

func (t BaseTask) ID() int

func (BaseTask) Inputs

func (t BaseTask) Inputs() []TaskDependency

func (BaseTask) OutputIndex

func (t BaseTask) OutputIndex() int32

func (BaseTask) Outputs

func (t BaseTask) Outputs() []Task

func (BaseTask) TaskMaxBackoff

func (t BaseTask) TaskMaxBackoff() time.Duration

func (BaseTask) TaskMinBackoff

func (t BaseTask) TaskMinBackoff() time.Duration

func (BaseTask) TaskRetries

func (t BaseTask) TaskRetries() uint32

func (BaseTask) TaskTimeout

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

type BoolParam

type BoolParam bool

func (*BoolParam) UnmarshalPipelineParam

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

type BridgeTask

type BridgeTask struct {
	BaseTask `mapstructure:",squash"`

	Name              string `json:"name"`
	RequestData       string `json:"requestData"`
	IncludeInputAtKey string `json:"includeInputAtKey"`
	Async             string `json:"async"`
	CacheTTL          string `json:"cacheTTL"`
	Headers           string `json:"headers"`
	// contains filtered or unexported fields
}

Return types:

string

func (*BridgeTask) Run

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

func (*BridgeTask) Type

func (t *BridgeTask) Type() TaskType

type BytesParam

type BytesParam []byte

func (*BytesParam) UnmarshalPipelineParam

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

type CBORParseTask

type CBORParseTask struct {
	BaseTask `mapstructure:",squash"`
	Data     string `json:"data"`
	Mode     string `json:"mode"`
}

Return types:

map[string]interface{} with potential value types:
    float64
    string
    bool
    map[string]interface{}
    []interface{}
    nil

func (*CBORParseTask) Run

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

func (*CBORParseTask) Type

func (t *CBORParseTask) Type() TaskType

type ConditionalTask

type ConditionalTask struct {
	BaseTask `mapstructure:",squash"`
	Data     string `json:"data"`
}

ConditionalTask checks if data is false for now this is all we need but in the future we can expand this to handle more general conditional statements

func (*ConditionalTask) Run

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

func (*ConditionalTask) Type

func (t *ConditionalTask) Type() TaskType

type Config

type Config interface {
	BridgeResponseURL() *url.URL
	BridgeCacheTTL() time.Duration
	DatabaseURL() url.URL
	DefaultHTTPLimit() int64
	DefaultHTTPTimeout() models.Duration
	TriggerFallbackDBPollInterval() time.Duration
	JobPipelineMaxRunDuration() time.Duration
	JobPipelineReaperInterval() time.Duration
	JobPipelineReaperThreshold() time.Duration
}

type DecimalParam

type DecimalParam decimal.Decimal

func (DecimalParam) Decimal

func (d DecimalParam) Decimal() decimal.Decimal

func (*DecimalParam) UnmarshalPipelineParam

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

type DecimalSliceParam

type DecimalSliceParam []decimal.Decimal

func (*DecimalSliceParam) UnmarshalPipelineParam

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

type DivideTask

type DivideTask struct {
	BaseTask  `mapstructure:",squash"`
	Input     string `json:"input"`
	Divisor   string `json:"divisor"`
	Precision string `json:"precision"`
}

Return types:

*decimal.Decimal

func (*DivideTask) Run

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

func (*DivideTask) Type

func (t *DivideTask) Type() TaskType

type ETHABIDecodeLogTask

type ETHABIDecodeLogTask struct {
	BaseTask `mapstructure:",squash"`
	ABI      string `json:"abi"`
	Data     string `json:"data"`
	Topics   string `json:"topics"`
}

Return types:

map[string]interface{} with any geth/abigen value type

func (*ETHABIDecodeLogTask) Run

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

func (*ETHABIDecodeLogTask) Type

func (t *ETHABIDecodeLogTask) Type() TaskType

type ETHABIDecodeTask

type ETHABIDecodeTask struct {
	BaseTask `mapstructure:",squash"`
	ABI      string `json:"abi"`
	Data     string `json:"data"`
}

Return types:

map[string]interface{} with any geth/abigen value type

func (*ETHABIDecodeTask) Run

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

func (*ETHABIDecodeTask) Type

func (t *ETHABIDecodeTask) Type() TaskType

type ETHABIEncodeTask

type ETHABIEncodeTask struct {
	BaseTask `mapstructure:",squash"`
	ABI      string `json:"abi"`
	Data     string `json:"data"`
}

Return types:

[]byte

func (*ETHABIEncodeTask) Run

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

func (*ETHABIEncodeTask) Type

func (t *ETHABIEncodeTask) Type() TaskType

type ETHABIEncodeTask2

type ETHABIEncodeTask2 struct {
	BaseTask `mapstructure:",squash"`
	ABI      string `json:"abi"`
	Data     string `json:"data"`
}

Return types:

[]byte

func (*ETHABIEncodeTask2) Run

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

func (*ETHABIEncodeTask2) Type

func (t *ETHABIEncodeTask2) Type() TaskType

type ETHCallTask

type ETHCallTask struct {
	BaseTask            `mapstructure:",squash"`
	Contract            string `json:"contract"`
	From                string `json:"from"`
	Data                string `json:"data"`
	Gas                 string `json:"gas"`
	GasPrice            string `json:"gasPrice"`
	GasTipCap           string `json:"gasTipCap"`
	GasFeeCap           string `json:"gasFeeCap"`
	GasUnlimited        string `json:"gasUnlimited"`
	ExtractRevertReason bool   `json:"extractRevertReason"`
	EVMChainID          string `json:"evmChainID" mapstructure:"evmChainID"`
	// contains filtered or unexported fields
}

Return types:

[]byte

func (*ETHCallTask) Run

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

func (*ETHCallTask) Type

func (t *ETHCallTask) Type() TaskType

type ETHGetBlockTask

type ETHGetBlockTask struct {
	BaseTask   `mapstructure:",squash"`
	EVMChainID string `json:"evmChainID" mapstructure:"evmChainID"`
	// contains filtered or unexported fields
}

Return types:

map[string]interface{}

Fields: - number: int64 - hash: common.Hash - parentHash: common.Hash - timestamp: time.Time - baseFeePerGas: *big.Int - receiptsRoot: common.Hash - transactionsRoot: common.Hash - stateRoot: common.Hash

func (*ETHGetBlockTask) Run

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

func (*ETHGetBlockTask) Type

func (t *ETHGetBlockTask) Type() TaskType

type ETHKeyStore

type ETHKeyStore interface {
	GetRoundRobinAddress(chainID *big.Int, addrs ...common.Address) (common.Address, error)
}

type ETHTxTask

type ETHTxTask struct {
	BaseTask         `mapstructure:",squash"`
	From             string `json:"from"`
	To               string `json:"to"`
	Data             string `json:"data"`
	GasLimit         string `json:"gasLimit"`
	TxMeta           string `json:"txMeta"`
	MinConfirmations string `json:"minConfirmations"`
	// FailOnRevert, if set, will error the task if the transaction reverted on-chain
	// If unset, the receipt will be passed as output
	// It has no effect if minConfirmations == 0
	FailOnRevert    string `json:"failOnRevert"`
	EVMChainID      string `json:"evmChainID" mapstructure:"evmChainID"`
	TransmitChecker string `json:"transmitChecker"`
	// contains filtered or unexported fields
}

Return types:

nil

func (*ETHTxTask) Run

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

func (*ETHTxTask) Type

func (t *ETHTxTask) Type() TaskType

type ErrRunPanicked

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

When a task panics, we catch the panic and wrap it in an error for reporting to the scheduler.

func (ErrRunPanicked) Error

func (err ErrRunPanicked) Error() string

type EstimateGasLimitTask

type EstimateGasLimitTask struct {
	BaseTask   `mapstructure:",squash"`
	Input      string `json:"input"`
	From       string `json:"from"`
	To         string `json:"to"`
	Multiplier string `json:"multiplier"`
	Data       string `json:"data"`
	EVMChainID string `json:"evmChainID" mapstructure:"evmChainID"`
	// contains filtered or unexported fields
}

Return types:

uint64

func (*EstimateGasLimitTask) Run

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

func (*EstimateGasLimitTask) Type

func (t *EstimateGasLimitTask) Type() TaskType

type FailTask

type FailTask struct {
	BaseTask `mapstructure:",squash"`
	Msg      string
}

FailTask is like the Panic task but without all the drama and stack unwinding of a panic

func (*FailTask) Run

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

func (*FailTask) Type

func (t *FailTask) Type() TaskType

type FinalResult

type FinalResult struct {
	Values      []interface{}
	AllErrors   []error
	FatalErrors []error
}

FinalResult is the result of a Run

func (FinalResult) CombinedError

func (result FinalResult) CombinedError() error

func (FinalResult) HasErrors

func (result FinalResult) HasErrors() bool

HasErrors returns true if the final result has any errors

func (FinalResult) HasFatalErrors

func (result FinalResult) HasFatalErrors() bool

HasFatalErrors returns true if the final result has any errors

func (FinalResult) SingularResult

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

SingularResult returns a single result if the FinalResult only has one set of outputs/errors

type GasEstimator

type GasEstimator interface {
	EstimateGas(ctx context.Context, call ethereum.CallMsg) (uint64, error)
}

type GetterFunc

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

GetterFunc is a function that either returns a value or an error.

func From

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

From creates []GetterFunc from a mix of getters or bare values.

func Input

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

Input creates a getter returning inputs[index] value, or error if index is out of range.

func Inputs

func Inputs(inputs []Result) GetterFunc

Inputs creates a getter returning array of Result.Value (or Result.Error where not nil).

func JSONWithVarExprs

func JSONWithVarExprs(jsExpr string, vars Vars, allowErrors bool) GetterFunc

JSONWithVarExprs creates a getter that unmarshals jsExpr string as JSON, and interpolates all variables expressions found in jsExpr from Vars. The getter returns the unmarshalled object having expressions interpolated from Vars. allowErrors flag indicates if interpolating values stored in Vars can be errors. jsExpr example: {"requestId": $(decode_log.requestId), "payment": $(decode_log.payment)}

func NonemptyString

func NonemptyString(s string) GetterFunc

NonemptyString creates a getter to ensure the string is non-empty.

func ValidDurationInSeconds

func ValidDurationInSeconds(s string) GetterFunc

ValidDurationInSeconds creates a getter to ensure the string is a valid duration and return duration in seconds.

func VarExpr

func VarExpr(expr string, vars Vars) GetterFunc

VarExpr creates a getter interpolating expr value using the given Vars. The expression allows whitespace on both ends that will be trimmed. Expr examples: $(foo.bar), $(arr.1), $(bar)

type Graph

type Graph struct {
	*simple.DirectedGraph
}

tree fulfills the graph.DirectedGraph interface, which makes it possible for us to `dot.Unmarshal(...)` a DOT string directly into it.

func NewGraph

func NewGraph() *Graph

func (*Graph) AddImplicitDependenciesAsEdges

func (g *Graph) AddImplicitDependenciesAsEdges()

Looks at node attributes and searches for implicit dependencies on other nodes expressed as attribute values. Adds those dependencies as implicit edges in the graph.

func (*Graph) IsImplicitEdge

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

Indicates whether there's an implicit edge from uid -> vid. Implicit edged are ones that weren't added via the TOML spec, but via the pipeline parsing code

func (*Graph) NewEdge

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

func (*Graph) NewNode

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

func (*Graph) UnmarshalText

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

type GraphEdge

type GraphEdge struct {
	graph.Edge
	// contains filtered or unexported fields
}

func (*GraphEdge) IsImplicit

func (e *GraphEdge) IsImplicit() bool

func (*GraphEdge) SetIsImplicit

func (e *GraphEdge) SetIsImplicit(isImplicit bool)

type GraphNode

type GraphNode struct {
	graph.Node
	// contains filtered or unexported fields
}

func (*GraphNode) Attributes

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

func (*GraphNode) DOTID

func (n *GraphNode) DOTID() string

func (*GraphNode) SetAttribute

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

func (*GraphNode) SetDOTID

func (n *GraphNode) SetDOTID(id string)

func (*GraphNode) String

func (n *GraphNode) String() string

type HTTPTask

type HTTPTask struct {
	BaseTask                       `mapstructure:",squash"`
	Method                         string
	URL                            string
	RequestData                    string `json:"requestData"`
	AllowUnrestrictedNetworkAccess string
	Headers                        string
	// contains filtered or unexported fields
}

Return types:

string

func (*HTTPTask) Run

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

func (*HTTPTask) Type

func (t *HTTPTask) Type() TaskType

type HashSliceParam

type HashSliceParam []common.Hash

func (*HashSliceParam) UnmarshalPipelineParam

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

type HexDecodeTask

type HexDecodeTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
}

Return types:

bytes

func (*HexDecodeTask) Run

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

func (*HexDecodeTask) Type

func (t *HexDecodeTask) Type() TaskType

type HexEncodeTask

type HexEncodeTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
}

Return types:

string

func (*HexEncodeTask) Run

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

func (*HexEncodeTask) Type

func (t *HexEncodeTask) Type() TaskType

type JSONParseTask

type JSONParseTask struct {
	BaseTask  `mapstructure:",squash"`
	Path      string `json:"path"`
	Separator string `json:"separator"`
	Data      string `json:"data"`
	// Lax when disabled will return an error if the path does not exist
	// Lax when enabled will return nil with no error if the path does not exist
	Lax string
}

Return types:

float64
string
bool
map[string]interface{}
[]interface{}
nil

func (*JSONParseTask) Run

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

func (*JSONParseTask) Type

func (t *JSONParseTask) Type() TaskType

type JSONPathParam

type JSONPathParam []string

func NewJSONPathParam

func NewJSONPathParam(sep string) JSONPathParam

NewJSONPathParam returns a new JSONPathParam using the given separator, or the default if empty.

func (*JSONPathParam) UnmarshalPipelineParam

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

UnmarshalPipelineParam unmarshals a slice of strings from val. If val is a string or []byte, it is split on a separator. The default separator is ',' but can be overridden by initializing via NewJSONPathParam.

type JSONSerializable

type JSONSerializable struct {
	Val   interface{}
	Valid bool
}

func (*JSONSerializable) Empty

func (js *JSONSerializable) Empty() bool

func (JSONSerializable) MarshalJSON

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

MarshalJSON implements custom marshaling logic

func (*JSONSerializable) Scan

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

func (*JSONSerializable) UnmarshalJSON

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

UnmarshalJSON implements custom unmarshaling logic

func (JSONSerializable) Value

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

type Keypath

type Keypath struct {
	Parts []string
}

Keypath contains keypath parsed by NewKeypathFromString.

func NewKeypathFromString

func NewKeypathFromString(keypathStr string) (Keypath, error)

NewKeypathFromString creates a new Keypath from the given string. Returns error if it fails to parse the given keypath string.

type LengthTask

type LengthTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
}

Return types:

*decimal.Decimal

func (*LengthTask) Run

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

func (*LengthTask) Type

func (t *LengthTask) Type() TaskType

type LessThanTask

type LessThanTask struct {
	BaseTask `mapstructure:",squash"`
	Left     string `json:"input"`
	Right    string `json:"limit"`
}

Return types:

bool

func (*LessThanTask) Run

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

func (*LessThanTask) Type

func (t *LessThanTask) Type() TaskType

type LookupTask

type LookupTask struct {
	BaseTask `mapstructure:",squash"`
	Key      string `json:"key"`
}

Look up a field on a map

Return types:

interface{}

func (*LookupTask) Run

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

func (*LookupTask) Type

func (t *LookupTask) Type() TaskType

type LowercaseTask

type LowercaseTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
}

Return types:

string

func (*LowercaseTask) Run

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

func (*LowercaseTask) Type

func (t *LowercaseTask) Type() TaskType

type MapParam

type MapParam map[string]interface{}

MapParam accepts maps or JSON-encoded strings

func (MapParam) Map

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

func (*MapParam) UnmarshalPipelineParam

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

type MaybeBigIntParam

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

func NewMaybeBigIntParam

func NewMaybeBigIntParam(n *big.Int) MaybeBigIntParam

NewMaybeBigIntParam creates a new instance of MaybeBigIntParam

func (MaybeBigIntParam) BigInt

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

func (*MaybeBigIntParam) UnmarshalPipelineParam

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

type MaybeInt32Param

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

func NewMaybeInt32Param

func NewMaybeInt32Param(n int32, isSet bool) MaybeInt32Param

NewMaybeInt32Param creates new instance of MaybeInt32Param

func (MaybeInt32Param) Int32

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

func (*MaybeInt32Param) UnmarshalPipelineParam

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

type MaybeUint64Param

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

func NewMaybeUint64Param

func NewMaybeUint64Param(n uint64, isSet bool) MaybeUint64Param

NewMaybeUint64Param creates new instance of MaybeUint64Param

func (MaybeUint64Param) Uint64

func (p MaybeUint64Param) Uint64() (uint64, bool)

func (*MaybeUint64Param) UnmarshalPipelineParam

func (p *MaybeUint64Param) UnmarshalPipelineParam(val interface{}) error

type MeanTask

type MeanTask struct {
	BaseTask      `mapstructure:",squash"`
	Values        string `json:"values"`
	AllowedFaults string `json:"allowedFaults"`
	Precision     string `json:"precision"`
}

Return types:

*decimal.Decimal

func (*MeanTask) Run

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

func (*MeanTask) Type

func (t *MeanTask) Type() TaskType

type MedianTask

type MedianTask struct {
	BaseTask      `mapstructure:",squash"`
	Values        string `json:"values"`
	AllowedFaults string `json:"allowedFaults"`
}

Return types:

*decimal.Decimal

func (*MedianTask) Run

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

func (*MedianTask) Type

func (t *MedianTask) Type() TaskType

type MemoTask

type MemoTask struct {
	BaseTask `mapstructure:",squash"`
	Value    string `json:"value"`
}

func (*MemoTask) Run

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

func (*MemoTask) Type

func (t *MemoTask) Type() TaskType

type MergeTask

type MergeTask struct {
	BaseTask `mapstructure:",squash"`
	Left     string `json:"left"`
	Right    string `json:"right"`
}

Return types:

map[string]interface{}

func (*MergeTask) Run

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

func (*MergeTask) Type

func (t *MergeTask) Type() TaskType

type Method

type Method struct {
	Name   string
	Inputs abi.Arguments
}

go-ethereum's abi.Method doesn't implement json.Marshal for Type, but otherwise would have worked fine, in any case we only care about these...

type ModeTask

type ModeTask struct {
	BaseTask      `mapstructure:",squash"`
	Values        string `json:"values"`
	AllowedFaults string `json:"allowedFaults"`
}

Return types:

map[string]interface{}{
    "results": []interface{} containing any other type other pipeline tasks can return
    "occurrences": (int64)
}

func (*ModeTask) Run

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

func (*ModeTask) Type

func (t *ModeTask) Type() TaskType

type MultiplyTask

type MultiplyTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
	Times    string `json:"times"`
}

Return types:

*decimal.Decimal

func (*MultiplyTask) Run

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

func (*MultiplyTask) Type

func (t *MultiplyTask) Type() TaskType

type ORM

type ORM interface {
	services.ServiceCtx
	CreateSpec(pipeline Pipeline, maxTaskTimeout models.Interval, qopts ...pg.QOpt) (int32, error)
	CreateRun(run *Run, qopts ...pg.QOpt) (err error)
	InsertRun(run *Run, qopts ...pg.QOpt) error
	DeleteRun(id int64) error
	StoreRun(run *Run, qopts ...pg.QOpt) (restart bool, err error)
	UpdateTaskRunResult(taskID uuid.UUID, result Result) (run Run, start bool, err error)
	InsertFinishedRun(run *Run, saveSuccessfulTaskRuns bool, qopts ...pg.QOpt) (err error)

	// InsertFinishedRuns inserts all the given runs into the database.
	// If saveSuccessfulTaskRuns is false, only errored runs are saved.
	InsertFinishedRuns(run []*Run, saveSuccessfulTaskRuns bool, qopts ...pg.QOpt) (err error)

	DeleteRunsOlderThan(context.Context, time.Duration) error
	FindRun(id int64) (Run, error)
	GetAllRuns() ([]Run, error)
	GetUnfinishedRuns(context.Context, time.Time, func(run Run) error) error
	GetQ() pg.Q
}

type ORMConfig

type ORMConfig interface {
	pg.QConfig
	JobPipelineMaxSuccessfulRuns() uint64
}

type ObjectParam

type ObjectParam struct {
	Type         ObjectType
	BoolValue    BoolParam
	DecimalValue DecimalParam
	StringValue  StringParam
	SliceValue   SliceParam
	MapValue     MapParam
}

ObjectParam represents a kind of any type that could be used by the memo task

func (ObjectParam) Marshal

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

func (ObjectParam) MarshalJSON

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

func (ObjectParam) String

func (o ObjectParam) String() string

func (*ObjectParam) UnmarshalPipelineParam

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

type ObjectType

type ObjectType int
const (
	NilType ObjectType = iota
	BoolType
	DecimalType
	StringType
	SliceType
	MapType
)

type PanicTask

type PanicTask struct {
	BaseTask `mapstructure:",squash"`
	Msg      string
}

func (*PanicTask) Run

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

func (*PanicTask) Type

func (t *PanicTask) Type() TaskType

type Pipeline

type Pipeline struct {
	Tasks []Task

	Source string
	// contains filtered or unexported fields
}

func Parse

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

func (*Pipeline) ByDotID

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

func (*Pipeline) MinTimeout

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

func (*Pipeline) RequiresPreInsert

func (p *Pipeline) RequiresPreInsert() bool

func (*Pipeline) UnmarshalText

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

type PipelineParamUnmarshaler

type PipelineParamUnmarshaler interface {
	UnmarshalPipelineParam(val interface{}) error
}

type PossibleErrorResponses

type PossibleErrorResponses struct {
	Error        string `json:"error"`
	ErrorMessage string `json:"errorMessage"`
}

type Result

type Result struct {
	Value interface{}
	Error error
}

Result is the result of a TaskRun

func (Result) ErrorDB

func (result Result) ErrorDB() null.String

ErrorDB dumps a single result error for a pipeline_task_run

func (Result) OutputDB

func (result Result) OutputDB() JSONSerializable

OutputDB dumps a single result output for a pipeline_run or pipeline_task_run

type ResumeRequest

type ResumeRequest struct {
	Error null.String     `json:"error"`
	Value json.RawMessage `json:"value"`
}

func (ResumeRequest) ToResult

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

type Run

type Run struct {
	ID             int64            `json:"-"`
	PipelineSpecID int32            `json:"-"`
	PipelineSpec   Spec             `json:"pipelineSpec"`
	Meta           JSONSerializable `json:"meta"`
	// The errors are only ever strings
	// DB example: [null, null, "my error"]
	AllErrors   RunErrors        `json:"all_errors"`
	FatalErrors RunErrors        `json:"fatal_errors"`
	Inputs      JSONSerializable `json:"inputs"`
	// Its expected that Output.Val is of type []interface{}.
	// DB example: [1234, {"a": 10}, null]
	Outputs          JSONSerializable `json:"outputs"`
	CreatedAt        time.Time        `json:"createdAt"`
	FinishedAt       null.Time        `json:"finishedAt"`
	PipelineTaskRuns []TaskRun        `json:"taskRuns"`
	State            RunStatus        `json:"state"`

	Pending bool
	// FailSilently is used to signal that a task with the failEarly flag has failed, and we want to not put this in the db
	FailSilently bool
}

func NewRun

func NewRun(spec Spec, vars Vars) Run

func (*Run) ByDotID

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

func (Run) GetID

func (r Run) GetID() string

func (Run) HasErrors

func (r Run) HasErrors() bool

func (Run) HasFatalErrors

func (r Run) HasFatalErrors() bool

func (*Run) SetID

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

func (*Run) Status

func (r *Run) Status() RunStatus

Status determines the status of the run.

func (*Run) StringAllErrors

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

func (*Run) StringFatalErrors

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

func (*Run) StringOutputs

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

type RunErrors

type RunErrors []null.String

func (RunErrors) HasError

func (re RunErrors) HasError() bool

func (*RunErrors) Scan

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

func (RunErrors) ToError

func (re RunErrors) ToError() error

ToError coalesces all non-nil errors into a single error object. This is useful for logging.

func (RunErrors) Value

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

type RunInfo

type RunInfo struct {
	IsRetryable bool
	IsPending   bool
}

RunInfo contains additional information about the finished TaskRun

type RunStatus

type RunStatus string

RunStatus represents the status of a run

const (
	// RunStatusUnknown is the when the run status cannot be determined.
	RunStatusUnknown RunStatus = "unknown"
	// RunStatusRunning is used for when a run is actively being executed.
	RunStatusRunning RunStatus = "running"
	// RunStatusSuspended is used when a run is paused and awaiting further results.
	RunStatusSuspended RunStatus = "suspended"
	// RunStatusErrored is used for when a run has errored and will not complete.
	RunStatusErrored RunStatus = "errored"
	// RunStatusCompleted is used for when a run has successfully completed execution.
	RunStatusCompleted RunStatus = "completed"
)

func (RunStatus) Completed

func (s RunStatus) Completed() bool

Completed returns true if the status is RunStatusCompleted.

func (RunStatus) Errored

func (s RunStatus) Errored() bool

Errored returns true if the status is RunStatusErrored.

func (RunStatus) Finished

func (s RunStatus) Finished() bool

Finished returns true if the status is final and can't be changed.

type Runner

type Runner interface {
	services.ServiceCtx

	// Run is a blocking call that will execute the run until no further progress can be made.
	// If `incomplete` is true, the run is only partially complete and is suspended, awaiting to be resumed when more data comes in.
	// Note that `saveSuccessfulTaskRuns` value is ignored if the run contains async tasks.
	Run(ctx context.Context, run *Run, l logger.Logger, saveSuccessfulTaskRuns bool, fn func(tx pg.Queryer) error) (incomplete bool, err error)
	ResumeRun(taskID uuid.UUID, value interface{}, err error) error

	// ExecuteRun executes a new run in-memory according to a spec and returns the results.
	// We expect spec.JobID and spec.JobName to be set for logging/prometheus.
	ExecuteRun(ctx context.Context, spec Spec, vars Vars, l logger.Logger) (run Run, trrs TaskRunResults, err error)
	// InsertFinishedRun saves the run results in the database.
	InsertFinishedRun(run *Run, saveSuccessfulTaskRuns bool, qopts ...pg.QOpt) error
	InsertFinishedRuns(runs []*Run, saveSuccessfulTaskRuns bool, qopts ...pg.QOpt) error

	// ExecuteAndInsertFinishedRun executes a new run in-memory according to a spec, persists and saves the results.
	// It is a combination of ExecuteRun and InsertFinishedRun.
	// Note that the spec MUST have a DOT graph for this to work.
	ExecuteAndInsertFinishedRun(ctx context.Context, spec Spec, vars Vars, l logger.Logger, saveSuccessfulTaskRuns bool) (runID int64, finalResult FinalResult, err error)

	OnRunFinished(func(*Run))
}

type SliceParam

type SliceParam []interface{}

func (SliceParam) FilterErrors

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

func (*SliceParam) UnmarshalPipelineParam

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

type Spec

type Spec struct {
	ID                int32
	DotDagSource      string          `json:"dotDagSource"`
	CreatedAt         time.Time       `json:"-"`
	MaxTaskDuration   models.Interval `json:"-"`
	GasLimit          *uint32         `json:"-"`
	ForwardingAllowed bool            `json:"-"`

	JobID   int32  `json:"-"`
	JobName string `json:"-"`
	JobType string `json:"-"`
}

func (Spec) Pipeline

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

type StringParam

type StringParam string

func (*StringParam) String

func (s *StringParam) String() string

func (*StringParam) UnmarshalPipelineParam

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

type StringSliceParam

type StringSliceParam []string

func (*StringSliceParam) UnmarshalPipelineParam

func (s *StringSliceParam) UnmarshalPipelineParam(val interface{}) error

type SumTask

type SumTask struct {
	BaseTask      `mapstructure:",squash"`
	Values        string `json:"values"`
	AllowedFaults string `json:"allowedFaults"`
}

Return types:

*decimal.Decimal

func (*SumTask) Run

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

func (*SumTask) Type

func (t *SumTask) Type() TaskType

type Task

type Task interface {
	Type() TaskType
	ID() int
	DotID() string
	Run(ctx context.Context, lggr logger.Logger, vars Vars, inputs []Result) (Result, RunInfo)
	Base() *BaseTask
	Outputs() []Task
	Inputs() []TaskDependency
	OutputIndex() int32
	TaskTimeout() (time.Duration, bool)
	TaskRetries() uint32
	TaskMinBackoff() time.Duration
	TaskMaxBackoff() time.Duration
}

func UnmarshalTaskFromMap

func UnmarshalTaskFromMap(taskType TaskType, taskMap interface{}, ID int, dotID string) (_ Task, err error)

type TaskDependency

type TaskDependency struct {
	PropagateResult bool
	InputTask       Task
}

Wraps the input Task for the given dependent task along with a bool variable PropagateResult, which Indicates whether result of InputTask should be propagated to its dependent task. If the edge between these tasks was an implicit edge, then results are not propagated. This is because some tasks cannot handle an input from an edge which wasn't specified in the spec.

type TaskRun

type TaskRun struct {
	ID            uuid.UUID        `json:"id"`
	Type          TaskType         `json:"type"`
	PipelineRun   Run              `json:"-"`
	PipelineRunID int64            `json:"-"`
	Output        JSONSerializable `json:"output"`
	Error         null.String      `json:"error"`
	CreatedAt     time.Time        `json:"createdAt"`
	FinishedAt    null.Time        `json:"finishedAt"`
	Index         int32            `json:"index"`
	DotID         string           `json:"dotId"`
	// contains filtered or unexported fields
}

func (TaskRun) GetDotID

func (tr TaskRun) GetDotID() string

func (TaskRun) GetID

func (tr TaskRun) GetID() string

func (*TaskRun) IsPending

func (tr *TaskRun) IsPending() bool

func (TaskRun) Result

func (tr TaskRun) Result() Result

func (*TaskRun) SetID

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

type TaskRunResult

type TaskRunResult struct {
	ID         uuid.UUID
	Task       Task
	TaskRun    TaskRun
	Result     Result
	Attempts   uint
	CreatedAt  time.Time
	FinishedAt null.Time
	// contains filtered or unexported fields
}

TaskRunResult describes the result of a task run, suitable for database update or insert. ID might be zero if the TaskRun has not been inserted yet TaskSpecID will always be non-zero

func (*TaskRunResult) IsPending

func (result *TaskRunResult) IsPending() bool

func (*TaskRunResult) IsTerminal

func (result *TaskRunResult) IsTerminal() bool

type TaskRunResults

type TaskRunResults []TaskRunResult

TaskRunResults represents a collection of results for all task runs for one pipeline run

func (TaskRunResults) FinalResult

func (trrs TaskRunResults) FinalResult(l logger.Logger) FinalResult

FinalResult pulls the FinalResult for the pipeline_run from the task runs It needs to respect the output index of each task

func (*TaskRunResults) GetNextTaskOf

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

GetNextTaskOf returns the task with the next id or nil if it does not exist

type TaskType

type TaskType string
const (
	TaskTypeAny              TaskType = "any"
	TaskTypeBase64Decode     TaskType = "base64decode"
	TaskTypeBase64Encode     TaskType = "base64encode"
	TaskTypeBridge           TaskType = "bridge"
	TaskTypeCBORParse        TaskType = "cborparse"
	TaskTypeConditional      TaskType = "conditional"
	TaskTypeDivide           TaskType = "divide"
	TaskTypeETHABIDecode     TaskType = "ethabidecode"
	TaskTypeETHABIDecodeLog  TaskType = "ethabidecodelog"
	TaskTypeETHABIEncode     TaskType = "ethabiencode"
	TaskTypeETHABIEncode2    TaskType = "ethabiencode2"
	TaskTypeETHCall          TaskType = "ethcall"
	TaskTypeETHGetBlock      TaskType = "ethgetblock"
	TaskTypeETHTx            TaskType = "ethtx"
	TaskTypeEstimateGasLimit TaskType = "estimategaslimit"
	TaskTypeHTTP             TaskType = "http"
	TaskTypeHexDecode        TaskType = "hexdecode"
	TaskTypeHexEncode        TaskType = "hexencode"
	TaskTypeJSONParse        TaskType = "jsonparse"
	TaskTypeLength           TaskType = "length"
	TaskTypeLessThan         TaskType = "lessthan"
	TaskTypeLookup           TaskType = "lookup"
	TaskTypeLowercase        TaskType = "lowercase"
	TaskTypeMean             TaskType = "mean"
	TaskTypeMedian           TaskType = "median"
	TaskTypeMerge            TaskType = "merge"
	TaskTypeMode             TaskType = "mode"
	TaskTypeMultiply         TaskType = "multiply"
	TaskTypeSum              TaskType = "sum"
	TaskTypeUppercase        TaskType = "uppercase"
	TaskTypeVRF              TaskType = "vrf"
	TaskTypeVRFV2            TaskType = "vrfv2"

	// Testing only.
	TaskTypePanic TaskType = "panic"
	TaskTypeMemo  TaskType = "memo"
	TaskTypeFail  TaskType = "fail"
)

func (TaskType) String

func (t TaskType) String() string

type URLParam

type URLParam url.URL

func (*URLParam) String

func (u *URLParam) String() string

func (*URLParam) UnmarshalPipelineParam

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

type Uint64Param

type Uint64Param uint64

func (*Uint64Param) UnmarshalPipelineParam

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

type UppercaseTask

type UppercaseTask struct {
	BaseTask `mapstructure:",squash"`
	Input    string `json:"input"`
}

Return types:

string

func (*UppercaseTask) Run

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

func (*UppercaseTask) Type

func (t *UppercaseTask) Type() TaskType

type VRFKeyStore

type VRFKeyStore interface {
	GenerateProof(id string, seed *big.Int) (vrfkey.Proof, error)
}

type VRFTask

type VRFTask struct {
	BaseTask           `mapstructure:",squash"`
	PublicKey          string `json:"publicKey"`
	RequestBlockHash   string `json:"requestBlockHash"`
	RequestBlockNumber string `json:"requestBlockNumber"`
	Topics             string `json:"topics"`
	// contains filtered or unexported fields
}

func (*VRFTask) Run

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

func (*VRFTask) Type

func (t *VRFTask) Type() TaskType

type VRFTaskV2

type VRFTaskV2 struct {
	BaseTask           `mapstructure:",squash"`
	PublicKey          string `json:"publicKey"`
	RequestBlockHash   string `json:"requestBlockHash"`
	RequestBlockNumber string `json:"requestBlockNumber"`
	Topics             string `json:"topics"`
	// contains filtered or unexported fields
}

func (*VRFTaskV2) Run

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

func (*VRFTaskV2) Type

func (t *VRFTaskV2) Type() TaskType

type Vars

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

func NewVarsFrom

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

NewVarsFrom creates new Vars from the given map. If the map is nil, a new map instance will be created.

func (Vars) Copy

func (vars Vars) Copy() Vars

Copy makes a copy of Vars by copying the underlying map. Used by scheduler for new tasks to avoid data races.

func (Vars) Get

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

Get returns the value for the given keypath or error. The keypath can consist of one or more parts, e.g. "foo" or "foo.6.a.b". Every part except for the first one can be an index of a slice.

func (Vars) Set

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

Set sets a top-level variable specified by dotID. Returns error if either dotID is empty or it is a compound keypath.

Directories

Path Synopsis

Jump to

Keyboard shortcuts

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