worker

Documentation

Index

• Constants
• type Allocator
• type BaseProcess
  • func FromSync(w *SyncWorkerImpl) BaseProcess
• type Options
  • func AddListeners(listeners ...events.Listener) Options
• type Process
  • func InitBaseWorker(cmd *exec.Cmd, options ...Options) (*Process, error)
  • func (w *Process) AttachRelay(rl relay.Relay)
  • func (w *Process) Created() time.Time
  • func (w *Process) Kill() error
  • func (w *Process) Pid() int64
  • func (w *Process) Relay() relay.Relay
  • func (w *Process) Start() error
  • func (w *Process) State() internal.State
  • func (w *Process) Stop() error
  • func (w *Process) String() string
  • func (w *Process) Wait() error
• type SyncWorker
• type SyncWorkerImpl
  • func From(process *Process) *SyncWorkerImpl
  • func (tw *SyncWorkerImpl) AttachRelay(rl relay.Relay)
  • func (tw *SyncWorkerImpl) Created() time.Time
  • func (tw *SyncWorkerImpl) Exec(p payload.Payload) (payload.Payload, error)
  • func (tw *SyncWorkerImpl) ExecWithTimeout(ctx context.Context, p payload.Payload) (payload.Payload, error)
  • func (tw *SyncWorkerImpl) Kill() error
  • func (tw *SyncWorkerImpl) Pid() int64
  • func (tw *SyncWorkerImpl) Relay() relay.Relay
  • func (tw *SyncWorkerImpl) Start() error
  • func (tw *SyncWorkerImpl) State() internal.State
  • func (tw *SyncWorkerImpl) Stop() error
  • func (tw *SyncWorkerImpl) String() string
  • func (tw *SyncWorkerImpl) Wait() error

Constants

const (
	// WaitDuration - for how long error buffer should attempt to aggregate error messages
	// before merging output together since lastError update (required to keep error update together).
	WaitDuration = 25 * time.Millisecond

	// ReadBufSize used to make a slice with specified length to read from stderr
	ReadBufSize = 10240 // Kb
)

Types

type Allocator

type Allocator func() (*SyncWorkerImpl, 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

	// 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() 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() relay.Relay

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

func FromSync

func FromSync(w *SyncWorkerImpl) BaseProcess

FromSync creates BaseProcess from SyncWorkerImpl

type Options

type Options func(p *Process)

func AddListeners

func AddListeners(listeners ...events.Listener) Options

type Process

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

Process - supervised process with api over goridge.Relay.

func InitBaseWorker

func InitBaseWorker(cmd *exec.Cmd, options ...Options) (*Process, error)

InitBaseWorker creates new Process over given exec.cmd.

func (*Process) AttachRelay

func (w *Process) AttachRelay(rl relay.Relay)

State return receive-only Process state object, state can be used to safely access Process status, time when status changed and number of Process executions.

func (*Process) Created

func (w *Process) Created() time.Time

Created returns time worker was created at.

func (*Process) Kill

func (w *Process) Kill() error

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

func (*Process) Pid

func (w *Process) Pid() int64

Pid returns worker pid.

func (*Process) Relay

func (w *Process) Relay() relay.Relay

State return receive-only Process state object, state can be used to safely access Process status, time when status changed and number of Process executions.

func (*Process) Start

func (w *Process) Start() error

func (*Process) State

func (w *Process) State() internal.State

State return receive-only Process state object, state can be used to safely access Process status, time when status changed and number of Process executions.

func (*Process) Stop

func (w *Process) Stop() error

Stop sends soft termination command to the Process and waits for process completion.

func (*Process) String

func (w *Process) String() string

String returns Process description. fmt.Stringer interface

func (*Process) Wait

func (w *Process) Wait() error

Wait must be called once for each Process, call will be released once Process 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.

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
	ExecWithTimeout(ctx context.Context, p payload.Payload) (payload.Payload, error)
}

type SyncWorkerImpl

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

func From

func From(process *Process) *SyncWorkerImpl

From creates SyncWorker from BaseProcess

func (*SyncWorkerImpl) AttachRelay

func (tw *SyncWorkerImpl) AttachRelay(rl relay.Relay)

func (*SyncWorkerImpl) Created

func (tw *SyncWorkerImpl) Created() time.Time

func (*SyncWorkerImpl) Exec

func (tw *SyncWorkerImpl) Exec(p payload.Payload) (payload.Payload, error)

Exec payload without TTL timeout.

func (*SyncWorkerImpl) ExecWithTimeout

func (tw *SyncWorkerImpl) ExecWithTimeout(ctx context.Context, p payload.Payload) (payload.Payload, error)

Exec payload without TTL timeout.

func (*SyncWorkerImpl) Kill

func (tw *SyncWorkerImpl) Kill() error

func (*SyncWorkerImpl) Pid

func (tw *SyncWorkerImpl) Pid() int64

func (*SyncWorkerImpl) Relay

func (tw *SyncWorkerImpl) Relay() relay.Relay

func (*SyncWorkerImpl) Start

func (tw *SyncWorkerImpl) Start() error

func (*SyncWorkerImpl) State

func (tw *SyncWorkerImpl) State() internal.State

func (*SyncWorkerImpl) Stop

func (tw *SyncWorkerImpl) Stop() error

func (*SyncWorkerImpl) String

func (tw *SyncWorkerImpl) String() string

func (*SyncWorkerImpl) Wait

func (tw *SyncWorkerImpl) Wait() error