API update

1 files changed, 93 insertions, 82 deletions

diff --git a/worker_test.go b/worker_test.go
index d4d24364..0cb6f2d4 100644
--- a/worker_test.go
+++ b/worker_test.go
@@ -1,130 +1,141 @@
 package roadrunner
 
 import (
-	"github.com/spiral/goridge"
 	"github.com/stretchr/testify/assert"
-	"io"
 	"os/exec"
 	"testing"
 	"time"
 )
 
-func getPipes(cmd *exec.Cmd) (io.ReadCloser, io.WriteCloser) {
-	in, err := cmd.StdoutPipe()
-	if err != nil {
-		panic(err)
-	}
+func Test_GetState(t *testing.T) {
+	cmd := exec.Command("php", "tests/client.php", "echo", "pipes")
 
-	out, err := cmd.StdinPipe()
-	if err != nil {
-		panic(err)
-	}
+	w, err := NewPipeFactory().SpawnWorker(cmd)
+	go func() {
+		assert.NoError(t, w.Wait())
+	}()
 
-	return in, out
-}
-
-func TestOnStarted(t *testing.T) {
-	pr := exec.Command("php", "tests/echo-client.php")
-	pr.Start()
+	assert.NoError(t, err)
+	assert.NotNil(t, w)
 
-	_, err := NewWorker(pr)
-	assert.NotNil(t, err)
-	assert.Equal(t, "can't attach to running process", err.Error())
+	assert.Equal(t, StateReady, w.State().Value())
+	w.Stop()
+	assert.Equal(t, StateStopped, w.State().Value())
 }
 
-func TestNewWorkerState(t *testing.T) {
-	w, err := NewWorker(exec.Command("php", "tests/echo-client.php"))
-	assert.Nil(t, err)
-	assert.Equal(t, StateInactive, w.State)
+func Test_Echo(t *testing.T) {
+	cmd := exec.Command("php", "tests/client.php", "echo", "pipes")
 
-	w.attach(goridge.NewPipeRelay(getPipes(w.cmd)))
-	assert.Equal(t, StateBooting, w.State)
+	w, _ := NewPipeFactory().SpawnWorker(cmd)
+	go func() {
+		assert.NoError(t, w.Wait())
+	}()
+	defer w.Stop()
 
-	assert.Nil(t, w.Start())
-	assert.Equal(t, StateReady, w.State)
-}
+	res, err := w.Exec(&Payload{Body: []byte("hello")})
 
-func TestStop(t *testing.T) {
-	w, err := NewWorker(exec.Command("php", "tests/echo-client.php"))
 	assert.Nil(t, err)
+	assert.NotNil(t, res)
+	assert.NotNil(t, res.Body)
+	assert.Nil(t, res.Head)
 
-	w.attach(goridge.NewPipeRelay(getPipes(w.cmd)))
-	assert.Nil(t, w.Start())
-
-	w.Stop()
-	assert.Equal(t, StateStopped, w.State)
+	assert.Equal(t, "hello", res.String())
 }
 
-func TestEcho(t *testing.T) {
-	w, err := NewWorker(exec.Command("php", "tests/echo-client.php"))
-	assert.Nil(t, err)
+func Test_Echo_Slow(t *testing.T) {
+	cmd := exec.Command("php", "tests/slow-client.php", "echo", "pipes", "10", "10")
 
-	w.attach(goridge.NewPipeRelay(getPipes(w.cmd)))
-	assert.Nil(t, w.Start())
+	w, _ := NewPipeFactory().SpawnWorker(cmd)
+	go func() {
+		assert.NoError(t, w.Wait())
+	}()
+	defer w.Stop()
+
+	res, err := w.Exec(&Payload{Body: []byte("hello")})
 
-	r, ctx, err := w.Execute([]byte("hello"), nil)
 	assert.Nil(t, err)
-	assert.Nil(t, ctx)
-	assert.Equal(t, "hello", string(r))
+	assert.NotNil(t, res)
+	assert.NotNil(t, res.Body)
+	assert.Nil(t, res.Head)
+
+	assert.Equal(t, "hello", res.String())
 }
 
-func TestError(t *testing.T) {
-	w, err := NewWorker(exec.Command("php", "tests/error-client.php"))
-	assert.Nil(t, err)
+func Test_Broken(t *testing.T) {
+	cmd := exec.Command("php", "tests/client.php", "broken", "pipes")
 
-	w.attach(goridge.NewPipeRelay(getPipes(w.cmd)))
-	assert.Nil(t, w.Start())
+	w, err := NewPipeFactory().SpawnWorker(cmd)
+	go func() {
+		err := w.Wait()
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "undefined_function()")
+	}()
+	defer w.Stop()
 
-	r, ctx, err := w.Execute([]byte("hello"), nil)
-	assert.Nil(t, r)
+	res, err := w.Exec(&Payload{Body: []byte("hello")})
+	assert.Nil(t, res)
 	assert.NotNil(t, err)
-	assert.Nil(t, ctx)
+}
 
-	assert.IsType(t, JobError{}, err)
-	assert.Equal(t, "hello", err.Error())
+func Test_OnStarted(t *testing.T) {
+	cmd := exec.Command("php", "tests/client.php", "broken", "pipes")
+	assert.Nil(t, cmd.Start())
+
+	w, err := newWorker(cmd)
+	assert.Nil(t, w)
+	assert.NotNil(t, err)
+
+	assert.Equal(t, "can't attach to running process", err.Error())
 }
 
-func TestBroken(t *testing.T) {
-	w, err := NewWorker(exec.Command("php", "tests/broken-client.php"))
-	assert.Nil(t, err)
+func Test_Error(t *testing.T) {
+	cmd := exec.Command("php", "tests/client.php", "error", "pipes")
 
-	w.attach(goridge.NewPipeRelay(getPipes(w.cmd)))
-	assert.Nil(t, w.Start())
+	w, _ := NewPipeFactory().SpawnWorker(cmd)
+	go func() {
+		assert.NoError(t, w.Wait())
+	}()
+	defer w.Stop()
 
-	r, ctx, err := w.Execute([]byte("hello"), nil)
-	assert.Nil(t, r)
+	res, err := w.Exec(&Payload{Body: []byte("hello")})
+	assert.Nil(t, res)
 	assert.NotNil(t, err)
-	assert.Nil(t, ctx)
 
-	assert.IsType(t, WorkerError(""), err)
-	assert.Contains(t, err.Error(), "undefined_function()")
+	assert.IsType(t, JobError{}, err)
+	assert.Equal(t, "hello", err.Error())
 }
 
-func TestNumExecutions(t *testing.T) {
-	w, err := NewWorker(exec.Command("php", "tests/echo-client.php"))
-	assert.Nil(t, err)
+func Test_NumExecs(t *testing.T) {
+	cmd := exec.Command("php", "tests/client.php", "echo", "pipes")
 
-	w.attach(goridge.NewPipeRelay(getPipes(w.cmd)))
-	assert.Nil(t, w.Start())
+	w, _ := NewPipeFactory().SpawnWorker(cmd)
+	go func() {
+		assert.NoError(t, w.Wait())
+	}()
+	defer w.Stop()
 
-	w.Execute([]byte("hello"), nil)
-	assert.Equal(t, uint64(1), w.NumExecutions)
+	w.Exec(&Payload{Body: []byte("hello")})
+	assert.Equal(t, uint64(1), w.State().NumExecs())
 
-	w.Execute([]byte("hello"), nil)
-	assert.Equal(t, uint64(2), w.NumExecutions)
+	w.Exec(&Payload{Body: []byte("hello")})
+	assert.Equal(t, uint64(2), w.State().NumExecs())
 
-	w.Execute([]byte("hello"), nil)
-	assert.Equal(t, uint64(3), w.NumExecutions)
+	w.Exec(&Payload{Body: []byte("hello")})
+	assert.Equal(t, uint64(3), w.State().NumExecs())
 }
 
-func TestLastExecution(t *testing.T) {
-	w, err := NewWorker(exec.Command("php", "tests/echo-client.php"))
-	assert.Nil(t, err)
+func Test_StateUpdated(t *testing.T) {
+	cmd := exec.Command("php", "tests/client.php", "echo", "pipes")
 
-	w.attach(goridge.NewPipeRelay(getPipes(w.cmd)))
-	assert.Nil(t, w.Start())
+	w, _ := NewPipeFactory().SpawnWorker(cmd)
+	go func() {
+		assert.NoError(t, w.Wait())
+	}()
+	defer w.Stop()
 
 	tm := time.Now()
-	w.Execute([]byte("hello"), nil)
-	assert.True(t, w.Last.After(tm))
+	time.Sleep(time.Millisecond)
+
+	w.Exec(&Payload{Body: []byte("hello")})
+	assert.True(t, w.State().Updated().After(tm))
 }