refactor: rename subj to queue

Author: appleboy

_example/producer-consumer/consumer/main.go

@@ -32,7 +32,7 @@ func main() {
 
 	// define the worker
 	w := rabbitmq.NewWorker(
-		rabbitmq.WithSubj("direct_queue"),
+		rabbitmq.WithQueue("direct_queue"),
 		rabbitmq.WithExchangeType(rabbitmq.ExchangeDirect),
 		rabbitmq.WithExchangeName("direct_queue"),
 		rabbitmq.WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {

_example/producer-consumer/producer/main.go

@@ -27,7 +27,7 @@ func main() {
 
 	// define the worker
 	w := rabbitmq.NewWorker(
-		rabbitmq.WithSubj("direct_queue"),
+		rabbitmq.WithQueue("direct_queue"),
 		rabbitmq.WithExchangeType(rabbitmq.ExchangeDirect),
 		rabbitmq.WithExchangeName("direct_queue"),
 	)

_example/worker/main.go

@@ -30,7 +30,7 @@ func main() {
 
 	// define the worker
 	w := rabbitmq.NewWorker(
-		rabbitmq.WithSubj("sample_worker"),
+		rabbitmq.WithQueue("sample_worker"),
 		rabbitmq.WithExchangeName("sample_worker"),
 		rabbitmq.WithRoutingKey("sample_worker"),
 		rabbitmq.WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {

example_test.go

@@ -15,7 +15,7 @@ func Example_direct_exchange() {
 		Message: "foo",
 	}
 	w1 := NewWorker(
-		WithSubj("direct_queue"),
+		WithQueue("direct_queue"),
 		WithExchangeName("direct_exchange"),
 		WithExchangeType("direct"),
 		WithRoutingKey("direct_exchange"),
@@ -35,7 +35,7 @@ func Example_direct_exchange() {
 	q1.Start()
 
 	w2 := NewWorker(
-		WithSubj("direct_queue"),
+		WithQueue("direct_queue"),
 		WithExchangeName("direct_exchange"),
 		WithExchangeType("direct"),
 		WithRoutingKey("direct_exchange"),
@@ -98,7 +98,7 @@ func Example_fanout_exchange() {
 		Message: "foo",
 	}
 	w1 := NewWorker(
-		WithSubj("fanout_queue_1"),
+		WithQueue("fanout_queue_1"),
 		WithExchangeName("fanout_exchange"),
 		WithExchangeType("fanout"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
@@ -116,7 +116,7 @@ func Example_fanout_exchange() {
 	q1.Start()
 
 	w2 := NewWorker(
-		WithSubj("fanout_queue_2"),
+		WithQueue("fanout_queue_2"),
 		WithExchangeName("fanout_exchange"),
 		WithExchangeType("fanout"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {

options.go

@@ -33,7 +33,7 @@ type options struct {
 	runFunc func(context.Context, core.QueuedMessage) error
 	logger  queue.Logger
 	addr    string
-	subj    string
+	queue   string
 	tag     string
 	// Durable AMQP exchange name
 	exchangeName string
@@ -94,10 +94,10 @@ func WithAutoAck(val bool) Option {
 	}
 }
 
-// WithSubj setup the topic
-func WithSubj(val string) Option {
+// WithQueue setup the queue name
+func WithQueue(val string) Option {
 	return func(w *options) {
-		w.subj = val
+		w.queue = val
 	}
 }
 
@@ -118,7 +118,7 @@ func WithLogger(l queue.Logger) Option {
 func newOptions(opts ...Option) options {
 	defaultOpts := options{
 		addr:         "amqp://guest:guest@localhost:5672/",
-		subj:         "golang-queue",
+		queue:        "golang-queue",
 		tag:          "golang-queue",
 		exchangeName: "test-exchange",
 		exchangeType: ExchangeDirect,

rabbitmq.go

@@ -64,12 +64,12 @@ func NewWorker(opts ...Option) *Worker {
 func (w *Worker) startConsumer() (err error) {
 	w.startOnce.Do(func() {
 		q, err := w.channel.QueueDeclare(
-			w.opts.subj, // name
-			true,        // durable
-			false,       // delete when unused
-			false,       // exclusive
-			false,       // no-wait
-			nil,         // arguments
+			w.opts.queue, // name
+			true,         // durable
+			false,        // delete when unused
+			false,        // exclusive
+			false,        // no-wait
+			nil,          // arguments
 		)
 		if err != nil {
 			w.opts.logger.Error(err)

rabbitmq_test.go

@@ -30,7 +30,7 @@ func (m mockMessage) Bytes() []byte {
 
 func TestShutdownWorkFlow(t *testing.T) {
 	w := NewWorker(
-		WithSubj("test"),
+		WithQueue("test"),
 	)
 	q, err := queue.NewQueue(
 		queue.WithWorker(w),
@@ -50,7 +50,7 @@ func TestCustomFuncAndWait(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithSubj("test"),
+		WithQueue("test"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
 			log.Println("show message: " + string(m.Bytes()))
 			time.Sleep(500 * time.Millisecond)
@@ -99,7 +99,7 @@ func TestJobReachTimeout(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithSubj("JobReachTimeout"),
+		WithQueue("JobReachTimeout"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
 			for {
 				select {
@@ -135,7 +135,7 @@ func TestCancelJobAfterShutdown(t *testing.T) {
 		Message: "test",
 	}
 	w := NewWorker(
-		WithSubj("CancelJob"),
+		WithQueue("CancelJob"),
 		WithLogger(queue.NewLogger()),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
 			for {
@@ -172,7 +172,7 @@ func TestGoroutineLeak(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithSubj("GoroutineLeak"),
+		WithQueue("GoroutineLeak"),
 		WithLogger(queue.NewEmptyLogger()),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
 			for {
@@ -216,7 +216,7 @@ func TestGoroutinePanic(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithSubj("GoroutinePanic"),
+		WithQueue("GoroutinePanic"),
 		WithRoutingKey("GoroutinePanic"),
 		WithExchangeName("GoroutinePanic"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {