runtime

Documentation

Index

• Constants
• type FlowDefinitionHandler
• type FlowExecutor
  • func (fe *FlowExecutor) Configure(requestID string)
  • func (fe *FlowExecutor) GetDataStore() (sdk.DataStore, error)
  • func (fe *FlowExecutor) GetEventHandler() (sdk.EventHandler, error)
  • func (fe *FlowExecutor) GetExecutionOption(_ sdk.Operation) map[string]interface{}
  • func (fe *FlowExecutor) GetFlowDefinition(pipeline *sdk.Pipeline, context *sdk.Context) error
  • func (fe *FlowExecutor) GetFlowName() string
  • func (fe *FlowExecutor) GetLogger() (sdk.Logger, error)
  • func (fe *FlowExecutor) GetReqAuthKey() (string, error)
  • func (fe *FlowExecutor) GetStateStore() (sdk.StateStore, error)
  • func (fe *FlowExecutor) GetValidationKey() (string, error)
  • func (fe *FlowExecutor) HandleExecutionCompletion(data []byte) error
  • func (fe *FlowExecutor) HandleNextNode(partial *executor.PartialState) error
  • func (fe *FlowExecutor) Init(request *runtime.Request) error
  • func (fe *FlowExecutor) LoggingEnabled() bool
  • func (fe *FlowExecutor) MonitoringEnabled() bool
  • func (fe *FlowExecutor) ReqAuthEnabled() bool
  • func (fe *FlowExecutor) ReqValidationEnabled() bool
• type FlowRuntime
  • func (fRuntime *FlowRuntime) Consume(message rmq.Delivery)
  • func (fRuntime *FlowRuntime) CreateExecutor(req *runtime.Request) (executor.Executor, error)
  • func (fRuntime *FlowRuntime) EnqueuePartialRequest(pr *runtime.Request) error
  • func (fRuntime *FlowRuntime) Execute(flowName string, request *runtime.Request) error
  • func (fRuntime *FlowRuntime) Init() error
  • func (fRuntime *FlowRuntime) Pause(flowName string, request *runtime.Request) error
  • func (fRuntime *FlowRuntime) Resume(flowName string, request *runtime.Request) error
  • func (fRuntime *FlowRuntime) StartQueueWorker(errorChan chan error) error
  • func (fRuntime *FlowRuntime) StartRuntime() error
  • func (fRuntime *FlowRuntime) StartServer() error
  • func (fRuntime *FlowRuntime) Stop(flowName string, request *runtime.Request) error
  • func (fRuntime *FlowRuntime) StopServer() error
• type Task
• type Worker

Constants

const (
	InternalRequestQueueInitial = "goflow-internal-request"
	FlowKeyInitial              = "goflow-flow"
	WorkerKeyInitial            = "goflow-worker"

	GoFlowRegisterInterval = 4
	RDBKeyTimeOut          = 10

	PartialRequest = "PARTIAL"
	NewRequest     = "NEW"
	PauseRequest   = "PAUSE"
	ResumeRequest  = "RESUME"
	StopRequest    = "STOP"
)

Types

type FlowDefinitionHandler

type FlowDefinitionHandler func(flow *v1.Workflow, context *v1.Context) error

type FlowExecutor

type FlowExecutor struct {
	CallbackURL             string // the callback url
	RequestAuthSharedSecret string
	RequestAuthEnabled      bool
	EnableMonitoring        bool
	IsLoggingEnabled        bool

	StateStore   sdk.StateStore
	DataStore    sdk.DataStore
	EventHandler sdk.EventHandler
	Logger       sdk.Logger
	Handler      FlowDefinitionHandler
	Runtime      *FlowRuntime
	// contains filtered or unexported fields
}

func (*FlowExecutor) Configure

func (fe *FlowExecutor) Configure(requestID string)

func (*FlowExecutor) GetDataStore

func (fe *FlowExecutor) GetDataStore() (sdk.DataStore, error)

func (*FlowExecutor) GetEventHandler

func (fe *FlowExecutor) GetEventHandler() (sdk.EventHandler, error)

func (*FlowExecutor) GetExecutionOption

func (fe *FlowExecutor) GetExecutionOption(_ sdk.Operation) map[string]interface{}

func (*FlowExecutor) GetFlowDefinition

