dispatcher

package

v0.77.21 Latest Latest Go to latest Published: Jan 23, 2026 License: MIT Imports: 35 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/hatchet-dev/hatchet

Links

Documentation ¶

Index ¶

Constants
Variables
func UnmarshalPayload[T any](payload interface{}) (T, error)
type Dispatcher
type DispatcherImpl
- func New(fs ...DispatcherOpt) (*DispatcherImpl, error)
type DispatcherOpt
type DispatcherOpts
type StreamEventBuffer
- func NewStreamEventBuffer(timeout time.Duration) *StreamEventBuffer

Constants ¶

View Source

const HeartbeatInterval = 4 * time.Second

Variables ¶

View Source

var ErrWorkerNotFound = fmt.Errorf("worker not found")

Functions ¶

func UnmarshalPayload ¶

func UnmarshalPayload[T any](payload interface{}) (T, error)

Types ¶

type Dispatcher ¶

type Dispatcher interface {
	contracts.DispatcherServer
	Start() (func() error, error)
}

type DispatcherImpl ¶

type DispatcherImpl struct {
	contracts.UnimplementedDispatcherServer
	// contains filtered or unexported fields
}

func New ¶

func New(fs ...DispatcherOpt) (*DispatcherImpl, error)

func (*DispatcherImpl) Heartbeat ¶

func (s *DispatcherImpl) Heartbeat(ctx context.Context, req *contracts.HeartbeatRequest) (*contracts.HeartbeatResponse, error)

Heartbeat is used to update the last heartbeat time for a worker

func (*DispatcherImpl) Listen ¶

func (s *DispatcherImpl) Listen(request *contracts.WorkerListenRequest, stream contracts.Dispatcher_ListenServer) error

Subscribe handles a subscribe request from a client

func (*DispatcherImpl) ListenV2 ¶

func (s *DispatcherImpl) ListenV2(request *contracts.WorkerListenRequest, stream contracts.Dispatcher_ListenV2Server) error

ListenV2 is like Listen, but implementation does not include heartbeats. This should only used by SDKs against engine version v0.18.1+

func (*DispatcherImpl) PutOverridesData ¶

func (s *DispatcherImpl) PutOverridesData(ctx context.Context, request *contracts.OverridesData) (*contracts.OverridesDataResponse, error)

func (*DispatcherImpl) RefreshTimeout ¶

func (d *DispatcherImpl) RefreshTimeout(ctx context.Context, request *contracts.RefreshTimeoutRequest) (*contracts.RefreshTimeoutResponse, error)

func (*DispatcherImpl) Register ¶

func (s *DispatcherImpl) Register(ctx context.Context, request *contracts.WorkerRegisterRequest) (*contracts.WorkerRegisterResponse, error)

func (*DispatcherImpl) ReleaseSlot ¶

func (s *DispatcherImpl) ReleaseSlot(ctx context.Context, req *contracts.ReleaseSlotRequest) (*contracts.ReleaseSlotResponse, error)

func (*DispatcherImpl) SendGroupKeyActionEvent ¶

func (s *DispatcherImpl) SendGroupKeyActionEvent(ctx context.Context, request *contracts.GroupKeyActionEvent) (*contracts.ActionEventResponse, error)

func (*DispatcherImpl) SendStepActionEvent ¶

func (s *DispatcherImpl) SendStepActionEvent(ctx context.Context, request *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)

func (*DispatcherImpl) Start ¶

func (d *DispatcherImpl) Start() (func() error, error)

func (*DispatcherImpl) SubscribeToWorkflowEvents ¶

func (s *DispatcherImpl) SubscribeToWorkflowEvents(request *contracts.SubscribeToWorkflowEventsRequest, stream contracts.Dispatcher_SubscribeToWorkflowEventsServer) error

func (*DispatcherImpl) SubscribeToWorkflowRuns ¶

func (s *DispatcherImpl) SubscribeToWorkflowRuns(server contracts.Dispatcher_SubscribeToWorkflowRunsServer) error

func (*DispatcherImpl) Unsubscribe ¶

func (s *DispatcherImpl) Unsubscribe(ctx context.Context, request *contracts.WorkerUnsubscribeRequest) (*contracts.WorkerUnsubscribeResponse, error)

func (*DispatcherImpl) UpsertWorkerLabels ¶

func (s *DispatcherImpl) UpsertWorkerLabels(ctx context.Context, request *contracts.UpsertWorkerLabelsRequest) (*contracts.UpsertWorkerLabelsResponse, error)

type DispatcherOpt ¶

type DispatcherOpt func(*DispatcherOpts)

func WithAlerter ¶

func WithAlerter(a hatcheterrors.Alerter) DispatcherOpt

func WithCache ¶

func WithCache(cache cache.Cacheable) DispatcherOpt

func WithDataDecoderValidator ¶

func WithDataDecoderValidator(dv datautils.DataDecoderValidator) DispatcherOpt

func WithDefaultMaxWorkerBacklogSize ¶ added in v0.73.78

func WithDefaultMaxWorkerBacklogSize(size int64) DispatcherOpt

func WithDispatcherId ¶

func WithDispatcherId(dispatcherId string) DispatcherOpt

func WithLogger ¶

func WithLogger(l *zerolog.Logger) DispatcherOpt

func WithMessageQueueV1 ¶

func WithMessageQueueV1(mqv1 msgqueue.MessageQueue) DispatcherOpt

func WithPayloadSizeThreshold ¶

func WithPayloadSizeThreshold(threshold int) DispatcherOpt

func WithRepositoryV1 ¶

func WithRepositoryV1(r v1.Repository) DispatcherOpt

type DispatcherOpts ¶

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

type StreamEventBuffer ¶

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

func NewStreamEventBuffer ¶

func NewStreamEventBuffer(timeout time.Duration) *StreamEventBuffer

func (*StreamEventBuffer) AddEvent ¶

func (b *StreamEventBuffer) AddEvent(event *contracts.WorkflowEvent)

func (*StreamEventBuffer) Close ¶

func (b *StreamEventBuffer) Close()

func (*StreamEventBuffer) Events ¶

func (b *StreamEventBuffer) Events() <-chan *contracts.WorkflowEvent

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
contracts
v1

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