chore(comsumer): remove Usage and Capacity

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

Author: appleboy

consumer.go

@@ -126,16 +126,6 @@ func (s *Consumer) Shutdown() error {
 	return nil
 }
 
-// Capacity for channel
-func (s *Consumer) Capacity() int {
-	return cap(s.taskQueue)
-}
-
-// Usage for count of channel usage
-func (s *Consumer) Usage() int {
-	return len(s.taskQueue)
-}
-
 // Queue send notification to queue
 func (s *Consumer) Queue(task QueuedMessage) error {
 	if atomic.LoadInt32(&s.stopFlag) == 1 {
@@ -155,7 +145,7 @@ func (s *Consumer) Request() (QueuedMessage, error) {
 	case task := <-s.taskQueue:
 		return task, nil
 	default:
-		return nil, errors.New("no message in queue")
+		return nil, errors.New("no task in queue")
 	}
 }
 

consumer_test.go

@@ -12,26 +12,12 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestQueueUsage(t *testing.T) {
-	w := NewConsumer()
-	assert.Equal(t, defaultQueueSize, w.Capacity())
-	assert.Equal(t, 0, w.Usage())
-
-	assert.NoError(t, w.Queue(&mockMessage{}))
-	assert.Equal(t, 1, w.Usage())
-}
-
 func TestMaxCapacity(t *testing.T) {
 	w := NewConsumer(WithQueueSize(2))
-	assert.Equal(t, 2, w.Capacity())
-	assert.Equal(t, 0, w.Usage())
 
 	assert.NoError(t, w.Queue(&mockMessage{}))
-	assert.Equal(t, 1, w.Usage())
 	assert.NoError(t, w.Queue(&mockMessage{}))
-	assert.Equal(t, 2, w.Usage())
 	assert.Error(t, w.Queue(&mockMessage{}))
-	assert.Equal(t, 2, w.Usage())
 
 	err := w.Queue(&mockMessage{})
 	assert.Equal(t, errMaxCapacity, err)

queue.go

@@ -272,30 +272,32 @@ func (q *Queue) start() {
 			case task = <-tasks:
 				// queue task before shutdown the service
 				if err := q.worker.Queue(task); err != nil {
-					q.logger.Errorf("can't re-queue message: %v", err)
+					q.logger.Errorf("can't re-queue task: %v", err)
 				}
 			default:
 			}
 			return
 		}
 
 		// check worker number
+		q.Lock()
 		if q.BusyWorkers() < q.workerCount {
 			q.schedule()
 		}
+		q.Unlock()
 
 		// get worker to execute new task
 		select {
 		case <-q.quit:
 			if err := q.worker.Queue(task); err != nil {
-				q.logger.Errorf("can't re-queue message: %v", err)
+				q.logger.Errorf("can't re-queue task: %v", err)
 			}
 			return
 		case <-q.ready:
 			select {
 			case <-q.quit:
 				if err := q.worker.Queue(task); err != nil {
-					q.logger.Errorf("can't re-queue message: %v", err)
+					q.logger.Errorf("can't re-queue task: %v", err)
 				}
 				return
 			default: