Merge pull request #462 from spiral/feature/CLIv2.0.0-alpha29

feature/CLI

4 files changed, 105 insertions, 126 deletions

diff --git a/pkg/pool/static_pool.go b/pkg/pool/static_pool.go
index 2a06b255..23bb2d5f 100755
--- a/pkg/pool/static_pool.go
+++ b/pkg/pool/static_pool.go
@@ -22,20 +22,16 @@ const StopRequest = "{\"stop\":true}"
 // ErrorEncoder encode error or make a decision based on the error type
 type ErrorEncoder func(err error, w worker.BaseProcess) (payload.Payload, error)
 
-// Before is set of functions that executes BEFORE Exec
-type Before func(req payload.Payload) payload.Payload
-
-// After is set of functions that executes AFTER Exec
-type After func(req payload.Payload, resp payload.Payload) payload.Payload
-
 type Options func(p *StaticPool)
 
+type Command func() *exec.Cmd
+
 // StaticPool controls worker creation, destruction and task routing. Pool uses fixed amount of stack.
 type StaticPool struct {
 	cfg Config
 
 	// worker command creator
-	cmd func() *exec.Cmd
+	cmd Command
 
 	// creates and connects to stack
 	factory worker.Factory
@@ -43,20 +39,22 @@ type StaticPool struct {
 	// distributes the events
 	events events.Handler
 
+	// saved list of event listeners
+	listeners []events.EventListener
+
 	// manages worker states and TTLs
 	ww worker.Watcher
 
 	// allocate new worker
 	allocator worker.Allocator
 
+	// errEncoder is the default Exec error encoder
 	errEncoder ErrorEncoder
-	before     []Before
-	after      []After
 }
 
-// NewPool creates new worker pool and task multiplexer. StaticPool will initiate with one worker.
-func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory worker.Factory, cfg Config, options ...Options) (pool.Pool, error) {
-	const op = errors.Op("NewPool")
+// Initialize creates new worker pool and task multiplexer. StaticPool will initiate with one worker.
+func Initialize(ctx context.Context, cmd Command, factory worker.Factory, cfg Config, options ...Options) (pool.Pool, error) {
+	const op = errors.Op("Initialize")
 	if factory == nil {
 		return nil, errors.E(op, errors.Str("no factory initialized"))
 	}
@@ -72,11 +70,14 @@ func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory worker.Factory,
 		cmd:     cmd,
 		factory: factory,
 		events:  eventsPkg.NewEventsHandler(),
-		after:   make([]After, 0, 0),
-		before:  make([]Before, 0, 0),
 	}
 
-	p.allocator = newPoolAllocator(ctx, p.cfg.AllocateTimeout, factory, cmd)
+	// add pool options
+	for i := 0; i < len(options); i++ {
+		options[i](p)
+	}
+
+	p.allocator = p.newPoolAllocator(ctx, p.cfg.AllocateTimeout, factory, cmd)
 	p.ww = workerWatcher.NewWorkerWatcher(p.allocator, p.cfg.NumWorkers, p.events)
 
 	workers, err := p.allocateWorkers(p.cfg.NumWorkers)
@@ -92,14 +93,9 @@ func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory worker.Factory,
 
 	p.errEncoder = defaultErrEncoder(p)
 
-	// add pool options
-	for i := 0; i < len(options); i++ {
-		options[i](p)
-	}
-
 	// if supervised config not nil, guess, that pool wanted to be supervised
 	if cfg.Supervisor != nil {
-		sp := newPoolWatcher(p, p.events, p.cfg.Supervisor)
+		sp := supervisorWrapper(p, p.events, p.cfg.Supervisor)
 		// start watcher timer
 		sp.Start()
 		return sp, nil
@@ -108,20 +104,17 @@ func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory worker.Factory,
 	return p, nil
 }
 
-func ExecBefore(before ...Before) Options {
-	return func(p *StaticPool) {
-		p.before = append(p.before, before...)
-	}
-}
-
-func ExecAfter(after ...After) Options {
+func AddListeners(listeners ...events.EventListener) Options {
 	return func(p *StaticPool) {
-		p.after = append(p.after, after...)
+		p.listeners = listeners
+		for i := 0; i < len(listeners); i++ {
+			p.addListener(listeners[i])
+		}
 	}
 }
 
 // AddListener connects event listener to the pool.
-func (sp *StaticPool) AddListener(listener events.EventListener) {
+func (sp *StaticPool) addListener(listener events.EventListener) {
 	sp.events.AddListener(listener)
 }
 
@@ -151,44 +144,30 @@ func (sp *StaticPool) Exec(p payload.Payload) (payload.Payload, error) {
 		return payload.Payload{}, errors.E(op, err)
 	}
 
-	sw := w.(worker.SyncWorker)
-
-	if len(sp.before) > 0 {
-		for i := 0; i < len(sp.before); i++ {
-			p = sp.before[i](p)
-		}
-	}
-
-	rsp, err := sw.Exec(p)
+	rsp, err := w.Exec(p)
 	if err != nil {
-		return sp.errEncoder(err, sw)
+		return sp.errEncoder(err, w)
 	}
 
 	// worker want's to be terminated
 	// TODO careful with string(rsp.Context)
 	if len(rsp.Body) == 0 && string(rsp.Context) == StopRequest {
-		sw.State().Set(internal.StateInvalid)
-		err = sw.Stop()
+		w.State().Set(internal.StateInvalid)
+		err = w.Stop()
 		if err != nil {
-			sp.events.Push(events.WorkerEvent{Event: events.EventWorkerError, Worker: sw, Payload: errors.E(op, err)})
+			sp.events.Push(events.WorkerEvent{Event: events.EventWorkerError, Worker: w, Payload: errors.E(op, err)})
 		}
 
 		return sp.Exec(p)
 	}
 
-	if sp.cfg.MaxJobs != 0 && sw.State().NumExecs() >= sp.cfg.MaxJobs {
+	if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs {
 		err = sp.ww.AllocateNew()
 		if err != nil {
 			return payload.Payload{}, errors.E(op, err)
 		}
 	} else {
-		sp.ww.PushWorker(sw)
-	}
-
-	if len(sp.after) > 0 {
-		for i := 0; i < len(sp.after); i++ {
-			rsp = sp.after[i](p, rsp)
-		}
+		sp.ww.PushWorker(w)
 	}
 
 	return rsp, nil
@@ -196,20 +175,13 @@ func (sp *StaticPool) Exec(p payload.Payload) (payload.Payload, error) {
 
 func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs payload.Payload) (payload.Payload, error) {
 	const op = errors.Op("exec with context")
-	ctxGetFree, cancel := context.WithTimeout(context.Background(), sp.cfg.AllocateTimeout)
+	ctxGetFree, cancel := context.WithTimeout(ctx, sp.cfg.AllocateTimeout)
 	defer cancel()
 	w, err := sp.getWorker(ctxGetFree, op)
 	if err != nil {
 		return payload.Payload{}, errors.E(op, err)
 	}
 
-	// apply all before function
-	if len(sp.before) > 0 {
-		for i := 0; i < len(sp.before); i++ {
-			rqs = sp.before[i](rqs)
-		}
-	}
-
 	rsp, err := w.ExecWithTimeout(ctx, rqs)
 	if err != nil {
 		return sp.errEncoder(err, w)
@@ -223,7 +195,7 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs payload.Payload)
 			sp.events.Push(events.WorkerEvent{Event: events.EventWorkerError, Worker: w, Payload: errors.E(op, err)})
 		}
 
-		return sp.Exec(rqs)
+		return sp.ExecWithContext(ctx, rqs)
 	}
 
 	if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs {
@@ -235,13 +207,6 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs payload.Payload)
 		sp.ww.PushWorker(w)
 	}
 
-	// apply all after functions
-	if len(sp.after) > 0 {
-		for i := 0; i < len(sp.after); i++ {
-			rsp = sp.after[i](rqs, rsp)
-		}
-	}
-
 	return rsp, nil
 }
 
@@ -300,11 +265,11 @@ func defaultErrEncoder(sp *StaticPool) ErrorEncoder {
 	}
 }
 
