Merge pull request #452 from spiral/refactor/splitv2.0.0-alpha23

Refactor/split

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)
+}