Update static_pool and worker to wait response from the worker

Signed-off-by: Valery Piashchynski <[email protected]>

2 files changed, 38 insertions, 34 deletions

diff --git a/transport/pipe/pipe_factory_spawn_test.go b/transport/pipe/pipe_factory_spawn_test.go
index 96dd37a6..7e04f113 100644
--- a/transport/pipe/pipe_factory_spawn_test.go
+++ b/transport/pipe/pipe_factory_spawn_test.go
@@ -133,18 +133,9 @@ func Test_Pipe_Invalid2(t *testing.T) {
 func Test_Pipe_Echo2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "echo", "pipes")
 	w, err := NewPipeFactory().SpawnWorker(cmd)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = w.Stop()
-		if err != nil {
-			t.Errorf("error stopping the Process: error %v", err)
-		}
-	}()
+	assert.NoError(t, err)
 
 	sw := worker.From(w)
-
 	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.NoError(t, err)
@@ -152,27 +143,31 @@ func Test_Pipe_Echo2(t *testing.T) {
 	assert.NotNil(t, res.Body)
 	assert.Empty(t, res.Context)
 
+	go func() {
+		if w.Wait() != nil {
+			t.Fail()
+		}
+	}()
+
 	assert.Equal(t, "hello", res.String())
+	err = w.Stop()
+	assert.NoError(t, err)
 }
 
 func Test_Pipe_Broken2(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "broken", "pipes")
 	w, err := NewPipeFactory().SpawnWorker(cmd)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		time.Sleep(time.Second)
-		err = w.Stop()
-		assert.Error(t, err)
-	}()
+	assert.NoError(t, err)
+	require.NotNil(t, w)
 
 	sw := worker.From(w)
-
 	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
-
 	assert.Error(t, err)
 	assert.Nil(t, res)
+
+	time.Sleep(time.Second)
+	err = w.Stop()
+	assert.Error(t, err)
 }
 
 func Benchmark_Pipe_SpawnWorker_Stop2(b *testing.B) {
@@ -315,7 +310,6 @@ func Test_BadPayload2(t *testing.T) {
 	}()
 
 	res, err := sw.Exec(&payload.Payload{})
-
 	assert.Error(t, err)
 	assert.Nil(t, res)
 
diff --git a/transport/pipe/pipe_factory_test.go b/transport/pipe/pipe_factory_test.go
index 7ca49d09..c69be298 100755
--- a/transport/pipe/pipe_factory_test.go
+++ b/transport/pipe/pipe_factory_test.go
@@ -159,6 +159,7 @@ func Test_Pipe_Echo(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+
 	defer func() {
 		err = w.Stop()
 		if err != nil {
@@ -167,14 +168,18 @@ func Test_Pipe_Echo(t *testing.T) {
 	}()
 
 	sw := worker.From(w)
-
 	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
-
 	assert.NoError(t, err)
 	assert.NotNil(t, res)
 	assert.NotNil(t, res.Body)
 	assert.Empty(t, res.Context)
 
+	go func() {
+		if w.Wait() != nil {
+			t.Fail()
+		}
+	}()
+
 	assert.Equal(t, "hello", res.String())
 }
 
@@ -194,14 +199,18 @@ func Test_Pipe_Echo_Script(t *testing.T) {
 	}()
 
 	sw := worker.From(w)
-
 	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
-
 	assert.NoError(t, err)
 	assert.NotNil(t, res)
 	assert.NotNil(t, res.Body)
 	assert.Empty(t, res.Context)
 
+	go func() {
+		if w.Wait() != nil {
+			t.Fail()
+		}
+	}()
+
 	assert.Equal(t, "hello", res.String())
 }
 
@@ -210,21 +219,22 @@ func Test_Pipe_Broken(t *testing.T) {
 	cmd := exec.Command("php", "../../tests/client.php", "broken", "pipes")
 	ctx := context.Background()
 	w, err := NewPipeFactory().SpawnWorkerWithTimeout(ctx, cmd)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		time.Sleep(time.Second)
-		err = w.Stop()
-		assert.Error(t, err)
+	require.NoError(t, err)
+	require.NotNil(t, w)
+
+	go func() {
+		errW := w.Wait()
+		require.Error(t, errW)
 	}()
 
 	sw := worker.From(w)
-
 	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
-
 	assert.Error(t, err)
 	assert.Nil(t, res)
+
+	time.Sleep(time.Second)
+	err = w.Stop()
+	assert.NoError(t, err)
 }
 
 func Benchmark_Pipe_SpawnWorker_Stop(b *testing.B) {