BC for the Pool, worker interfaces, pass/return payload by pointer

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

2 files changed, 34 insertions, 42 deletions

diff --git a/pkg/transport/pipe/pipe_factory_spawn_test.go b/pkg/transport/pipe/pipe_factory_spawn_test.go
index 51befb1e..f5e9669b 100644
--- a/pkg/transport/pipe/pipe_factory_spawn_test.go
+++ b/pkg/transport/pipe/pipe_factory_spawn_test.go
@@ -144,7 +144,7 @@ func Test_Pipe_Echo2(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.NoError(t, err)
 	assert.NotNil(t, res)
@@ -168,11 +168,10 @@ func Test_Pipe_Broken2(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.Error(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 }
 
 func Benchmark_Pipe_SpawnWorker_Stop2(b *testing.B) {
@@ -215,7 +214,7 @@ func Benchmark_Pipe_Worker_ExecEcho2(b *testing.B) {
 	}()
 
 	for n := 0; n < b.N; n++ {
-		if _, err := sw.Exec(payload.Payload{Body: []byte("hello")}); err != nil {
+		if _, err := sw.Exec(&payload.Payload{Body: []byte("hello")}); err != nil {
 			b.Fail()
 		}
 	}
@@ -238,7 +237,7 @@ func Benchmark_Pipe_Worker_ExecEcho4(b *testing.B) {
 	sw := worker.From(w)
 
 	for n := 0; n < b.N; n++ {
-		if _, err := sw.Exec(payload.Payload{Body: []byte("hello")}); err != nil {
+		if _, err := sw.Exec(&payload.Payload{Body: []byte("hello")}); err != nil {
 			b.Fail()
 		}
 	}
@@ -261,7 +260,7 @@ func Benchmark_Pipe_Worker_ExecEchoWithoutContext2(b *testing.B) {
 	sw := worker.From(w)
 
 	for n := 0; n < b.N; n++ {
-		if _, err := sw.Exec(payload.Payload{Body: []byte("hello")}); err != nil {
+		if _, err := sw.Exec(&payload.Payload{Body: []byte("hello")}); err != nil {
 			b.Fail()
 		}
 	}
@@ -287,7 +286,7 @@ func Test_Echo2(t *testing.T) {
 		}
 	}()
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.Nil(t, err)
 	assert.NotNil(t, res)
@@ -314,11 +313,10 @@ func Test_BadPayload2(t *testing.T) {
 		}
 	}()
 
-	res, err := sw.Exec(payload.Payload{})
+	res, err := sw.Exec(&payload.Payload{})
 
 	assert.Error(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 
 	assert.Contains(t, err.Error(), "payload can not be empty")
 }
@@ -358,7 +356,7 @@ func Test_Echo_Slow2(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.Nil(t, err)
 	assert.NotNil(t, res)
@@ -387,10 +385,9 @@ func Test_Broken2(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 	assert.NotNil(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 
 	time.Sleep(time.Second * 3)
 	mu.Lock()
@@ -418,10 +415,9 @@ func Test_Error2(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 	assert.NotNil(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 
 	if errors.Is(errors.SoftJob, err) == false {
 		t.Fatal("error should be of type errors.ErrSoftJob")
@@ -445,19 +441,19 @@ func Test_NumExecs2(t *testing.T) {
 
 	sw := worker.From(w)
 
-	_, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	_, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}
 	assert.Equal(t, uint64(1), w.State().NumExecs())
 
-	_, err = sw.Exec(payload.Payload{Body: []byte("hello")})
+	_, err = sw.Exec(&payload.Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}
 	assert.Equal(t, uint64(2), w.State().NumExecs())
 
-	_, err = sw.Exec(payload.Payload{Body: []byte("hello")})
+	_, err = sw.Exec(&payload.Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}
diff --git a/pkg/transport/pipe/pipe_factory_test.go b/pkg/transport/pipe/pipe_factory_test.go
index 3ef65be8..5c937a97 100755
--- a/pkg/transport/pipe/pipe_factory_test.go
+++ b/pkg/transport/pipe/pipe_factory_test.go
@@ -159,7 +159,7 @@ func Test_Pipe_Echo(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.NoError(t, err)
 	assert.NotNil(t, res)
@@ -184,11 +184,10 @@ func Test_Pipe_Broken(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.Error(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 }
 
 func Benchmark_Pipe_SpawnWorker_Stop(b *testing.B) {
@@ -231,7 +230,7 @@ func Benchmark_Pipe_Worker_ExecEcho(b *testing.B) {
 	}()
 
 	for n := 0; n < b.N; n++ {
-		if _, err := sw.Exec(payload.Payload{Body: []byte("hello")}); err != nil {
+		if _, err := sw.Exec(&payload.Payload{Body: []byte("hello")}); err != nil {
 			b.Fail()
 		}
 	}
@@ -255,7 +254,7 @@ func Benchmark_Pipe_Worker_ExecEcho3(b *testing.B) {
 	sw := worker.From(w)
 
 	for n := 0; n < b.N; n++ {
-		if _, err := sw.Exec(payload.Payload{Body: []byte("hello")}); err != nil {
+		if _, err := sw.Exec(&payload.Payload{Body: []byte("hello")}); err != nil {
 			b.Fail()
 		}
 	}
@@ -279,7 +278,7 @@ func Benchmark_Pipe_Worker_ExecEchoWithoutContext(b *testing.B) {
 	sw := worker.From(w)
 
 	for n := 0; n < b.N; n++ {
-		if _, err := sw.Exec(payload.Payload{Body: []byte("hello")}); err != nil {
+		if _, err := sw.Exec(&payload.Payload{Body: []byte("hello")}); err != nil {
 			b.Fail()
 		}
 	}
@@ -305,7 +304,7 @@ func Test_Echo(t *testing.T) {
 		}
 	}()
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.Nil(t, err)
 	assert.NotNil(t, res)
@@ -333,11 +332,10 @@ func Test_BadPayload(t *testing.T) {
 		}
 	}()
 
-	res, err := sw.Exec(payload.Payload{})
+	res, err := sw.Exec(&payload.Payload{})
 
 	assert.Error(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 
 	assert.Contains(t, err.Error(), "payload can not be empty")
 }
@@ -379,7 +377,7 @@ func Test_Echo_Slow(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 
 	assert.Nil(t, err)
 	assert.NotNil(t, res)
@@ -409,10 +407,9 @@ func Test_Broken(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 	assert.NotNil(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 
 	time.Sleep(time.Second * 3)
 	mu.Lock()
@@ -441,10 +438,9 @@ func Test_Error(t *testing.T) {
 
 	sw := worker.From(w)
 
-	res, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	res, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 	assert.NotNil(t, err)
-	assert.Nil(t, res.Body)
-	assert.Nil(t, res.Context)
+	assert.Nil(t, res)
 
 	if errors.Is(errors.SoftJob, err) == false {
 		t.Fatal("error should be of type errors.ErrSoftJob")
@@ -469,19 +465,19 @@ func Test_NumExecs(t *testing.T) {
 
 	sw := worker.From(w)
 
-	_, err := sw.Exec(payload.Payload{Body: []byte("hello")})
+	_, err := sw.Exec(&payload.Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}
 	assert.Equal(t, uint64(1), w.State().NumExecs())
 
-	_, err = sw.Exec(payload.Payload{Body: []byte("hello")})
+	_, err = sw.Exec(&payload.Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}
 	assert.Equal(t, uint64(2), w.State().NumExecs())
 
-	_, err = sw.Exec(payload.Payload{Body: []byte("hello")})
+	_, err = sw.Exec(&payload.Payload{Body: []byte("hello")})
 	if err != nil {
 		t.Errorf("fail to execute payload: error %v", err)
 	}