test: add Example_direct_queue testing

Author: appleboy

example_test.go

@@ -14,34 +14,74 @@ func Example_direct_exchange() {
 	m := mockMessage{
 		Message: "foo",
 	}
-	w := NewWorker(
+	w1 := NewWorker(
 		WithSubj("direct_queue"),
 		WithExchangeName("direct_exchange"),
-		WithRoutingKey("direct_queue"),
-		WithTag("direct_queue"),
+		WithExchangeType("direct"),
+		WithRoutingKey("direct_exchange"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
-			fmt.Println("get data:", string(m.Bytes()))
+			fmt.Println("worker01 get data:", string(m.Bytes()))
+			time.Sleep(100 * time.Millisecond)
 			return nil
 		}),
 	)
+
+	q1, err := queue.NewQueue(
+		queue.WithWorker(w1),
+	)
+	if err != nil {
+		w1.opts.logger.Error(err)
+	}
+	q1.Start()
+
+	w2 := NewWorker(
+		WithSubj("direct_queue"),
+		WithExchangeName("direct_exchange"),
+		WithExchangeType("direct"),
+		WithRoutingKey("direct_exchange"),
+		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+			fmt.Println("worker02 get data:", string(m.Bytes()))
+			time.Sleep(100 * time.Millisecond)
+			return nil
+		}),
+	)
+
+	q2, err := queue.NewQueue(
+		queue.WithWorker(w2),
+	)
+	if err != nil {
+		w2.opts.logger.Error(err)
+	}
+	q2.Start()
+
+	w := NewWorker(
+		WithExchangeName("direct_exchange"),
+		WithExchangeType("direct"),
+		WithRoutingKey("direct_exchange"),
+	)
+
 	q, err := queue.NewQueue(
 		queue.WithWorker(w),
-		queue.WithWorkerCount(1),
 	)
 	if err != nil {
 		w.opts.logger.Error(err)
 	}
 
-	q.Start()
 	time.Sleep(200 * time.Millisecond)
 	q.Queue(m)
 	q.Queue(m)
+	q.Queue(m)
+	q.Queue(m)
 	time.Sleep(200 * time.Millisecond)
 	q.Release()
+	q1.Release()
+	q2.Release()
 
-	// Output:
-	// get data: foo
-	// get data: foo
+	// Unordered Output:
+	// worker01 get data: foo
+	// worker02 get data: foo
+	// worker01 get data: foo
+	// worker02 get data: foo
 }
 
 // Fanout Exchange
@@ -65,9 +105,7 @@ func Example_fanout_exchange() {
 	if err != nil {
 		w1.opts.logger.Error(err)
 	}
-
 	q1.Start()
-	time.Sleep(200 * time.Millisecond)
 
 	w2 := NewWorker(
 		WithSubj("fanout_queue_2"),
@@ -85,9 +123,7 @@ func Example_fanout_exchange() {
 	if err != nil {
 		w2.opts.logger.Error(err)
 	}
-
 	q2.Start()
-	time.Sleep(200 * time.Millisecond)
 
 	w := NewWorker(
 		WithExchangeName("fanout_exchange"),