worker

Documentation

Index

• type Allocator
• type Options
  • func WithLog(z *zap.Logger) 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() worker.State
  • func (w *Process) Stop() error
  • func (w *Process) String() string
  • func (w *Process) Wait() error
  • func (w *Process) Write(p []byte) (n int, err error)
• type StateImpl
  • func NewWorkerState(value int64) *StateImpl
  • func (s *StateImpl) IsActive() bool
  • func (s *StateImpl) LastUsed() uint64
  • func (s *StateImpl) NumExecs() uint64
  • func (s *StateImpl) RegisterExec()
  • func (s *StateImpl) Set(value int64)
  • func (s *StateImpl) SetLastUsed(lu uint64)
  • func (s *StateImpl) String() string
  • func (s *StateImpl) Value() int64
• type SyncWorkerImpl
  • func From(process worker.BaseProcess) *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) ExecWithTTL(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() worker.State
  • func (tw *SyncWorkerImpl) Stop() error
  • func (tw *SyncWorkerImpl) String() string
  • func (tw *SyncWorkerImpl) Wait() error

Types

type Allocator

type Allocator func() (worker.SyncWorker, error)

Allocator is responsible for worker allocation in the pool

type Options

type Options func(p *Process)

func WithLog

func WithLog(z *zap.Logger) 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)

AttachRelay attaches relay to the worker

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 wait for process completion!

func (*Process) Pid

func (w *Process) Pid() int64

Pid returns worker pid.

func (*Process) Relay

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

Relay returns relay attached to the worker

func (*Process) Start

func (w *Process) Start() error

func (*Process) State

func (w *Process) State() worker.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.

func (*Process) Write

func (w *Process) Write(p []byte) (n int, err error)

Worker stderr

type StateImpl

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

func NewWorkerState

func NewWorkerState(value int64) *StateImpl

NewWorkerState initializes a state for the sync.Worker

func (*StateImpl) IsActive

func (s *StateImpl) IsActive() bool

IsActive returns true if WorkerProcess not Inactive or Stopped

func (*StateImpl) LastUsed

func (s *StateImpl) LastUsed() uint64

func (*StateImpl) NumExecs

func (s *StateImpl) NumExecs() uint64

NumExecs returns number of registered WorkerProcess execs.

func (*StateImpl) RegisterExec

func (s *StateImpl) RegisterExec()

RegisterExec register new execution atomically

func (*StateImpl) Set

func (s *StateImpl) Set(value int64)

Set change StateImpl value (status)

func (*StateImpl) SetLastUsed

func (s *StateImpl) SetLastUsed(lu uint64)

SetLastUsed Update last used time

func (*StateImpl) String

func (s *StateImpl) String() string

String returns current StateImpl as string.

func (*StateImpl) Value

func (s *StateImpl) Value() int64

Value StateImpl returns StateImpl value

type SyncWorkerImpl

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

func From

func From(process worker.BaseProcess) *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) ExecWithTTL

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

ExecWithTTL executes 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() worker.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