polishing the behaviour

Author: wolfy-j

broker/ephemeral/broker.go

@@ -37,7 +37,7 @@ func (b *Broker) Register(pipe *jobs.Pipeline) error {
 		return fmt.Errorf("queue `%s` has already been registered", pipe.Name())
 	}
 
-	b.queues[pipe] = newQueue(pipe.Integer("concurrency", 0))
+	b.queues[pipe] = newQueue(pipe.Integer("maxThreads", 0))
 
 	return nil
 }

broker/ephemeral/queue.go

@@ -8,14 +8,14 @@ import (
 )
 
 type queue struct {
-	active int32
-	st     *jobs.Stat
+	on    int32
+	state *jobs.Stat
 
 	// job pipeline
 	concurPool chan interface{}
 	jobs       chan *entry
 
-	// active operations
+	// on operations
 	muw sync.Mutex
 	wg  sync.WaitGroup
 
@@ -34,56 +34,50 @@ type entry struct {
 }
 
 // create new queue
-func newQueue(concurrency int) *queue {
-	q := &queue{st: &jobs.Stat{}, jobs: make(chan *entry)}
+func newQueue(maxConcur int) *queue {
+	q := &queue{state: &jobs.Stat{}, jobs: make(chan *entry)}
 
-	if concurrency != 0 {
-		q.concurPool = make(chan interface{}, concurrency)
-		for i := 0; i < concurrency; i++ {
+	if maxConcur != 0 {
+		q.concurPool = make(chan interface{}, maxConcur)
+		for i := 0; i < maxConcur; i++ {
 			q.concurPool <- nil
 		}
 	}
 
 	return q
 }
 
-// associate queue with new do pool
-func (q *queue) configure(execPool chan jobs.Handler, err jobs.ErrorHandler) error {
-	q.execPool = execPool
-	q.errHandler = err
-
-	return nil
-}
-
 // serve consumers
 func (q *queue) serve() {
 	q.wait = make(chan interface{})
-	atomic.StoreInt32(&q.active, 1)
+	atomic.StoreInt32(&q.on, 1)
 
 	for {
 		e := q.consume()
 		if e == nil {
+			q.wg.Wait()
 			return
 		}
 
 		if q.concurPool != nil {
 			<-q.concurPool
 		}
 
-		atomic.AddInt64(&q.st.Active, 1)
+		atomic.AddInt64(&q.state.Active, 1)
 		h := <-q.execPool
-		go func(e *entry) {
+
+		go func(h jobs.Handler, e *entry) {
+			defer q.wg.Done()
+
 			q.do(h, e)
-			atomic.AddInt64(&q.st.Active, ^int64(0))
+			atomic.AddInt64(&q.state.Active, ^int64(0))
 
 			q.execPool <- h
 
 			if q.concurPool != nil {
 				q.concurPool <- nil
 			}
-
-			q.wg.Done()
-		}(e)
+		}(h, e)
 	}
 }
 
@@ -107,14 +101,14 @@ func (q *queue) do(h jobs.Handler, e *entry) {
 	err := h(e.id, e.job)
 
 	if err == nil {
-		atomic.AddInt64(&q.st.Queue, ^int64(0))
+		atomic.AddInt64(&q.state.Queue, ^int64(0))
 		return
 	}
 
 	q.errHandler(e.id, e.job, err)
 
 	if !e.job.Options.CanRetry(e.attempt) {
-		atomic.AddInt64(&q.st.Queue, ^int64(0))
+		atomic.AddInt64(&q.state.Queue, ^int64(0))
 		return
 	}
 
@@ -123,35 +117,35 @@ func (q *queue) do(h jobs.Handler, e *entry) {
 
 // stop the queue consuming
 func (q *queue) stop() {
-	if atomic.LoadInt32(&q.active) == 0 {
+	if atomic.LoadInt32(&q.on) == 0 {
 		return
 	}
 
-	atomic.StoreInt32(&q.active, 0)
-
 	close(q.wait)
-	
+
 	q.muw.Lock()
 	q.wg.Wait()
-	q.muw.Unlock()	
+	q.muw.Unlock()
+
+	atomic.StoreInt32(&q.on, 0)
 }
 
 // add job to the queue
 func (q *queue) push(id string, j *jobs.Job, attempt int, delay time.Duration) {
 	if delay == 0 {
-		atomic.AddInt64(&q.st.Queue, 1)
+		atomic.AddInt64(&q.state.Queue, 1)
 		go func() {
 			q.jobs <- &entry{id: id, job: j, attempt: attempt}
 		}()
 
 		return
 	}
 
-	atomic.AddInt64(&q.st.Delayed, 1)
+	atomic.AddInt64(&q.state.Delayed, 1)
 	go func() {
 		time.Sleep(delay)
-		atomic.AddInt64(&q.st.Delayed, ^int64(0))
-		atomic.AddInt64(&q.st.Queue, 1)
+		atomic.AddInt64(&q.state.Delayed, ^int64(0))
+		atomic.AddInt64(&q.state.Queue, 1)
 
 		q.jobs <- &entry{id: id, job: j, attempt: attempt}
 	}()
@@ -160,8 +154,8 @@ func (q *queue) push(id string, j *jobs.Job, attempt int, delay time.Duration) {
 func (q *queue) stat() *jobs.Stat {
 	return &jobs.Stat{
 		InternalName: ":memory:",
-		Queue:        atomic.LoadInt64(&q.st.Queue),
-		Active:       atomic.LoadInt64(&q.st.Active),
-		Delayed:      atomic.LoadInt64(&q.st.Delayed),
+		Queue:        atomic.LoadInt64(&q.state.Queue),
+		Active:       atomic.LoadInt64(&q.state.Active),
+		Delayed:      atomic.LoadInt64(&q.state.Delayed),
 	}
 }