Merge pull request #373 from spiral/feature/new-worker-produces-active-worker

Feature/new worker produces active worker

1 files changed, 14 insertions, 28 deletions

diff --git a/sync_worker_test.go b/sync_worker_test.go
index f4868009..7f969283 100755
--- a/sync_worker_test.go
+++ b/sync_worker_test.go
@@ -2,11 +2,9 @@ package roadrunner
 
 import (
 	"context"
-	"errors"
 	"os/exec"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -34,7 +32,7 @@ func Test_Echo(t *testing.T) {
 		}
 	}()
 
-	res, err := syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	res, err := syncWorker.Exec(Payload{Body: []byte("hello")})
 
 	assert.Nil(t, err)
 	assert.NotNil(t, res)
@@ -65,7 +63,7 @@ func Test_BadPayload(t *testing.T) {
 		}
 	}()
 
-	res, err := syncWorker.ExecWithContext(ctx, EmptyPayload)
+	res, err := syncWorker.Exec(EmptyPayload)
 
 	assert.Error(t, err)
 	assert.Nil(t, res.Body)
@@ -84,7 +82,6 @@ func Test_NotStarted_String(t *testing.T) {
 }
 
 func Test_NotStarted_Exec(t *testing.T) {
-	ctx := context.Background()
 	cmd := exec.Command("php", "tests/client.php", "echo", "pipes")
 
 	w, _ := InitBaseWorker(cmd)
@@ -94,7 +91,7 @@ func Test_NotStarted_Exec(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	res, err := syncWorker.Exec(Payload{Body: []byte("hello")})
 
 	assert.Error(t, err)
 	assert.Nil(t, res.Body)
@@ -143,7 +140,7 @@ func Test_Echo_Slow(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	res, err := syncWorker.Exec(Payload{Body: []byte("hello")})
 
 	assert.Nil(t, err)
 	assert.NotNil(t, res)
@@ -164,28 +161,17 @@ func Test_Broken(t *testing.T) {
 
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
-	go func() {
-		assert.NotNil(t, w)
-		tt := time.NewTimer(time.Second * 10)
-		defer wg.Done()
-		for {
-			select {
-			case ev := <-w.Events():
-				assert.Contains(t, string(ev.Payload.([]byte)), "undefined_function()")
-				return
-			case <-tt.C:
-				assert.Error(t, errors.New("no events from worker"))
-				return
-			}
-		}
-	}()
 
+	w.AddListener(func(event interface{}) {
+		assert.Contains(t, string(event.(WorkerEvent).Payload.([]byte)), "undefined_function()")
+		wg.Done()
+	})
 	syncWorker, err := NewSyncWorker(w)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	res, err := syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	res, err := syncWorker.Exec(Payload{Body: []byte("hello")})
 	assert.NotNil(t, err)
 	assert.Nil(t, res.Body)
 	assert.Nil(t, res.Context)
@@ -215,12 +201,12 @@ func Test_Error(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	res, err := syncWorker.Exec(Payload{Body: []byte("hello")})
 	assert.NotNil(t, err)
 	assert.Nil(t, res.Body)
 	assert.Nil(t, res.Context)
 
-	assert.IsType(t, TaskError{}, err)
+	assert.IsType(t, JobError{}, err)
 	assert.Equal(t, "hello", err.Error())
 }
 
@@ -244,19 +230,19 @@ func Test_NumExecs(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	_, err = syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	_, err = syncWorker.Exec(Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}
 	assert.Equal(t, int64(1), w.State().NumExecs())
 
-	_, err = syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	_, err = syncWorker.Exec(Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}
 	assert.Equal(t, int64(2), w.State().NumExecs())
 
-	_, err = syncWorker.ExecWithContext(ctx, Payload{Body: []byte("hello")})
+	_, err = syncWorker.Exec(Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}