runner

Documentation

Index

• type ActionData
• type ActionResult
• type ActionWorkRequest
• type ActionWorker
  • func NewWorker(id int, runner *DirectRunner, workerQueue chan chan ActionWorkRequest) ActionWorker
  • func (w ActionWorker) Start()
  • func (w ActionWorker) Stop()
• type AsyncResultHandler
  • func (rh *AsyncResultHandler) Done()
  • func (rh *AsyncResultHandler) HandleResult(results map[string]*data.Attribute, err error)
• type DirectRunner
  • func NewDirect() *DirectRunner
  • func (runner *DirectRunner) Execute(ctx context.Context, act action.Action, inputs map[string]*data.Attribute) (results map[string]*data.Attribute, err error)
  • func (runner *DirectRunner) Run(ctx context.Context, act action.Action, uri string, options interface{}) (code int, data interface{}, err error)
  • func (runner *DirectRunner) RunAction(ctx context.Context, act action.Action, options map[string]interface{}) (results map[string]*data.Attribute, err error)
  • func (runner *DirectRunner) Start() error
  • func (runner *DirectRunner) Stop() error
• type PooledConfig
• type PooledRunner
  • func NewPooled(config *PooledConfig) *PooledRunner
  • func (runner *PooledRunner) Execute(ctx context.Context, act action.Action, inputs map[string]*data.Attribute) (results map[string]*data.Attribute, err error)
  • func (runner *PooledRunner) Run(ctx context.Context, act action.Action, uri string, options interface{}) (code int, data interface{}, err error)
  • func (runner *PooledRunner) RunAction(ctx context.Context, act action.Action, options map[string]interface{}) (results map[string]*data.Attribute, err error)
  • func (runner *PooledRunner) Start() error
  • func (runner *PooledRunner) Stop() error
• type RequestType
• type SyncResultHandler
  • func (rh *SyncResultHandler) Done()
  • func (rh *SyncResultHandler) HandleResult(resultData map[string]*data.Attribute, err error)
  • func (rh *SyncResultHandler) Result() (resultData map[string]*data.Attribute, err error)

Types

type ActionData

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

ActionData action related data to pass along in a ActionWorkRequest

type ActionResult

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

ActionResult is a simple struct to hold the results for an Action

type ActionWorkRequest

type ActionWorkRequest struct {
	ReqType RequestType
	ID      string
	// contains filtered or unexported fields
}

ActionWorkRequest describes a Request that Worker should handle

type ActionWorker

type ActionWorker struct {
	ID int

	Work        chan ActionWorkRequest
	WorkerQueue chan chan ActionWorkRequest
	QuitChan    chan bool
	// contains filtered or unexported fields
}

A ActionWorker handles WorkRequest, work requests consist of start, restart and resume of FlowInstances

func NewWorker

func NewWorker(id int, runner *DirectRunner, workerQueue chan chan ActionWorkRequest) ActionWorker

NewWorker creates, and returns a new Worker object. Its only argument is a channel that the worker can add itself to whenever it is done its work.

func (ActionWorker) Start

func (w ActionWorker) Start()

Start function "starts" the worker by starting a goroutine, that is an infinite "for-select" loop. This is where all the request are handled

func (ActionWorker) Stop

func (w ActionWorker) Stop()

Stop tells the worker to stop listening for work requests.

Note that the worker will only stop *after* it has finished its work.

type AsyncResultHandler

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

AsyncResultHandler simple ResultHandler to use in the asynchronous case

func (*AsyncResultHandler) Done

func (rh *AsyncResultHandler) Done()

Done implements action.ResultHandler.Done

func (*AsyncResultHandler) HandleResult

func (rh *AsyncResultHandler) HandleResult(results map[string]*data.Attribute, err error)

HandleResult implements action.ResultHandler.HandleResult

type DirectRunner

type DirectRunner struct {
}

DirectRunner runs an action synchronously

func NewDirect

func NewDirect() *DirectRunner

NewDirectRunner create a new DirectRunner

func (*DirectRunner) Execute

func (runner *DirectRunner) Execute(ctx context.Context, act action.Action, inputs map[string]*data.Attribute) (results map[string]*data.Attribute, err error)

Execute implements action.Runner.Execute

func (*DirectRunner) Run

func (runner *DirectRunner) Run(ctx context.Context, act action.Action, uri string, options interface{}) (code int, data interface{}, err error)

Deprecated: Use Execute() instead

func (*DirectRunner) RunAction

func (runner *DirectRunner) RunAction(ctx context.Context, act action.Action, options map[string]interface{}) (results map[string]*data.Attribute, err error)

Deprecated: Use Execute() instead

func (*DirectRunner) Start

func (runner *DirectRunner) Start() error

Start will start the engine, by starting all of its workers

func (*DirectRunner) Stop

func (runner *DirectRunner) Stop() error

Stop will stop the engine, by stopping all of its workers

type PooledConfig

type PooledConfig struct {
	NumWorkers    int `json:"numWorkers"`
	WorkQueueSize int `json:"workQueueSize"`
}

PooledConfig is the configuration object for a PooledRunner

type PooledRunner

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

PooledRunner is a action runner that queues and runs a action in a worker pool

func NewPooled

func NewPooled(config *PooledConfig) *PooledRunner

NewPooledRunner create a new pooled

func (*PooledRunner) Execute

func (runner *PooledRunner) Execute(ctx context.Context, act action.Action, inputs map[string]*data.Attribute) (results map[string]*data.Attribute, err error)

Execute implements action.Runner.Execute

func (*PooledRunner) Run

func (runner *PooledRunner) Run(ctx context.Context, act action.Action, uri string, options interface{}) (code int, data interface{}, err error)

Deprecated: Use Execute() instead

func (*PooledRunner) RunAction

func (runner *PooledRunner) RunAction(ctx context.Context, act action.Action, options map[string]interface{}) (results map[string]*data.Attribute, err error)

Deprecated: Use Execute() instead

func (*PooledRunner) Start

func (runner *PooledRunner) Start() error

Start will start the engine, by starting all of its workers

func (*PooledRunner) Stop

func (runner *PooledRunner) Stop() error

Stop will stop the engine, by stopping all of its workers

type RequestType

type RequestType int

RequestType is value that indicates the type of Request

const (
	// RtRun denotes a run action request
	RtRun RequestType = 10
)

type SyncResultHandler

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

SyncResultHandler simple result handler to use in synchronous case

func (*SyncResultHandler) Done

func (rh *SyncResultHandler) Done()

Done implements action.ResultHandler.Done

func (*SyncResultHandler) HandleResult

func (rh *SyncResultHandler) HandleResult(resultData map[string]*data.Attribute, err error)

HandleResult implements action.ResultHandler.HandleResult

func (*SyncResultHandler) Result

func (rh *SyncResultHandler) Result() (resultData map[string]*data.Attribute, err error)

Result returns the latest Result set on the handler