Merge pull request #173 from RichardKnop/feature/fix-send-group-concu…

…rrency

Fix pool logic to handle concurrency in SendGroup call.

integration-tests/suite_test.go

@@ -22,7 +22,8 @@ func (a ascendingInt64s) Less(i, j int) bool { return a[i] < a[j] }
 
 func testAll(server *machinery.Server, t *testing.T) {
 	testSendTask(server, t)
-	testSendGroup(server, t)
+	testSendGroup(server, t, 0) // with unlimited concurrency
+	testSendGroup(server, t, 2) // with limited concurrency (2 parallel tasks at the most)
 	testSendChord(server, t)
 	testSendChain(server, t)
 	testReturnJustError(server, t)
@@ -57,11 +58,11 @@ func testSendTask(server *machinery.Server, t *testing.T) {
 	}
 }
 
-func testSendGroup(server *machinery.Server, t *testing.T) {
+func testSendGroup(server *machinery.Server, t *testing.T, sendConcurrency int) {
 	t1, t2, t3 := newAddTask(1, 1), newAddTask(2, 2), newAddTask(5, 6)
 
 	group := tasks.NewGroup(t1, t2, t3)
-	asyncResults, err := server.SendGroup(group, 10)
+	asyncResults, err := server.SendGroup(group, sendConcurrency)
 	if err != nil {
 		t.Error(err)
 	}

v1/server.go

@@ -185,7 +185,9 @@ func (server *Server) SendGroup(group *tasks.Group, sendConcurrency int) ([]*bac
 
 	pool := make(chan struct{}, sendConcurrency)
 	go func() {
-		pool <- struct{}{}
+		for i := 0; i < sendConcurrency; i++ {
+			pool <- struct{}{}
+		}
 	}()
 
 	for i, signature := range group.Tasks {
@@ -198,13 +200,17 @@ func (server *Server) SendGroup(group *tasks.Group, sendConcurrency int) ([]*bac
 			defer wg.Done()
 
 			// Publish task
-			if err := server.broker.Publish(s); err != nil {
-				errorsChan <- fmt.Errorf("Publish message error: %s", err)
+
+			err := server.broker.Publish(s)
+
+			if sendConcurrency > 0 {
 				pool <- struct{}{}
-				return
 			}
 
-			pool <- struct{}{}
+			if err != nil {
+				errorsChan <- fmt.Errorf("Publish message error: %s", err)
+				return
+			}
 
 			asyncResults[index] = backends.NewAsyncResult(s, server.backend)
 		}(signature, i)