-func newPoolAllocator(ctx context.Context, timeout time.Duration, factory worker.Factory, cmd func() *exec.Cmd) worker.Allocator {
+func (sp *StaticPool) newPoolAllocator(ctx context.Context, timeout time.Duration, factory worker.Factory, cmd func() *exec.Cmd) worker.Allocator {
 	return func() (worker.BaseProcess, error) {
 		ctx, cancel := context.WithTimeout(ctx, timeout)
 		defer cancel()
-		w, err := factory.SpawnWorkerWithTimeout(ctx, cmd())
+		w, err := factory.SpawnWorkerWithTimeout(ctx, cmd(), sp.listeners...)
 		if err != nil {
 			return nil, err
 		}
@@ -313,6 +278,11 @@ func newPoolAllocator(ctx context.Context, timeout time.Duration, factory worker
 		if err != nil {
 			return nil, err
 		}
+
+		sp.events.Push(events.PoolEvent{
+			Event:   events.EventWorkerConstruct,
+			Payload: sw,
+		})
 		return sw, nil
 	}
 }
diff --git a/pkg/pool/static_pool_test.go b/pkg/pool/static_pool_test.go
index 30345aee..acdd6ab7 100755
--- a/pkg/pool/static_pool_test.go
+++ b/pkg/pool/static_pool_test.go
@@ -27,7 +27,7 @@ var cfg = Config{
 
 func Test_NewPool(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -41,7 +41,7 @@ func Test_NewPool(t *testing.T) {
 }
 
 func Test_StaticPool_Invalid(t *testing.T) {
-	p, err := NewPool(
+	p, err := Initialize(
 		context.Background(),
 		func() *exec.Cmd { return exec.Command("php", "../../tests/invalid.php") },
 		pipe.NewPipeFactory(),
@@ -53,7 +53,7 @@ func Test_StaticPool_Invalid(t *testing.T) {
 }
 
 func Test_ConfigNoErrorInitDefaults(t *testing.T) {
-	p, err := NewPool(
+	p, err := Initialize(
 		context.Background(),
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -69,7 +69,7 @@ func Test_ConfigNoErrorInitDefaults(t *testing.T) {
 
 func Test_StaticPool_Echo(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -93,7 +93,7 @@ func Test_StaticPool_Echo(t *testing.T) {
 
 func Test_StaticPool_Echo_NilContext(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -117,7 +117,7 @@ func Test_StaticPool_Echo_NilContext(t *testing.T) {
 
 func Test_StaticPool_Echo_Context(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "head", "pipes") },
 		pipe.NewPipeFactory(),
@@ -141,7 +141,7 @@ func Test_StaticPool_Echo_Context(t *testing.T) {
 
 func Test_StaticPool_JobError(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "error", "pipes") },
 		pipe.NewPipeFactory(),
@@ -167,18 +167,9 @@ func Test_StaticPool_JobError(t *testing.T) {
 
 func Test_StaticPool_Broken_Replace(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
-		ctx,
-		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "broken", "pipes") },
-		pipe.NewPipeFactory(),
-		cfg,
-	)
-	assert.NoError(t, err)
-	assert.NotNil(t, p)
-
 	block := make(chan struct{}, 1)
 
-	p.AddListener(func(event interface{}) {
+	listener := func(event interface{}) {
 		if wev, ok := event.(events.WorkerEvent); ok {
 			if wev.Event == events.EventWorkerLog {
 				e := string(wev.Payload.([]byte))
@@ -188,7 +179,17 @@ func Test_StaticPool_Broken_Replace(t *testing.T) {
 				}
 			}
 		}
-	})
+	}
+
+	p, err := Initialize(
+		ctx,
+		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "broken", "pipes") },
+		pipe.NewPipeFactory(),
+		cfg,
+		AddListeners(listener),
+	)
+	assert.NoError(t, err)
+	assert.NotNil(t, p)
 
 	time.Sleep(time.Second)
 	res, err := p.ExecWithContext(ctx, payload.Payload{Body: []byte("hello")})
@@ -203,11 +204,28 @@ func Test_StaticPool_Broken_Replace(t *testing.T) {
 
 func Test_StaticPool_Broken_FromOutside(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	// Consume pool events
+	ev := make(chan struct{}, 1)
+	listener := func(event interface{}) {
+		if pe, ok := event.(events.PoolEvent); ok {
+			if pe.Event == events.EventWorkerConstruct {
+				ev <- struct{}{}
+			}
+		}
+	}
+
+	var cfg = Config{
+		NumWorkers:      1,
+		AllocateTimeout: time.Second * 5,
+		DestroyTimeout:  time.Second * 5,
+	}
+
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
 		cfg,
+		AddListeners(listener),
 	)
 	assert.NoError(t, err)
 	defer p.Destroy(ctx)
@@ -222,36 +240,27 @@ func Test_StaticPool_Broken_FromOutside(t *testing.T) {
 	assert.Empty(t, res.Context)
 
 	assert.Equal(t, "hello", res.String())
-	assert.Equal(t, runtime.NumCPU(), len(p.Workers()))
-
-	// Consume pool events
-	wg := sync.WaitGroup{}
-	wg.Add(1)
-	p.AddListener(func(event interface{}) {
-		if pe, ok := event.(events.PoolEvent); ok {
-			if pe.Event == events.EventWorkerConstruct {
-				wg.Done()
-			}
-		}
-	})
+	assert.Equal(t, 1, len(p.Workers()))
 
+	// first creation
+	<-ev
 	// killing random worker and expecting pool to replace it
 	err = p.Workers()[0].Kill()
 	if err != nil {
 		t.Errorf("error killing the process: error %v", err)
 	}
 
-	wg.Wait()
+	// re-creation
+	<-ev
 
 	list := p.Workers()
 	for _, w := range list {
 		assert.Equal(t, internal.StateReady, w.State().Value())
 	}
-	wg.Wait()
 }
 
 func Test_StaticPool_AllocateTimeout(t *testing.T) {
-	p, err := NewPool(
+	p, err := Initialize(
 		context.Background(),
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "delay", "pipes") },
 		pipe.NewPipeFactory(),
@@ -270,7 +279,7 @@ func Test_StaticPool_AllocateTimeout(t *testing.T) {
 
 func Test_StaticPool_Replace_Worker(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "pid", "pipes") },
 		pipe.NewPipeFactory(),
@@ -307,7 +316,7 @@ func Test_StaticPool_Replace_Worker(t *testing.T) {
 
 func Test_StaticPool_Debug_Worker(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "pid", "pipes") },
 		pipe.NewPipeFactory(),
@@ -347,7 +356,7 @@ func Test_StaticPool_Debug_Worker(t *testing.T) {
 // identical to replace but controlled on worker side
 func Test_StaticPool_Stop_Worker(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "stop", "pipes") },
 		pipe.NewPipeFactory(),
@@ -387,7 +396,7 @@ func Test_StaticPool_Stop_Worker(t *testing.T) {
 // identical to replace but controlled on worker side
 func Test_Static_Pool_Destroy_And_Close(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "delay", "pipes") },
 		pipe.NewPipeFactory(),
@@ -409,7 +418,7 @@ func Test_Static_Pool_Destroy_And_Close(t *testing.T) {
 // identical to replace but controlled on worker side
 func Test_Static_Pool_Destroy_And_Close_While_Wait(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "delay", "pipes") },
 		pipe.NewPipeFactory(),
@@ -439,7 +448,7 @@ func Test_Static_Pool_Destroy_And_Close_While_Wait(t *testing.T) {
 // identical to replace but controlled on worker side
 func Test_Static_Pool_Handle_Dead(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		context.Background(),
 		func() *exec.Cmd { return exec.Command("php", "../../tests/slow-destroy.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -464,7 +473,7 @@ func Test_Static_Pool_Handle_Dead(t *testing.T) {
 
 // identical to replace but controlled on worker side
 func Test_Static_Pool_Slow_Destroy(t *testing.T) {
-	p, err := NewPool(
+	p, err := Initialize(
 		context.Background(),
 		func() *exec.Cmd { return exec.Command("php", "../../tests/slow-destroy.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -483,7 +492,7 @@ func Test_Static_Pool_Slow_Destroy(t *testing.T) {
 
 func Benchmark_Pool_Echo(b *testing.B) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -505,7 +514,7 @@ func Benchmark_Pool_Echo(b *testing.B) {
 //
 func Benchmark_Pool_Echo_Batched(b *testing.B) {
 	ctx := context.Background()
-	p, _ := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -515,6 +524,7 @@ func Benchmark_Pool_Echo_Batched(b *testing.B) {
 			DestroyTimeout:  time.Second,
 		},
 	)
+	assert.NoError(b, err)
 	defer p.Destroy(ctx)
 
 	var wg sync.WaitGroup
@@ -535,7 +545,7 @@ func Benchmark_Pool_Echo_Batched(b *testing.B) {
 //
 func Benchmark_Pool_Echo_Replaced(b *testing.B) {
 	ctx := context.Background()
-	p, _ := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/client.php", "echo", "pipes") },
 		pipe.NewPipeFactory(),
@@ -546,6 +556,7 @@ func Benchmark_Pool_Echo_Replaced(b *testing.B) {
 			DestroyTimeout:  time.Second,
 		},
 	)
+	assert.NoError(b, err)
 	defer p.Destroy(ctx)
 	b.ResetTimer()
 	b.ReportAllocs()
diff --git a/pkg/pool/supervisor_pool.go b/pkg/pool/supervisor_pool.go
index 6faa609c..378be7dd 100755
--- a/pkg/pool/supervisor_pool.go
+++ b/pkg/pool/supervisor_pool.go
@@ -6,12 +6,12 @@ import (
 	"time"
 
 	"github.com/spiral/errors"
-	"github.com/spiral/roadrunner/v2"
 	"github.com/spiral/roadrunner/v2/interfaces/events"
 	"github.com/spiral/roadrunner/v2/interfaces/pool"
 	"github.com/spiral/roadrunner/v2/interfaces/worker"
 	"github.com/spiral/roadrunner/v2/internal"
 	"github.com/spiral/roadrunner/v2/pkg/payload"
+	"github.com/spiral/roadrunner/v2/tools"
 )
 
 const MB = 1024 * 1024
@@ -30,7 +30,7 @@ type supervised struct {
 	mu     *sync.RWMutex
 }
 
-func newPoolWatcher(pool pool.Pool, events events.Handler, cfg *SupervisorConfig) Supervised {
+func supervisorWrapper(pool pool.Pool, events events.Handler, cfg *SupervisorConfig) Supervised {
 	sp := &supervised{
 		cfg:    cfg,
 		events: events,
@@ -38,6 +38,7 @@ func newPoolWatcher(pool pool.Pool, events events.Handler, cfg *SupervisorConfig
 		mu:     &sync.RWMutex{},
 		stopCh: make(chan struct{}),
 	}
+
 	return sp
 }
 
@@ -93,10 +94,6 @@ func (sp *supervised) Exec(p payload.Payload) (payload.Payload, error) {
 	return rsp, nil
 }
 
-func (sp *supervised) AddListener(listener events.EventListener) {
-	sp.pool.AddListener(listener)
-}
-
 func (sp *supervised) GetConfig() interface{} {
 	return sp.pool.GetConfig()
 }
@@ -149,7 +146,7 @@ func (sp *supervised) control() {
 			continue
 		}
 
-		s, err := roadrunner.WorkerProcessState(workers[i])
+		s, err := tools.WorkerProcessState(workers[i])
 		if err != nil {
 			// worker not longer valid for supervision
 			continue
diff --git a/pkg/pool/supervisor_test.go b/pkg/pool/supervisor_test.go
index 7dd423b8..cb67ebe1 100644
--- a/pkg/pool/supervisor_test.go
+++ b/pkg/pool/supervisor_test.go
@@ -6,9 +6,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/spiral/roadrunner/v2"
 	"github.com/spiral/roadrunner/v2/pkg/payload"
 	"github.com/spiral/roadrunner/v2/pkg/pipe"
+	"github.com/spiral/roadrunner/v2/tools"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -27,7 +27,7 @@ var cfgSupervised = Config{
 
 func TestSupervisedPool_Exec(t *testing.T) {
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/memleak.php", "pipes") },
 		pipe.NewPipeFactory(),
@@ -47,7 +47,7 @@ func TestSupervisedPool_Exec(t *testing.T) {
 			default:
 				workers := p.Workers()
 				if len(workers) > 0 {
-					s, err := roadrunner.WorkerProcessState(workers[0])
+					s, err := tools.WorkerProcessState(workers[0])
 					assert.NoError(t, err)
 					assert.NotNil(t, s)
 					// since this is soft limit, double max memory limit watch
@@ -85,7 +85,7 @@ func TestSupervisedPool_ExecTTL_TimedOut(t *testing.T) {
 		},
 	}
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/sleep.php", "pipes") },
 		pipe.NewPipeFactory(),
@@ -104,7 +104,8 @@ func TestSupervisedPool_ExecTTL_TimedOut(t *testing.T) {
 	})
 
 	assert.Error(t, err)
-	assert.Empty(t, resp)
+	assert.Empty(t, resp.Body)
+	assert.Empty(t, resp.Context)
 
 	time.Sleep(time.Second * 1)
 	// should be new worker with new pid
@@ -125,7 +126,7 @@ func TestSupervisedPool_ExecTTL_OK(t *testing.T) {
 		},
 	}
 	ctx := context.Background()
-	p, err := NewPool(
+	p, err := Initialize(
 		ctx,
 		func() *exec.Cmd { return exec.Command("php", "../../tests/sleep.php", "pipes") },
 		pipe.NewPipeFactory(),