Merge #41

41: Jobs 2.1.4 r=48d90782 a=48d90782



Co-authored-by: Valery Piashchynski <[email protected]>

Author: bors[bot]

CHANGELOG.md

@@ -0,0 +1,10 @@
+CHANGELOG
+=========
+
+## v2.1.4 (23.03.2020)
+- Replaced std encoding/json package with the https://github.com/json-iterator/go 
+- Added BORS and GHA support
+- Added new exponential backoff mechanism for AMQP and Beanstalk modules
+- Fixed few concurrency issues with goroutines usage in loops
+- Other under-the-hood improvements with tests and code simplify
+- RR 1.7.0

broker/amqp/conn.go

@@ -14,7 +14,7 @@ type chanPool struct {
 	tout time.Duration
 	url  string
 
-	mu sync.Mutex
+	mu *sync.Mutex
 
 	conn      *amqp.Connection
 	channels  map[string]*channel
@@ -42,6 +42,7 @@ func newConn(url string, tout time.Duration) (*chanPool, error) {
 		url:       url,
 		tout:      tout,
 		conn:      conn,
+		mu:        &sync.Mutex{},
 		channels:  make(map[string]*channel),
 		wait:      make(chan interface{}),
 		connected: make(chan interface{}),

broker/amqp/consume_test.go

@@ -14,7 +14,10 @@ func TestBroker_Consume_Job(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	b.Register(pipe)
+	err = b.Register(pipe)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	ready := make(chan interface{})
 	b.Listen(func(event int, ctx interface{}) {
@@ -57,7 +60,10 @@ func TestBroker_ConsumeAfterStart_Job(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	b.Register(pipe)
+	err = b.Register(pipe)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	ready := make(chan interface{})
 	b.Listen(func(event int, ctx interface{}) {
@@ -101,7 +107,10 @@ func TestBroker_Consume_Delayed(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	b.Register(pipe)
+	err = b.Register(pipe)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	ready := make(chan interface{})
 	b.Listen(func(event int, ctx interface{}) {
@@ -140,7 +149,7 @@ func TestBroker_Consume_Delayed(t *testing.T) {
 
 	elapsed := time.Since(start)
 	assert.True(t, elapsed >= time.Second)
-	assert.True(t, elapsed < 2*time.Second)
+	assert.True(t, elapsed < 3*time.Second)
 }
 
 func TestBroker_Consume_Errored(t *testing.T) {
@@ -149,7 +158,10 @@ func TestBroker_Consume_Errored(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	b.Register(pipe)
+	err = b.Register(pipe)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	ready := make(chan interface{})
 	b.Listen(func(event int, ctx interface{}) {
@@ -195,7 +207,10 @@ func TestBroker_Consume_Errored_Attempts(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	b.Register(pipe)
+	err = b.Register(pipe)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	ready := make(chan interface{})
 	b.Listen(func(event int, ctx interface{}) {

broker/amqp/stat_test.go

@@ -3,6 +3,7 @@ package amqp
 import (
 	"github.com/spiral/jobs/v2"
 	"github.com/stretchr/testify/assert"
+	"sync"
 	"testing"
 )
 
@@ -40,20 +41,21 @@ func TestBroker_Stat(t *testing.T) {
 
 	assert.NoError(t, b.Consume(pipe, exec, func(id string, j *jobs.Job, err error) {}))
 
-	waitJob := make(chan interface{})
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
 	exec <- func(id string, j *jobs.Job) error {
+		defer wg.Done()
 		assert.Equal(t, jid, id)
 		assert.Equal(t, "body", j.Payload)
 
 		stat, err := b.Stat(pipe)
 		assert.NoError(t, err)
 		assert.Equal(t, int64(1), stat.Active)
 
-		close(waitJob)
 		return nil
 	}
 
-	<-waitJob
+	wg.Wait()
 	stat, err = b.Stat(pipe)
 	assert.NoError(t, err)
 	assert.Equal(t, int64(0), stat.Queue)