client

Documentation

Index

• Constants
• type Action
• type ActionEvent
• type ActionEventResponse
• type ActionEventType
• type ActionPayload
• type ActionType
• type AdminClient
• type ChildWorkflowOpts
• type Client
  • func New(fs ...ClientOpt) (Client, error)
  • func NewFromConfigFile(cf *client.ClientConfigFile, fs ...ClientOpt) (Client, error)
• type ClientEventListener
• type ClientOpt
  • func InitWorkflows() ClientOpt
  • func WithHostPort(host string, port int) ClientOpt
  • func WithLogLevel(lvl string) ClientOpt
  • func WithNamespace(namespace string) ClientOpt
  • func WithTenantId(tenantId string) ClientOpt
  • func WithToken(token string) ClientOpt
  • func WithWorkflows(files []*types.Workflow) ClientOpt
• type ClientOpts
• type DispatcherClient
• type EventClient
• type GetActionListenerRequest
• type ListenerStrategy
• type PutOptFunc
• type RunHandler
• type ScheduleOptFunc
  • func WithInput(input any) ScheduleOptFunc
  • func WithSchedules(schedules ...time.Time) ScheduleOptFunc
• type StreamEvent
• type StreamHandler
• type SubscribeClient
• type WorkerActionListener
• type WorkflowEvent
• type WorkflowRunEvent
• type WorkflowRunEventHandler
• type WorkflowRunsListener
  • func (l *WorkflowRunsListener) AddWorkflowRun(workflowRunId string, handler WorkflowRunEventHandler) error
  • func (l *WorkflowRunsListener) Close() error
  • func (l *WorkflowRunsListener) Listen(ctx context.Context) error

Constants

const (
	DefaultActionListenerRetryInterval = 5 * time.Second
	DefaultActionListenerRetryCount    = 5
)

Types

type Action

type Action struct {
	// the worker id
	WorkerId string

	// the tenant id
	TenantId string

	// the workflow run id
	WorkflowRunId string

	// the get group key run id
	GetGroupKeyRunId string

	// the job id
	JobId string

	// the job name
	JobName string

	// the job run id
	JobRunId string

	// the step id
	StepId string

	// the step name
	StepName string

	// the step run id
	StepRunId string

	// the action id
	ActionId string

	// the action payload
	ActionPayload []byte

	// the action type
	ActionType ActionType
}

type ActionEvent

type ActionEvent struct {
	*Action

	// the event timestamp
	EventTimestamp *time.Time

	// the step event type
	EventType ActionEventType

	// The event payload. This must be JSON-compatible as it gets marshalled to a JSON string.
	EventPayload interface{}
}

type ActionEventResponse

type ActionEventResponse struct {
	// the tenant id
	TenantId string

	// the id of the worker
	WorkerId string
}

type ActionEventType

type ActionEventType string

const (
	ActionEventTypeUnknown   ActionEventType = "STEP_EVENT_TYPE_UNKNOWN"
	ActionEventTypeStarted   ActionEventType = "STEP_EVENT_TYPE_STARTED"
	ActionEventTypeCompleted ActionEventType = "STEP_EVENT_TYPE_COMPLETED"
	ActionEventTypeFailed    ActionEventType = "STEP_EVENT_TYPE_FAILED"
)

type ActionPayload

type ActionPayload func(target interface{}) error

ActionPayload unmarshals the action payload into the target. It also validates the resulting target.

type ActionType

type ActionType string

const (
	ActionTypeStartStepRun     ActionType = "START_STEP_RUN"
	ActionTypeCancelStepRun    ActionType = "CANCEL_STEP_RUN"
	ActionTypeStartGetGroupKey ActionType = "START_GET_GROUP_KEY"
)

type AdminClient

type AdminClient interface {
	PutWorkflow(workflow *types.Workflow, opts ...PutOptFunc) error
	ScheduleWorkflow(workflowName string, opts ...ScheduleOptFunc) error

	// RunWorkflow triggers a workflow run and returns the run id
	RunWorkflow(workflowName string, input interface{}) (string, error)

	RunChildWorkflow(workflowName string, input interface{}, opts *ChildWorkflowOpts) (string, error)

	PutRateLimit(key string, opts *types.RateLimitOpts) error
}

type ChildWorkflowOpts

