Merge pull request #28 from spiral/update_package_versioning

Update package versioning

Author: rustatian

go.mod

@@ -1,5 +1,7 @@
 module github.com/spiral/jobs
 
+go 1.14
+
 require (
 	github.com/aws/aws-sdk-go v1.16.14
 	github.com/beanstalkd/go-beanstalk v0.0.0-20180822062812-53ecdaa3bcfb

v2/broker.go

@@ -0,0 +1,47 @@
+package jobs
+
+// Broker manages set of pipelines and provides ability to push jobs into them.
+type Broker interface {
+	// Register broker pipeline.
+	Register(pipe *Pipeline) error
+
+	// 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
+
+	// Push job into the worker.
+	Push(pipe *Pipeline, j *Job) (string, error)
+
+	// Stat must fetch statistics about given pipeline or return error.
+	Stat(pipe *Pipeline) (stat *Stat, err error)
+}
+
+// EventProvider defines the ability to throw events for the broker.
+type EventProvider interface {
+	// Listen attaches the even listener.
+	Listen(lsn func(event int, ctx interface{}))
+}
+
+// Stat contains information about pipeline.
+type Stat struct {
+	// Pipeline name.
+	Pipeline string
+
+	// Broken is name of associated broker.
+	Broker string
+
+	// InternalName defines internal broker specific pipeline name.
+	InternalName string
+
+	// Consuming indicates that pipeline is pipelines jobs.
+	Consuming bool
+
+	// testQueue defines number of pending jobs.
+	Queue int64
+
+	// Active defines number of jobs which are currently being processed.
+	Active int64
+
+	// Delayed defines number of jobs which are being processed.
+	Delayed int64
+}

v2/broker/amqp/broker.go

@@ -0,0 +1,215 @@
+package amqp
+
+import (
+	"fmt"
+	"github.com/gofrs/uuid"
+	"github.com/spiral/jobs/v2"
+	"sync"
+	"sync/atomic"
+)
+
+// Broker represents AMQP broker.
+type Broker struct {
+	cfg     *Config
+	lsn     func(event int, ctx interface{})
+	publish *chanPool
+	consume *chanPool
+	mu      sync.Mutex
+	wait    chan error
+	stopped chan interface{}
+	queues  map[*jobs.Pipeline]*queue
+}
+
+// Listen attaches server event watcher.
+func (b *Broker) Listen(lsn func(event int, ctx interface{})) {
+	b.lsn = lsn
+}
+
+// Init configures AMQP job broker (always 2 connections).
+func (b *Broker) Init(cfg *Config) (ok bool, err error) {
+	b.cfg = cfg
+	b.queues = make(map[*jobs.Pipeline]*queue)
+
+	return true, nil
+}
+
+// Register broker pipeline.
+func (b *Broker) Register(pipe *jobs.Pipeline) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if _, ok := b.queues[pipe]; ok {
+		return fmt.Errorf("queue `%s` has already been registered", pipe.Name())
+	}
+
+	q, err := newQueue(pipe, b.throw)
+	if err != nil {
+		return err
+	}
+
+	b.queues[pipe] = q
+
+	return nil
+}
+
+// Serve broker pipelines.
+func (b *Broker) Serve() (err error) {
+	b.mu.Lock()
+
+	if b.publish, err = newConn(b.cfg.dial, b.cfg.TimeoutDuration()); err != nil {
+		b.mu.Unlock()
+		return err
+	}
+	defer b.publish.Close()
+
+	if b.consume, err = newConn(b.cfg.dial, b.cfg.TimeoutDuration()); err != nil {
+		b.mu.Unlock()
+		return err
+	}
+	defer b.consume.Close()
+
+	for _, q := range b.queues {
+		err := q.declare(b.publish, q.name, q.name, nil)
+		if err != nil {
+			b.mu.Unlock()
+			return err
+		}
+	}
+
+	for _, q := range b.queues {
+		if q.execPool != nil {
+			go q.serve(b.publish, b.consume)
+		}
+	}
+
+	b.wait = make(chan error)
+	b.stopped = make(chan interface{})
+	defer close(b.stopped)
+
+	b.mu.Unlock()
+
+	b.throw(jobs.EventBrokerReady, b)
+
+	return <-b.wait
+}
+
+// Stop all pipelines.
+func (b *Broker) Stop() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.wait == nil {
+		return
+	}
+
+	for _, q := range b.queues {
+		q.stop()
+	}
+
+	close(b.wait)
+	<-b.stopped
+}
+
+// Consume configures pipeline to be consumed. With execPool to nil to disable consuming. Method can be called before
+// the service is started!
+func (b *Broker) Consume(pipe *jobs.Pipeline, execPool chan jobs.Handler, errHandler jobs.ErrorHandler) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	q, ok := b.queues[pipe]
+	if !ok {
+		return fmt.Errorf("undefined queue `%s`", pipe.Name())
+	}
+
+	q.stop()
+
+	q.execPool = execPool
+	q.errHandler = errHandler
+
+	if b.publish != nil && q.execPool != nil {
+		if q.execPool != nil {
+			go q.serve(b.publish, b.consume)
+		}
+	}
+
+	return nil
+}
+
+// Push job into the worker.
+func (b *Broker) Push(pipe *jobs.Pipeline, j *jobs.Job) (string, error) {
+	if err := b.isServing(); err != nil {
+		return "", err
+	}
+
+	id, err := uuid.NewV4()
+	if err != nil {
+		return "", err
+	}
+
+	q := b.queue(pipe)
+	if q == nil {
+		return "", fmt.Errorf("undefined queue `%s`", pipe.Name())
+	}
+
+	if err := q.publish(b.publish, id.String(), 0, j, j.Options.DelayDuration()); err != nil {
+		return "", err
+	}
+
+	return id.String(), nil
+}
+
+// Stat must fetch statistics about given pipeline or return error.
+func (b *Broker) Stat(pipe *jobs.Pipeline) (stat *jobs.Stat, err error) {
+	if err := b.isServing(); err != nil {
+		return nil, err
+	}
+
+	q := b.queue(pipe)
+	if q == nil {
+		return nil, fmt.Errorf("undefined queue `%s`", pipe.Name())
+	}
+
+	queue, err := q.inspect(b.publish)
+	if err != nil {
+		return nil, err
+	}
+
+	// this the closest approximation we can get for now
+	return &jobs.Stat{
+		InternalName: queue.Name,
+		Queue:        int64(queue.Messages),
+		Active:       int64(atomic.LoadInt32(&q.running)),
+	}, nil
+}
+
+// check if broker is serving
+func (b *Broker) isServing() error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.wait == nil {
+		return fmt.Errorf("broker is not running")
+	}
+
+	return nil
+}
+
+// queue returns queue associated with the pipeline.
+func (b *Broker) queue(pipe *jobs.Pipeline) *queue {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	q, ok := b.queues[pipe]
+	if !ok {
+		return nil
+	}
+
+	return q
+}
+
+// throw handles service, server and pool events.
+func (b *Broker) throw(event int, ctx interface{}) {
+	if b.lsn != nil {
+		b.lsn(event, ctx)
+	}
+}