Add unit test for RabbitMQ broker.

Author: Ron Cemer

celery_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/go-kit/log"
 
 	"github.com/marselester/gopher-celery/goredis"
+	"github.com/marselester/gopher-celery/rabbitmq"
 	"github.com/marselester/gopher-celery/protocol"
 )
 
@@ -243,6 +244,50 @@ func TestGoredisProduceAndConsume100times(t *testing.T) {
 	}
 }
 
+func TestRabbitmqProduceAndConsume100times(t *testing.T) {
+    app := NewApp(
+        WithBroker(rabbitmq.NewBroker(rabbitmq.WithAmqpUri("amqp://guest:guest@localhost:5672/"))),
+		WithLogger(log.NewJSONLogger(os.Stderr)),
+    )
+	for i := 0; i < 100; i++ {
+		err := app.Delay(
+			"myproject.apps.myapp.tasks.mytask",
+			"important",
+			2,
+			3,
+		)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// The test finishes either when ctx times out or all the tasks finish.
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	t.Cleanup(cancel)
+
+	var sum int32
+	app.Register(
+		"myproject.apps.myapp.tasks.mytask",
+		"important",
+		func(ctx context.Context, p *TaskParam) error {
+			p.NameArgs("a", "b")
+			atomic.AddInt32(
+				&sum,
+				int32(p.MustInt("a")+p.MustInt("b")),
+			)
+			return nil
+		},
+	)
+	if err := app.Run(ctx); err != nil {
+		t.Error(err)
+	}
+
+	var want int32 = 500
+	if want != sum {
+		t.Errorf("expected sum %d got %d", want, sum)
+	}
+}
+
 func TestConsumeSequentially(t *testing.T) {
 	app := NewApp(
 		WithLogger(log.NewJSONLogger(os.Stderr)),