SQS driver.

Fix isssues in the AMQP driver.

Signed-off-by: Valery Piashchynski <[email protected]>

3 files changed, 357 insertions, 0 deletions

diff --git a/plugins/jobs/drivers/ephemeral/consumer.go b/plugins/jobs/drivers/ephemeral/consumer.go
new file mode 100644
index 00000000..45ee8083
--- /dev/null
+++ b/plugins/jobs/drivers/ephemeral/consumer.go
@@ -0,0 +1,204 @@
+package ephemeral
+
+import (
+	"sync"
+	"time"
+
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/pkg/events"
+	priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
+	"github.com/spiral/roadrunner/v2/plugins/config"
+	"github.com/spiral/roadrunner/v2/plugins/jobs/job"
+	"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+)
+
+const (
+	pipelineSize string = "pipeline_size"
+)
+
+type Config struct {
+	PipelineSize uint64 `mapstructure:"pipeline_size"`
+}
+
+type JobBroker struct {
+	cfg        *Config
+	log        logger.Logger
+	eh         events.Handler
+	pipeline   sync.Map
+	pq         priorityqueue.Queue
+	localQueue chan *Item
+
+	stopCh chan struct{}
+}
+
+func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, eh events.Handler, pq priorityqueue.Queue) (*JobBroker, error) {
+	const op = errors.Op("new_ephemeral_pipeline")
+
+	jb := &JobBroker{
+		log:    log,
+		pq:     pq,
+		eh:     eh,
+		stopCh: make(chan struct{}, 1),
+	}
+
+	err := cfg.UnmarshalKey(configKey, &jb.cfg)
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	if jb.cfg.PipelineSize == 0 {
+		jb.cfg.PipelineSize = 100_000
+	}
+
+	// initialize a local queue
+	jb.localQueue = make(chan *Item, jb.cfg.PipelineSize)
+
+	// consume from the queue
+	go jb.consume()
+
+	return jb, nil
+}
+
+func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, eh events.Handler, pq priorityqueue.Queue) (*JobBroker, error) {
+	jb := &JobBroker{
+		log:    log,
+		pq:     pq,
+		eh:     eh,
+		stopCh: make(chan struct{}, 1),
+	}
+
+	jb.cfg.PipelineSize = uint64(pipeline.Int(pipelineSize, 100_000))
+
+	// initialize a local queue
+	jb.localQueue = make(chan *Item, jb.cfg.PipelineSize)
+
+	// consume from the queue
+	go jb.consume()
+
+	return jb, nil
+}
+
+func (j *JobBroker) Push(jb *job.Job) error {
+	const op = errors.Op("ephemeral_push")
+
+	// check if the pipeline registered
+	if b, ok := j.pipeline.Load(jb.Options.Pipeline); ok {
+		if !b.(bool) {
+			return errors.E(op, errors.Errorf("pipeline disabled: %s", jb.Options.Pipeline))
+		}
+
+		msg := fromJob(jb)
+		// handle timeouts
+		if msg.Options.Timeout > 0 {
+			go func(jj *job.Job) {
+				time.Sleep(jj.Options.TimeoutDuration())
+
+				// send the item after timeout expired
+				j.localQueue <- msg
+			}(jb)
+
+			return nil
+		}
+
+		// insert to the local, limited pipeline
+		j.localQueue <- msg
+
+		return nil
+	}
+
+	return errors.E(op, errors.Errorf("no such pipeline: %s", jb.Options.Pipeline))
+}
+
+func (j *JobBroker) consume() {
+	// redirect
+	for {
+		select {
+		case item := <-j.localQueue:
+			j.pq.Insert(item)
+		case <-j.stopCh:
+			return
+		}
+	}
+}
+
+func (j *JobBroker) Register(pipeline *pipeline.Pipeline) error {
+	const op = errors.Op("ephemeral_register")
+	if _, ok := j.pipeline.Load(pipeline.Name()); ok {
+		return errors.E(op, errors.Errorf("queue %s has already been registered", pipeline))
+	}
+
+	j.pipeline.Store(pipeline.Name(), true)
+
+	return nil
+}
+
+func (j *JobBroker) Pause(pipeline string) {
+	if q, ok := j.pipeline.Load(pipeline); ok {
+		if q == true {
+			// mark pipeline as turned off
+			j.pipeline.Store(pipeline, false)
+		}
+	}
+
+	j.eh.Push(events.JobEvent{
+		Event:    events.EventPipeStopped,
+		Pipeline: pipeline,
+		Start:    time.Now(),
+		Elapsed:  0,
+	})
+}
+
+func (j *JobBroker) Resume(pipeline string) {
+	if q, ok := j.pipeline.Load(pipeline); ok {
+		if q == false {
+			// mark pipeline as turned off
+			j.pipeline.Store(pipeline, true)
+		}
+	}
+
+	j.eh.Push(events.JobEvent{
+		Event:    events.EventPipeActive,
+		Pipeline: pipeline,
+		Start:    time.Now(),
+		Elapsed:  0,
+	})
+}
+
+func (j *JobBroker) List() []string {
+	out := make([]string, 0, 2)
+
+	j.pipeline.Range(func(key, value interface{}) bool {
+		pipe := key.(string)
+		out = append(out, pipe)
+		return true
+	})
+
+	return out
+}
+
+// Run is no-op for the ephemeral
+func (j *JobBroker) Run(_ *pipeline.Pipeline) error {
+	return nil
+}
+
+func (j *JobBroker) Stop() error {
+	var pipe string
+	j.pipeline.Range(func(key, _ interface{}) bool {
+		pipe = key.(string)
+		j.pipeline.Delete(key)
+		return true
+	})
+
+	// return from the consumer
+	j.stopCh <- struct{}{}
+
+	j.eh.Push(events.JobEvent{
+		Event:    events.EventPipeStopped,
+		Pipeline: pipe,
+		Start:    time.Now(),
+		Elapsed:  0,
+	})
+
+	return nil
+}
diff --git a/plugins/jobs/drivers/ephemeral/item.go b/plugins/jobs/drivers/ephemeral/item.go
new file mode 100644
index 00000000..442533c5
--- /dev/null
+++ b/plugins/jobs/drivers/ephemeral/item.go
@@ -0,0 +1,112 @@
+package ephemeral
+
+import (
+	"time"
+
+	json "github.com/json-iterator/go"
+	"github.com/spiral/roadrunner/v2/plugins/jobs/job"
+	"github.com/spiral/roadrunner/v2/utils"
+)
+
+func fromJob(job *job.Job) *Item {
+	return &Item{
+		Job:     job.Job,
+		Ident:   job.Ident,
+		Payload: job.Payload,
+		Options: &Options{
+			Priority: job.Options.Priority,
+			Pipeline: job.Options.Pipeline,
+			Delay:    job.Options.Delay,
+			Timeout:  job.Options.Timeout,
+		},
+	}
+}
+
+type Item struct {
+	// Job contains name of job broker (usually PHP class).
+	Job string `json:"job"`
+
+	// Ident is unique identifier of the job, should be provided from outside
+	Ident string `json:"id"`
+
+	// Payload is string data (usually JSON) passed to Job broker.
+	Payload string `json:"payload"`
+
+	// Headers with key-values pairs
+	Headers map[string][]string `json:"headers"`
+
+	// Options contains set of PipelineOptions specific to job execution. Can be empty.
+	Options *Options `json:"options,omitempty"`
+}
+
+// Options carry information about how to handle given job.
+type Options struct {
+	// Priority is job priority, default - 10
+	// pointer to distinguish 0 as a priority and nil as priority not set
+	Priority int64 `json:"priority"`
+
+	// Pipeline manually specified pipeline.
+	Pipeline string `json:"pipeline,omitempty"`
+
+	// Delay defines time duration to delay execution for. Defaults to none.
+	Delay int64 `json:"delay,omitempty"`
+
+	// Timeout defines for how broker should wait until treating job are failed. Defaults to 30 min.
+	Timeout int64 `json:"timeout,omitempty"`
+}
+
+// DelayDuration returns delay duration in a form of time.Duration.
+func (o *Options) DelayDuration() time.Duration {
+	return time.Second * time.Duration(o.Delay)
+}
+
+// TimeoutDuration returns timeout duration in a form of time.Duration.
+func (o *Options) TimeoutDuration() time.Duration {
+	if o.Timeout == 0 {
+		return 30 * time.Minute
+	}
+
+	return time.Second * time.Duration(o.Timeout)
+}
+
+func (j *Item) ID() string {
+	return j.Ident
+}
+
+func (j *Item) Priority() int64 {
+	return j.Options.Priority
+}
+
+// Body packs job payload into binary payload.
+func (j *Item) Body() []byte {
+	return utils.AsBytes(j.Payload)
+}
+
+// Context packs job context (job, id) into binary payload.
+func (j *Item) Context() ([]byte, error) {
+	ctx, err := json.Marshal(
+		struct {
+			ID       string              `json:"id"`
+			Job      string              `json:"job"`
+			Headers  map[string][]string `json:"headers"`
+			Timeout  int64               `json:"timeout"`
+			Pipeline string              `json:"pipeline"`
+		}{ID: j.Ident, Job: j.Job, Headers: j.Headers, Timeout: j.Options.Timeout, Pipeline: j.Options.Pipeline},
+	)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return ctx, nil
+}
+
+func (j *Item) Ack() error {
+	// noop for the in-memory
+	return nil
+}
+
+func (j *Item) Nack() error {
+	// noop for the in-memory
+	return nil
+}
diff --git a/plugins/jobs/drivers/ephemeral/plugin.go b/plugins/jobs/drivers/ephemeral/plugin.go
new file mode 100644
index 00000000..28495abb
--- /dev/null
+++ b/plugins/jobs/drivers/ephemeral/plugin.go
@@ -0,0 +1,41 @@
+package ephemeral
+
+import (
+	"github.com/spiral/roadrunner/v2/common/jobs"
+	"github.com/spiral/roadrunner/v2/pkg/events"
+	priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
+	"github.com/spiral/roadrunner/v2/plugins/config"
+	"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+)
+
+const (
+	PluginName string = "ephemeral"
+)
+
+type Plugin struct {
+	log logger.Logger
+	cfg config.Configurer
+}
+
+func (p *Plugin) Init(log logger.Logger, cfg config.Configurer) error {
+	p.log = log
+	p.cfg = cfg
+	return nil
+}
+
+func (p *Plugin) Name() string {
+	return PluginName
+}
+
+func (p *Plugin) Available() {}
+
+// JobsConstruct creates new ephemeral consumer from the configuration
+func (p *Plugin) JobsConstruct(configKey string, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error) {
+	return NewJobBroker(configKey, p.log, p.cfg, e, pq)
+}
+
+// FromPipeline creates new ephemeral consumer from the provided pipeline
+func (p *Plugin) FromPipeline(pipeline *pipeline.Pipeline, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error) {
+	return FromPipeline(pipeline, p.log, e, pq)
+}