diff options
Diffstat (limited to 'interfaces/worker')
-rw-r--r-- | interfaces/worker/factory.go | 18 | ||||
-rw-r--r-- | interfaces/worker/watcher.go | 26 | ||||
-rw-r--r-- | interfaces/worker/worker.go | 62 |
3 files changed, 106 insertions, 0 deletions
diff --git a/interfaces/worker/factory.go b/interfaces/worker/factory.go new file mode 100644 index 00000000..19e2bf5d --- /dev/null +++ b/interfaces/worker/factory.go @@ -0,0 +1,18 @@ +package worker + +import ( + "context" + "os/exec" +) + +// Factory is responsible of wrapping given command into tasks WorkerProcess. +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 +} diff --git a/interfaces/worker/watcher.go b/interfaces/worker/watcher.go new file mode 100644 index 00000000..ce2c1c5a --- /dev/null +++ b/interfaces/worker/watcher.go @@ -0,0 +1,26 @@ +package worker + +import "context" + +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 +} diff --git a/interfaces/worker/worker.go b/interfaces/worker/worker.go new file mode 100644 index 00000000..edbc68d9 --- /dev/null +++ b/interfaces/worker/worker.go @@ -0,0 +1,62 @@ +package worker + +import ( + "context" + "fmt" + "time" + + "github.com/spiral/goridge/v3" + "github.com/spiral/roadrunner/v2/interfaces/events" + "github.com/spiral/roadrunner/v2/internal" +) + +// Allocator is responsible for worker allocation in the pool +type Allocator func() (BaseProcess, error) + +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 SyncWorker interface { + // BaseProcess provides basic functionality for the SyncWorker + BaseProcess + // Exec used to execute payload on the SyncWorker, there is no TIMEOUTS + Exec(rqs internal.Payload) (internal.Payload, error) + // ExecWithContext used to handle Exec with TTL + ExecWithContext(ctx context.Context, p internal.Payload) (internal.Payload, error) +} |