feat: refactor GitHub Actions and modify NATS tests

- Modify the `go.yml` file in the `.github/workflows` directory
- Add a new job named `nats01` in the `go.yml` file
- Add a new job named `nats02` in the `go.yml` file
- Modify the `nats_test.go` file
- Change the value of the `host` variable in the `nats_test.go` file
- Modify the `TestDefaultFlow` function in the `nats_test.go` file
- Modify the `TestShutdown` function in the `nats_test.go` file
- Modify the `TestCustomFuncAndWait` function in the `nats_test.go` file
- Modify the `TestEnqueueJobAfterShutdown` function in the `nats_test.go` file
- Modify the `TestJobReachTimeout` function in the `nats_test.go` file
- Modify the `TestCancelJobAfterShutdown` function in the `nats_test.go` file
- Modify the `TestGoroutineLeak` function in the `nats_test.go` file
- Modify the `TestGoroutinePanic` function in the `nats_test.go` file
- Modify the `TestReQueueTaskInWorkerBeforeShutdown` function in the `nats_test.go` file
- Modify the `options.go` file
- Change the `WithAddr` function in the `options.go` file

Signed-off-by: Bo-Yi Wu <[email protected]>

Author: appleboy

.github/workflows/go.yml

@@ -33,10 +33,14 @@ jobs:
 
     # Service containers to run with `container-job`
     services:
-      nats:
+      nats01:
         image: nats
         ports:
           - 4222:4222
+      nats02:
+        image: nats
+        ports:
+          - 4223:4222
 
     steps:
       - name: Set up Go ${{ matrix.go }}

nats_test.go

@@ -12,12 +12,13 @@ import (
 	"github.com/golang-queue/queue"
 	"github.com/golang-queue/queue/core"
 	"github.com/golang-queue/queue/job"
+	"github.com/nats-io/nats.go"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/goleak"
 )
 
-var host = "127.0.0.1"
+var host = nats.DefaultURL
 
 func TestMain(m *testing.M) {
 	goleak.VerifyTestMain(m)
@@ -36,7 +37,7 @@ func TestDefaultFlow(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("test"),
 		WithQueue("test"),
 	)
@@ -54,7 +55,7 @@ func TestDefaultFlow(t *testing.T) {
 
 func TestShutdown(t *testing.T) {
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("test"),
 		WithQueue("test"),
 	)
@@ -76,7 +77,7 @@ func TestCustomFuncAndWait(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("test"),
 		WithQueue("test"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
@@ -107,7 +108,7 @@ func TestEnqueueJobAfterShutdown(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithAddr(host + ":4222"),
+		WithAddr(host),
 	)
 	q, err := queue.NewQueue(
 		queue.WithWorker(w),
@@ -129,7 +130,7 @@ func TestJobReachTimeout(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("JobReachTimeout"),
 		WithQueue("test"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
@@ -169,7 +170,7 @@ func TestCancelJobAfterShutdown(t *testing.T) {
 		Message: "test",
 	}
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("CancelJob"),
 		WithQueue("test"),
 		WithLogger(queue.NewLogger()),
@@ -210,7 +211,7 @@ func TestGoroutineLeak(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("GoroutineLeak"),
 		WithQueue("test"),
 		WithLogger(queue.NewEmptyLogger()),
@@ -256,7 +257,7 @@ func TestGoroutinePanic(t *testing.T) {
 		Message: "foo",
 	}
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("GoroutinePanic"),
 		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
 			panic("missing something")
@@ -282,7 +283,7 @@ func TestReQueueTaskInWorkerBeforeShutdown(t *testing.T) {
 		Payload: []byte("foo"),
 	}
 	w := NewWorker(
-		WithAddr(host+":4222"),
+		WithAddr(host),
 		WithSubj("test02"),
 		WithQueue("test02"),
 	)

options.go

@@ -2,9 +2,12 @@ package nats
 
 import (
 	"context"
+	"strings"
 
 	"github.com/golang-queue/queue"
 	"github.com/golang-queue/queue/core"
+
+	"github.com/nats-io/nats.go"
 )
 
 // Option for queue system
@@ -19,9 +22,11 @@ type options struct {
 }
 
 // WithAddr setup the addr of NATS
-func WithAddr(addr string) Option {
+func WithAddr(addrs ...string) Option {
 	return func(w *options) {
-		w.addr = "nats://" + addr
+		if len(addrs) > 0 {
+			w.addr = strings.Join(addrs, ",")
+		}
 	}
 }
 
@@ -55,7 +60,7 @@ func WithLogger(l queue.Logger) Option {
 
 func newOptions(opts ...Option) options {
 	defaultOpts := options{
-		addr:   "127.0.0.1:4222",
+		addr:   nats.DefaultURL,
 		subj:   "foobar",
 		queue:  "foobar",
 		logger: queue.NewLogger(),