Documentation
¶
Overview ¶
Package grpcoperator serves v1.OperatorService, the engine API for operators that run outside the engine process (kind = GRPC in v1_operator). It is the gRPC adapter over internal/services/operatorsvc, which holds the session logic every operator host shares: this package parses the protocol, authenticates the caller, and runs the Listen message loop, and delegates the durable task stream to the dispatcher.
Index ¶
- Constants
- type OperatorService
- type OperatorServiceImpl
- func (s *OperatorServiceImpl) Cleanup() error
- func (s *OperatorServiceImpl) DurableTask(ctx context.Context, ...) error
- func (s *OperatorServiceImpl) Listen(ctx context.Context, ...) error
- func (s *OperatorServiceImpl) Register(ctx context.Context, req *v1contracts.OperatorRegisterRequest) (*v1contracts.OperatorRegisterResponse, error)
- func (s *OperatorServiceImpl) SendStepActionEvent(ctx context.Context, req *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)
- type OperatorServiceOpt
- func WithAnalytics(a analytics.Analytics) OperatorServiceOpt
- func WithDispatcher(d *dispatcher.DispatcherImpl) OperatorServiceOpt
- func WithLogger(l *zerolog.Logger) OperatorServiceOpt
- func WithMaxActionsPerOperator(n int64) OperatorServiceOpt
- func WithMaxListenStreamsPerOperator(n int) OperatorServiceOpt
- func WithRepository(r repository.Repository) OperatorServiceOpt
- func WithValidator(v validator.Validator) OperatorServiceOpt
- type OperatorServiceOpts
Constants ¶
const OperatorIdMetadataKey = "hatchet-operator-id"
OperatorIdMetadataKey is the incoming gRPC metadata key that carries the operator id on every RPC after Register (Listen, SendStepActionEvent, DurableTask).
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type OperatorService ¶
type OperatorService interface {
v1connect.OperatorServiceHandler
Cleanup() error
}
type OperatorServiceImpl ¶
type OperatorServiceImpl struct {
v1connect.UnimplementedOperatorServiceHandler
// contains filtered or unexported fields
}
func New ¶
func New(fs ...OperatorServiceOpt) (*OperatorServiceImpl, error)
func (*OperatorServiceImpl) Cleanup ¶
func (s *OperatorServiceImpl) Cleanup() error
Cleanup releases what the service holds between requests.
func (*OperatorServiceImpl) DurableTask ¶
func (s *OperatorServiceImpl) DurableTask(ctx context.Context, stream *connect.BidiStream[v1contracts.DurableTaskRequest, v1contracts.DurableTaskResponse]) error
DurableTask is the durable task event stream. The operator is authorized once when the stream opens, and the worker named by the first register_worker message is checked for ownership before the dispatcher sees it; the dispatcher then owns the session. The SDK's durable listener registers exactly once per stream (it opens a new stream to register again), so a second register_worker is refused as a protocol error rather than re-checked: the stream stays bound to the worker it was authorized for.
func (*OperatorServiceImpl) Listen ¶
func (s *OperatorServiceImpl) Listen(ctx context.Context, stream *connect.BidiStream[v1contracts.OperatorListenRequest, v1contracts.OperatorListenResponse]) error
Listen activates a registered worker for the lifetime of the stream and fans assigned actions out to it. One guarded sender owns the send side of the stream: actions go out through the dispatcher's fan-out and acks go out through the session handle, and both write through it. This goroutine consumes heartbeats, action deltas and pauses.
func (*OperatorServiceImpl) Register ¶
func (s *OperatorServiceImpl) Register(ctx context.Context, req *v1contracts.OperatorRegisterRequest) (*v1contracts.OperatorRegisterResponse, error)
Register upserts the operator by name and creates the worker for this connection, or resumes the worker named by req.WorkerId when it still belongs to the operator. The worker starts with no actions: the client adds them on the Listen stream.
func (*OperatorServiceImpl) SendStepActionEvent ¶
func (s *OperatorServiceImpl) SendStepActionEvent(ctx context.Context, req *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)
SendStepActionEvent reports task progress for an action delivered on a Listen stream. The event's worker must belong to the calling operator; the dispatcher then handles it exactly like an SDK worker's event.
type OperatorServiceOpt ¶
type OperatorServiceOpt func(*OperatorServiceOpts)
func WithAnalytics ¶
func WithAnalytics(a analytics.Analytics) OperatorServiceOpt
func WithDispatcher ¶
func WithDispatcher(d *dispatcher.DispatcherImpl) OperatorServiceOpt
func WithLogger ¶
func WithLogger(l *zerolog.Logger) OperatorServiceOpt
func WithMaxActionsPerOperator ¶
func WithMaxActionsPerOperator(n int64) OperatorServiceOpt
WithMaxActionsPerOperator caps the action links held across all workers of one operator; zero disables the cap.
func WithMaxListenStreamsPerOperator ¶
func WithMaxListenStreamsPerOperator(n int) OperatorServiceOpt
WithMaxListenStreamsPerOperator caps the Listen streams one operator may hold open on this replica; zero disables the cap.
func WithRepository ¶
func WithRepository(r repository.Repository) OperatorServiceOpt
func WithValidator ¶
func WithValidator(v validator.Validator) OperatorServiceOpt
type OperatorServiceOpts ¶
type OperatorServiceOpts struct {
// contains filtered or unexported fields
}
Source Files