func (fe *FlowExecutor) GetFlowDefinition(pipeline *sdk.Pipeline, context *sdk.Context) error

func (*FlowExecutor) GetFlowName

func (fe *FlowExecutor) GetFlowName() string

func (*FlowExecutor) GetLogger

func (fe *FlowExecutor) GetLogger() (sdk.Logger, error)

func (*FlowExecutor) GetReqAuthKey

func (fe *FlowExecutor) GetReqAuthKey() (string, error)

func (*FlowExecutor) GetStateStore

func (fe *FlowExecutor) GetStateStore() (sdk.StateStore, error)

func (*FlowExecutor) GetValidationKey

func (fe *FlowExecutor) GetValidationKey() (string, error)

func (*FlowExecutor) HandleExecutionCompletion

func (fe *FlowExecutor) HandleExecutionCompletion(data []byte) error

func (*FlowExecutor) HandleNextNode

func (fe *FlowExecutor) HandleNextNode(partial *executor.PartialState) error

func (*FlowExecutor) Init

func (fe *FlowExecutor) Init(request *runtime.Request) error

func (*FlowExecutor) LoggingEnabled

func (fe *FlowExecutor) LoggingEnabled() bool

func (*FlowExecutor) MonitoringEnabled

func (fe *FlowExecutor) MonitoringEnabled() bool

func (*FlowExecutor) ReqAuthEnabled

func (fe *FlowExecutor) ReqAuthEnabled() bool

func (*FlowExecutor) ReqValidationEnabled

func (fe *FlowExecutor) ReqValidationEnabled() bool

type FlowRuntime

type FlowRuntime struct {
	Flows          map[string]FlowDefinitionHandler
	OpenTracingUrl string
	RedisURL       string

	DataStore               sdk.DataStore
	Logger                  sdk.Logger
	Concurrency             int
	ServerPort              int
	ReadTimeout             time.Duration
	WriteTimeout            time.Duration
	RequestAuthSharedSecret string
	RequestAuthEnabled      bool
	EnableMonitoring        bool
	RetryQueueCount         int
	DebugEnabled            bool
	// contains filtered or unexported fields
}

func (*FlowRuntime) Consume

func (fRuntime *FlowRuntime) Consume(message rmq.Delivery)

Consume messages from queue

func (*FlowRuntime) CreateExecutor

func (fRuntime *FlowRuntime) CreateExecutor(req *runtime.Request) (executor.Executor, error)

func (*FlowRuntime) EnqueuePartialRequest

func (fRuntime *FlowRuntime) EnqueuePartialRequest(pr *runtime.Request) error

func (*FlowRuntime) Execute

func (fRuntime *FlowRuntime) Execute(flowName string, request *runtime.Request) error

func (*FlowRuntime) Init

func (fRuntime *FlowRuntime) Init() error

func (*FlowRuntime) Pause

func (fRuntime *FlowRuntime) Pause(flowName string, request *runtime.Request) error

func (*FlowRuntime) Resume

func (fRuntime *FlowRuntime) Resume(flowName string, request *runtime.Request) error

func (*FlowRuntime) StartQueueWorker

func (fRuntime *FlowRuntime) StartQueueWorker(errorChan chan error) error

StartQueueWorker starts listening for request in queue

func (*FlowRuntime) StartRuntime

func (fRuntime *FlowRuntime) StartRuntime() error

StartRuntime starts the runtime

func (*FlowRuntime) StartServer

func (fRuntime *FlowRuntime) StartServer() error

StartServer starts listening for new request

func (*FlowRuntime) Stop

func (fRuntime *FlowRuntime) Stop(flowName string, request *runtime.Request) error

func (*FlowRuntime) StopServer

func (fRuntime *FlowRuntime) StopServer() error

StopServer stops the server

type Task

type Task struct {
	FlowName    string              `json:"flow_name"`
	RequestID   string              `json:"request_id"`
	Body        string              `json:"body"`
	Header      map[string][]string `json:"header"`
	RawQuery    string              `json:"raw_query"`
	Query       map[string][]string `json:"query"`
	RequestType string              `json:"request_type"`
}

type Worker

type Worker struct {
	ID          string   `json:"id"`
	Flows       []string `json:"flows"`
	Concurrency int      `json:"concurrency"`
}