chore(worker): remove metrics and add new method (#7)

Author: appleboy

go.mod

@@ -3,7 +3,7 @@ module github.com/golang-queue/nats
 go 1.16
 
 require (
-	github.com/golang-queue/queue v0.0.11
+	github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/nats-io/nats-server/v2 v2.7.1 // indirect
 	github.com/nats-io/nats.go v1.13.1-0.20220121202836-972a071d373d

go.sum

@@ -2,6 +2,8 @@ github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/golang-queue/queue v0.0.11 h1:qaE7rOYUh+11NcO6U9zZ7NazglIwEp0j7Dn0sSoavcI=
 github.com/golang-queue/queue v0.0.11/go.mod h1:ku8iyjYffqYY6Duts+xl+QYfN3/KDK4MEvXMZUkHyio=
+github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747 h1:uNTbCoWORAcna89KcKgP22WFGv5fsij05e70DCnLrUU=
+github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747/go.mod h1:ku8iyjYffqYY6Duts+xl+QYfN3/KDK4MEvXMZUkHyio=
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
 github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
 github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=

nats.go

@@ -23,19 +23,6 @@ type Worker struct {
 	opts     options
 }
 
-func (w *Worker) incBusyWorker() {
-	w.opts.metric.IncBusyWorker()
-}
-
-func (w *Worker) decBusyWorker() {
-	w.opts.metric.DecBusyWorker()
-}
-
-// BusyWorkers return count of busy workers currently.
-func (w *Worker) BusyWorkers() uint64 {
-	return w.opts.metric.BusyWorkers()
-}
-
 // NewWorker for struc
 func NewWorker(opts ...Option) *Worker {
 	var err error
@@ -68,10 +55,8 @@ func (w *Worker) handle(job queue.Job) error {
 	panicChan := make(chan interface{}, 1)
 	startTime := time.Now()
 	ctx, cancel := context.WithTimeout(context.Background(), job.Timeout)
-	w.incBusyWorker()
 	defer func() {
 		cancel()
-		w.decBusyWorker()
 	}()
 
 	// run the job
@@ -112,7 +97,7 @@ func (w *Worker) handle(job queue.Job) error {
 }
 
 // Run start the worker
-func (w *Worker) Run() error {
+func (w *Worker) Run(task queue.QueuedMessage) error {
 	wg := &sync.WaitGroup{}
 	panicChan := make(chan interface{}, 1)
 	_, err := w.client.QueueSubscribe(w.opts.subj, w.opts.queue, func(m *nats.Msg) {
@@ -184,3 +169,8 @@ func (w *Worker) Queue(job queue.QueuedMessage) error {
 
 	return nil
 }
+
+// Request a new task
+func (w *Worker) Request() (queue.QueuedMessage, error) {
+	return nil, nil
+}

nats_test.go

@@ -124,27 +124,6 @@ func TestEnqueueJobAfterShutdown(t *testing.T) {
 	q.Wait()
 }
 
-func TestWorkerNumAfterShutdown(t *testing.T) {
-	w := NewWorker(
-		WithAddr(host + ":4222"),
-	)
-	q, err := queue.NewQueue(
-		queue.WithWorker(w),
-		queue.WithWorkerCount(2),
-	)
-	assert.NoError(t, err)
-	q.Start()
-	q.Start()
-	time.Sleep(100 * time.Millisecond)
-	assert.Equal(t, 4, q.Workers())
-	q.Shutdown()
-	q.Wait()
-	assert.Equal(t, 0, q.Workers())
-	q.Start()
-	q.Start()
-	assert.Equal(t, 0, q.Workers())
-}
-
 func TestJobReachTimeout(t *testing.T) {
 	m := mockMessage{
 		Message: "foo",
@@ -297,7 +276,7 @@ func TestGoroutinePanic(t *testing.T) {
 func TestHandleTimeout(t *testing.T) {
 	job := queue.Job{
 		Timeout: 100 * time.Millisecond,
-		Body:    []byte("foo"),
+		Payload: []byte("foo"),
 	}
 	w := NewWorker(
 		WithAddr(host+":4222"),
@@ -314,7 +293,7 @@ func TestHandleTimeout(t *testing.T) {
 
 	job = queue.Job{
 		Timeout: 150 * time.Millisecond,
-		Body:    []byte("foo"),
+		Payload: []byte("foo"),
 	}
 
 	w = NewWorker(
@@ -340,7 +319,7 @@ func TestHandleTimeout(t *testing.T) {
 func TestJobComplete(t *testing.T) {
 	job := queue.Job{
 		Timeout: 100 * time.Millisecond,
-		Body:    []byte("foo"),
+		Payload: []byte("foo"),
 	}
 	w := NewWorker(
 		WithAddr(host+":4222"),
@@ -356,7 +335,7 @@ func TestJobComplete(t *testing.T) {
 
 	job = queue.Job{
 		Timeout: 250 * time.Millisecond,
-		Body:    []byte("foo"),
+		Payload: []byte("foo"),
 	}
 
 	w = NewWorker(
@@ -378,32 +357,3 @@ func TestJobComplete(t *testing.T) {
 	assert.Error(t, err)
 	assert.Equal(t, errors.New("job completed"), err)
 }
-
-func TestBusyWorkerCount(t *testing.T) {
-	job := queue.Job{
-		Timeout: 500 * time.Millisecond,
-		Body:    []byte("foo"),
-	}
-
-	w := NewWorker(
-		WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
-			time.Sleep(200 * time.Millisecond)
-			return nil
-		}),
-	)
-
-	assert.Equal(t, uint64(0), w.BusyWorkers())
-	go func() {
-		assert.NoError(t, w.handle(job))
-	}()
-	go func() {
-		assert.NoError(t, w.handle(job))
-	}()
-
-	time.Sleep(50 * time.Millisecond)
-	assert.Equal(t, uint64(2), w.BusyWorkers())
-	time.Sleep(200 * time.Millisecond)
-	assert.Equal(t, uint64(0), w.BusyWorkers())
-
-	assert.NoError(t, w.Shutdown())
-}

options.go

@@ -12,7 +12,6 @@ type Option func(*options)
 type options struct {
 	runFunc func(context.Context, queue.QueuedMessage) error
 	logger  queue.Logger
-	metric  queue.Metric
 	addr    string
 	subj    string
 	queue   string
@@ -53,13 +52,6 @@ func WithLogger(l queue.Logger) Option {
 	}
 }
 
-// WithMetric set custom Metric
-func WithMetric(m queue.Metric) Option {
-	return func(w *options) {
-		w.metric = m
-	}
-}
-
 func newOptions(opts ...Option) options {
 	defaultOpts := options{
 		addr:  "127.0.0.1:4222",
@@ -68,7 +60,6 @@ func newOptions(opts ...Option) options {
 		runFunc: func(context.Context, queue.QueuedMessage) error {
 			return nil
 		},
-		metric: queue.NewMetric(),
 	}
 
 	// Loop through each option