type ChildWorkflowOpts struct {
	ParentId        string
	ParentStepRunId string
	ChildIndex      int
	ChildKey        *string
}

type Client

type Client interface {
	Admin() AdminClient
	Dispatcher() DispatcherClient
	Event() EventClient
	Subscribe() SubscribeClient
	API() *rest.ClientWithResponses
	TenantId() string
	Namespace() string
}

func New

func New(fs ...ClientOpt) (Client, error)

New creates a new client instance.

func NewFromConfigFile

func NewFromConfigFile(cf *client.ClientConfigFile, fs ...ClientOpt) (Client, error)

type ClientEventListener

type ClientEventListener interface {
	OnWorkflowEvent(ctx context.Context, event *WorkflowEvent) error
}

type ClientOpt

type ClientOpt func(*ClientOpts)

func InitWorkflows

func InitWorkflows() ClientOpt

func WithHostPort

func WithHostPort(host string, port int) ClientOpt

func WithLogLevel

func WithLogLevel(lvl string) ClientOpt

func WithNamespace

func WithNamespace(namespace string) ClientOpt

func WithTenantId

func WithTenantId(tenantId string) ClientOpt

func WithToken

func WithToken(token string) ClientOpt

func WithWorkflows

func WithWorkflows(files []*types.Workflow) ClientOpt

WithWorkflows sets the workflow files to use for the worker. If this is not passed in, the workflows files will be loaded from the .hatchet folder in the current directory.

type ClientOpts

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

type DispatcherClient

type DispatcherClient interface {
	GetActionListener(ctx context.Context, req *GetActionListenerRequest) (WorkerActionListener, error)

	SendStepActionEvent(ctx context.Context, in *ActionEvent) (*ActionEventResponse, error)

	SendGroupKeyActionEvent(ctx context.Context, in *ActionEvent) (*ActionEventResponse, error)
}

type EventClient

type EventClient interface {
	Push(ctx context.Context, eventKey string, payload interface{}) error

	PutLog(ctx context.Context, stepRunId, msg string) error

	PutStreamEvent(ctx context.Context, stepRunId string, message []byte) error
}

type GetActionListenerRequest

type GetActionListenerRequest struct {
	WorkerName string
	Services   []string
	Actions    []string
	MaxRuns    *int
}

TODO: add validator to client side

type ListenerStrategy

type ListenerStrategy string

const (
	ListenerStrategyV1 ListenerStrategy = "v1"
	ListenerStrategyV2 ListenerStrategy = "v2"
)

type PutOptFunc

type PutOptFunc func(*putOpts)

type RunHandler

type RunHandler func(event WorkflowEvent) error

type ScheduleOptFunc

type ScheduleOptFunc func(*scheduleOpts)

func WithInput

func WithInput(input any) ScheduleOptFunc

func WithSchedules

func WithSchedules(schedules ...time.Time) ScheduleOptFunc

type StreamEvent

type StreamEvent struct {
	Message []byte
}

type StreamHandler

type StreamHandler func(event StreamEvent) error

type SubscribeClient

type SubscribeClient interface {
	On(ctx context.Context, workflowRunId string, handler RunHandler) error

	Stream(ctx context.Context, workflowRunId string, handler StreamHandler) error

	SubscribeToWorkflowRunEvents(ctx context.Context) (*WorkflowRunsListener, error)
}

type WorkerActionListener

type WorkerActionListener interface {
	Actions(ctx context.Context) (<-chan *Action, error)

	Unregister() error
}

type WorkflowEvent

type WorkflowEvent *dispatchercontracts.WorkflowEvent

type WorkflowRunEvent

type WorkflowRunEvent *dispatchercontracts.WorkflowRunEvent

type WorkflowRunEventHandler

type WorkflowRunEventHandler func(event WorkflowRunEvent) error

type WorkflowRunsListener

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

func (*WorkflowRunsListener) AddWorkflowRun

func (l *WorkflowRunsListener) AddWorkflowRun(
	workflowRunId string,
	handler WorkflowRunEventHandler,
) error

func (*WorkflowRunsListener) Close

func (l *WorkflowRunsListener) Close() error

func (*WorkflowRunsListener) Listen

func (l *WorkflowRunsListener) Listen(ctx context.Context) error