Merge remote-tracking branch 'origin/master'

Author: wolfy-j

broker.go

@@ -5,7 +5,7 @@ type Broker interface {
 	// Register broker pipeline.
 	Register(pipe *Pipeline) error
 
-	// Consume configures pipeline to be consumed. With execPool to nil to disable consuming. Method can be called before
+	// Consume configures pipeline to be consumed. With execPool to nil to disable pipelines. Method can be called before
 	// the service is started!
 	Consume(pipe *Pipeline, execPool chan Handler, errHandler ErrorHandler) error
 
@@ -33,7 +33,7 @@ type Stat struct {
 	// InternalName defines internal broker specific pipeline name.
 	InternalName string
 
-	// Consuming indicates that pipeline is consuming jobs.
+	// Consuming indicates that pipeline is pipelines jobs.
 	Consuming bool
 
 	// testQueue defines number of pending jobs.

broker/amqp/queue.go

@@ -261,6 +261,6 @@ func (q *queue) inspect(cp *chanPool) (*amqp.Queue, error) {
 // throw handles service, server and pool events.
 func (q *queue) report(err error) {
 	if err != nil {
-		q.lsn(jobs.EventPipelineError, &jobs.PipelineError{Pipeline: q.pipe, Caused: err})
+		q.lsn(jobs.EventPipeError, &jobs.PipelineError{Pipeline: q.pipe, Caused: err})
 	}
 }

broker/beanstalk/broker.go

@@ -61,7 +61,7 @@ func (b *Broker) Serve() (err error) {
 
 	for _, t := range b.tubes {
 		if t.execPool != nil {
-			go t.serve(connFactory(b.cfg))
+			go t.serve(b.cfg)
 		}
 	}
 

broker/beanstalk/broker_test.go

@@ -1,7 +1,7 @@
 package beanstalk
 
 import (
-	"github.com/kr/beanstalk"
+	"github.com/beanstalkd/go-beanstalk"
 	"github.com/spiral/jobs"
 	"github.com/stretchr/testify/assert"
 	"testing"

broker/beanstalk/tube.go

@@ -245,6 +245,6 @@ func (t *tube) doStat(cn *conn) (stat *jobs.Stat, err error) {
 // report tube specific error
 func (t *tube) report(err error) {
 	if err != nil {
-		t.lsn(jobs.EventPipelineError, &jobs.PipelineError{Pipeline: t.pipe, Caused: err})
+		t.lsn(jobs.EventPipeError, &jobs.PipelineError{Pipeline: t.pipe, Caused: err})
 	}
 }

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),
 	}
 }

broker/ephemeral/queue_test.go

@@ -96,7 +96,7 @@ func (b *Broker) Stop() {
 		q.stop()
 	}
 
-	close(b.wait)
+	b.wait <- nil
 	<-b.stopped
 }
 

broker/sqs/broker.go

@@ -11,8 +11,8 @@ import (
 
 var jobAttributes = []*string{
 	aws.String("rr-job"),
-	aws.String("rr-delay"),
 	aws.String("rr-maxAttempts"),
+	aws.String("rr-delay"),
 	aws.String("rr-timeout"),
 	aws.String("rr-retryDelay"),
 }
@@ -42,6 +42,12 @@ func unpack(msg *sqs.Message) (id string, attempt int, j *jobs.Job, err error) {
 	}
 	attempt, _ = strconv.Atoi(*msg.Attributes["ApproximateReceiveCount"])
 
+	for _, attr := range jobAttributes {
+		if _, ok := msg.MessageAttributes[*attr]; !ok {
+			return "", 0, nil, fmt.Errorf("missing message attribute `%s` (mixed queue?)", *attr)
+		}
+	}
+
 	j = &jobs.Job{
 		Job:     *msg.MessageAttributes["rr-job"].StringValue,
 		Payload: *msg.Body,

broker/sqs/job.go

@@ -261,6 +261,6 @@ func (q *queue) stat(s *sqs.SQS) (stat *jobs.Stat, err error) {
 // throw handles service, server and pool events.
 func (q *queue) report(err error) {
 	if err != nil {
-		q.lsn(jobs.EventPipelineError, &jobs.PipelineError{Pipeline: q.pipe, Caused: err})
+		q.lsn(jobs.EventPipeError, &jobs.PipelineError{Pipeline: q.pipe, Caused: err})
 	}
 }

broker/sqs/queue.go

@@ -45,7 +45,7 @@ func (b *testBroker) Register(pipe *Pipeline) error {
 
 // Serve broker pipelines.
 func (b *testBroker) Serve() error {
-	// start consuming
+	// start pipelines
 	b.mu.Lock()
 	for _, q := range b.queues {
 		if q.execPool != nil {
@@ -72,7 +72,7 @@ func (b *testBroker) Stop() {
 		return
 	}
 
-	// stop all consuming
+	// stop all pipelines
 	for _, q := range b.queues {
 		q.stop()
 	}
@@ -81,7 +81,7 @@ func (b *testBroker) Stop() {
 	<-b.stopped
 }
 
-// Consume configures pipeline to be consumed. With execPool to nil to disable consuming. Method can be called before
+// Consume configures pipeline to be consumed. With execPool to nil to disable pipelines. Method can be called before
 // the service is started!
 func (b *testBroker) Consume(pipe *Pipeline, execPool chan Handler, errHandler ErrorHandler) error {
 	b.mu.Lock()
@@ -180,7 +180,7 @@ type testQueue struct {
 	// job pipeline
 	jobs chan *entry
 
-	// active operations
+	// pipelines operations
 	muw sync.Mutex
 	wg  sync.WaitGroup
 
@@ -267,7 +267,7 @@ func (q *testQueue) do(h Handler, e *entry) {
 	q.push(e.id, e.job, e.attempt+1, e.job.Options.RetryDuration())
 }
 
-// stop the testQueue consuming
+// stop the testQueue pipelines
 func (q *testQueue) stop() {
 	if atomic.LoadInt32(&q.active) == 0 {
 		return

broker_test.go

@@ -52,23 +52,23 @@ func (s *debugger) listener(event int, ctx interface{}) {
 	}
 
 	switch event {
-	case jobs.EventPushComplete:
+	case jobs.EventPushOK:
 		e := ctx.(*jobs.JobEvent)
 		s.logger.Info(util.Sprintf(
-			"job.<magenta+h>PUSH</reset> <white+hb>%s</reset> <gray+hb>%s</reset>",
+			"job.<yellow+h>PUSH</reset> <white+hb>%s</reset> <gray+hb>%s</reset>",
 			e.Job.Job,
 			e.ID,
 		))
 
-	// case jobs.EventJobReceived:
+	// case jobs.EventJobStart:
 	// 	e := ctx.(*jobs.JobEvent)
 	// 	s.logger.Info(util.Sprintf(
 	// 		"job.<cyan+h>RECV</reset> <white+hb>%s</reset> <gray+hb>%s</reset>",
 	// 		e.Job.Job,
 	// 		e.ID,
 	// 	))
 
-	case jobs.EventJobComplete:
+	case jobs.EventJobOK:
 		e := ctx.(*jobs.JobEvent)
 		s.logger.Info(util.Sprintf(
 			"job.<green+h>DONE</reset> <green>%s</reset> <gray+hb>%s</reset> %s",
@@ -95,39 +95,39 @@ func (s *debugger) listener(event int, ctx interface{}) {
 			e.Error(),
 		))
 
-	// case jobs.EventPipelineConsume:
+	// case jobs.EventPipeConsume:
 	// 	e := ctx.(*jobs.Pipeline)
 	// 	s.logger.Info(util.Sprintf(
 	// 		"[%s]: resuming {<yellow+hb>%s</reset>}",
 	// 		e.Broker(),
 	// 		e.Name(),
 	// 	))
 
-	case jobs.EventPipelineConsuming:
+	case jobs.EventPipeActive:
 		e := ctx.(*jobs.Pipeline)
 		s.logger.Info(util.Sprintf(
 			"[%s]: resumed {<green+hb>%s</reset>}",
 			e.Broker(),
 			e.Name(),
 		))
 
-	// case jobs.EventPipelineStop:
+	// case jobs.EventPipeStop:
 	// 	e := ctx.(*jobs.Pipeline)
 	// 	s.logger.Info(util.Sprintf(
 	// 		"[%s]: stopping {<yellow+hb>%s</reset>}",
 	// 		e.Broker(),
 	// 		e.Name(),
 	// 	))
 
-	case jobs.EventPipelineStopped:
+	case jobs.EventPipeStopped:
 		e := ctx.(*jobs.Pipeline)
 		s.logger.Debugf(util.Sprintf(
 			"[%s]: stopped {<magenta+hb>%s</reset>}",
 			e.Broker(),
 			e.Name(),
 		))
 
-	case jobs.EventPipelineError:
+	case jobs.EventPipeError:
 		e := ctx.(*jobs.PipelineError)
 		s.logger.Error(util.Sprintf(
 			"[%s]: <red>{%s}</reset> <red+hb>%s</reset>",