worker

Documentation

Index

• type Allocator
• type BaseProcess
• type Factory
• type SyncWorker
• type Watcher

Types

type Allocator

type Allocator func() (BaseProcess, error)

Allocator is responsible for worker allocation in the pool

type BaseProcess

type BaseProcess interface {
	fmt.Stringer

	// Pid returns worker pid.
	Pid() int64

	// Created returns time worker was created at.
	Created() time.Time

	// AddListener attaches listener to consume worker events.
	AddListener(listener events.EventListener)

	// State return receive-only WorkerProcess state object, state can be used to safely access
	// WorkerProcess status, time when status changed and number of WorkerProcess executions.
	State() internal.State

	// Start used to run Cmd and immediately return
	Start() error

	// Wait must be called once for each WorkerProcess, call will be released once WorkerProcess is
	// complete and will return process error (if any), if stderr is presented it's value
	// will be wrapped as WorkerError. Method will return error code if php process fails
	// to find or Start the script.
	Wait() error

	// Stop sends soft termination command to the WorkerProcess and waits for process completion.
	Stop(ctx context.Context) error

	// Kill kills underlying process, make sure to call Wait() func to gather
	// error log from the stderr. Does not waits for process completion!
	Kill() error

	// Relay returns attached to worker goridge relay
	Relay() goridge.Relay

	// AttachRelay used to attach goridge relay to the worker process
	AttachRelay(rl goridge.Relay)
}

type Factory

type Factory interface {
	// SpawnWorkerWithContext creates new WorkerProcess process based on given command with context.
	// Process must not be started.
	SpawnWorkerWithContext(context.Context, *exec.Cmd) (BaseProcess, error)
	// SpawnWorker creates new WorkerProcess process based on given command.
	// Process must not be started.
	SpawnWorker(*exec.Cmd) (BaseProcess, error)
	// Close the factory and underlying connections.
	Close(ctx context.Context) error
}

Factory is responsible of wrapping given command into tasks WorkerProcess.

type SyncWorker

type SyncWorker interface {
	// BaseProcess provides basic functionality for the SyncWorker
	BaseProcess
	// Exec used to execute payload on the SyncWorker, there is no TIMEOUTS
	Exec(rqs payload.Payload) (payload.Payload, error)
	// ExecWithContext used to handle Exec with TTL
	ExecWithContext(ctx context.Context, p payload.Payload) (payload.Payload, error)
}

type Watcher

type Watcher interface {
	// AddToWatch used to add stack to wait its state
	AddToWatch(workers []BaseProcess) error

	// GetFreeWorker provide first free worker
	GetFreeWorker(ctx context.Context) (BaseProcess, error)

	// PutWorker enqueues worker back
	PushWorker(w BaseProcess)

	// AllocateNew used to allocate new worker and put in into the WorkerWatcher
	AllocateNew() error

	// Destroy destroys the underlying stack
	Destroy(ctx context.Context)

	// WorkersList return all stack w/o removing it from internal storage
	WorkersList() []BaseProcess

	// RemoveWorker remove worker from the stack
	RemoveWorker(wb BaseProcess) error
}