SQS initial durability test

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

7 files changed, 32 insertions, 29 deletions

diff --git a/plugins/jobs/drivers/amqp/consumer.go b/plugins/jobs/drivers/amqp/consumer.go
index 32119273..8c55399c 100644
--- a/plugins/jobs/drivers/amqp/consumer.go
+++ b/plugins/jobs/drivers/amqp/consumer.go
@@ -17,7 +17,7 @@ import (
 	"github.com/streadway/amqp"
 )
 
-type JobsConsumer struct {
+type JobConsumer struct {
 	sync.Mutex
 	log logger.Logger
 	pq  priorityqueue.Queue
@@ -50,7 +50,7 @@ type JobsConsumer struct {
 }
 
 // NewAMQPConsumer initializes rabbitmq pipeline
-func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobsConsumer, error) {
+func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
 	const op = errors.Op("new_amqp_consumer")
 	// we need to obtain two parts of the amqp information here.
 	// firs part - address to connect, it is located in the global section under the amqp pluginName
@@ -84,7 +84,7 @@ func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer,
 	globalCfg.InitDefault()
 	// PARSE CONFIGURATION END -------
 
-	jb := &JobsConsumer{
+	jb := &JobConsumer{
 		log:       log,
 		pq:        pq,
 		eh:        e,
@@ -129,7 +129,7 @@ func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer,
 	return jb, nil
 }
 
-func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobsConsumer, error) {
+func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
 	const op = errors.Op("new_amqp_consumer_from_pipeline")
 	// we need to obtain two parts of the amqp information here.
 	// firs part - address to connect, it is located in the global section under the amqp pluginName
@@ -152,7 +152,7 @@ func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, cfg config.Con
 
 	// PARSE CONFIGURATION -------
 
-	jb := &JobsConsumer{
+	jb := &JobConsumer{
 		log:          log,
 		eh:           e,
 		pq:           pq,
@@ -200,7 +200,7 @@ func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, cfg config.Con
 	return jb, nil
 }
 
-func (j *JobsConsumer) Push(job *job.Job) error {
+func (j *JobConsumer) Push(job *job.Job) error {
 	const op = errors.Op("rabbitmq_push")
 	// check if the pipeline registered
 
@@ -298,12 +298,12 @@ func (j *JobsConsumer) Push(job *job.Job) error {
 	return nil
 }
 
-func (j *JobsConsumer) Register(pipeline *pipeline.Pipeline) error {
+func (j *JobConsumer) Register(pipeline *pipeline.Pipeline) error {
 	j.pipeline.Store(pipeline)
 	return nil
 }
 
-func (j *JobsConsumer) Run(p *pipeline.Pipeline) error {
+func (j *JobConsumer) Run(p *pipeline.Pipeline) error {
 	const op = errors.Op("rabbit_consume")
 
 	pipe := j.pipeline.Load().(*pipeline.Pipeline)
@@ -353,7 +353,7 @@ func (j *JobsConsumer) Run(p *pipeline.Pipeline) error {
 	return nil
 }
 
-func (j *JobsConsumer) Pause(p string) {
+func (j *JobConsumer) Pause(p string) {
 	pipe := j.pipeline.Load().(*pipeline.Pipeline)
 	if pipe.Name() != p {
 		j.log.Error("no such pipeline", "requested pause on: ", p)
@@ -391,7 +391,7 @@ func (j *JobsConsumer) Pause(p string) {
 	})
 }
 
-func (j *JobsConsumer) Resume(p string) {
+func (j *JobConsumer) Resume(p string) {
 	pipe := j.pipeline.Load().(*pipeline.Pipeline)
 	if pipe.Name() != p {
 		j.log.Error("no such pipeline", "requested resume on: ", p)
@@ -450,7 +450,7 @@ func (j *JobsConsumer) Resume(p string) {
 	})
 }
 
-func (j *JobsConsumer) Stop() error {
+func (j *JobConsumer) Stop() error {
 	j.stopCh <- struct{}{}
 
 	pipe := j.pipeline.Load().(*pipeline.Pipeline)
diff --git a/plugins/jobs/drivers/amqp/item.go b/plugins/jobs/drivers/amqp/item.go
index 1ebe2751..295ccfd3 100644
--- a/plugins/jobs/drivers/amqp/item.go
+++ b/plugins/jobs/drivers/amqp/item.go
@@ -111,7 +111,7 @@ func (j *Item) Nack() error {
 	return j.NackFunc(false, j.Options.requeue)
 }
 
-func (j *JobsConsumer) fromDelivery(d amqp.Delivery) (*Item, error) {
+func (j *JobConsumer) fromDelivery(d amqp.Delivery) (*Item, error) {
 	const op = errors.Op("from_delivery_convert")
 	item, err := j.unpack(d)
 	if err != nil {
@@ -161,7 +161,7 @@ func pack(id string, j *Item) (amqp.Table, error) {
 }
 
 // unpack restores jobs.Options
-func (j *JobsConsumer) unpack(d amqp.Delivery) (*Item, error) {
+func (j *JobConsumer) unpack(d amqp.Delivery) (*Item, error) {
 	item := &Item{Payload: utils.AsString(d.Body), Options: &Options{
 		multipleAsk: j.multipleAck,
 		requeue:     j.requeueOnFail,
diff --git a/plugins/jobs/drivers/amqp/listener.go b/plugins/jobs/drivers/amqp/listener.go
index 7241c717..8011aa3b 100644
--- a/plugins/jobs/drivers/amqp/listener.go
+++ b/plugins/jobs/drivers/amqp/listener.go
@@ -2,7 +2,7 @@ package amqp
 
 import "github.com/streadway/amqp"
 
-func (j *JobsConsumer) listener(deliv <-chan amqp.Delivery) {
+func (j *JobConsumer) listener(deliv <-chan amqp.Delivery) {
 	go func() {
 		for { //nolint:gosimple
 			select {
diff --git a/plugins/jobs/drivers/amqp/rabbit_init.go b/plugins/jobs/drivers/amqp/rabbit_init.go
index d6b8a708..570498e9 100644
--- a/plugins/jobs/drivers/amqp/rabbit_init.go
+++ b/plugins/jobs/drivers/amqp/rabbit_init.go
@@ -7,7 +7,7 @@ import (
 	"github.com/spiral/roadrunner/v2/pkg/events"
 )
 
-func (j *JobsConsumer) initRabbitMQ() error {
+func (j *JobConsumer) initRabbitMQ() error {
 	const op = errors.Op("jobs_plugin_rmq_init")
 	// Channel opens a unique, concurrent server channel to process the bulk of AMQP
 	// messages.  Any error from methods on this receiver will render the receiver
diff --git a/plugins/jobs/drivers/amqp/redial.go b/plugins/jobs/drivers/amqp/redial.go
index 47afdcb3..d61c75b2 100644
--- a/plugins/jobs/drivers/amqp/redial.go
+++ b/plugins/jobs/drivers/amqp/redial.go
@@ -11,7 +11,7 @@ import (
 )
 
 // redialer used to redial to the rabbitmq in case of the connection interrupts
-func (j *JobsConsumer) redialer() { //nolint:gocognit
+func (j *JobConsumer) redialer() { //nolint:gocognit
 	go func() {
 		const op = errors.Op("rabbitmq_redial")
 
diff --git a/plugins/jobs/drivers/ephemeral/consumer.go b/plugins/jobs/drivers/ephemeral/consumer.go
index ff8f7860..9de64b82 100644
--- a/plugins/jobs/drivers/ephemeral/consumer.go
+++ b/plugins/jobs/drivers/ephemeral/consumer.go
@@ -22,7 +22,7 @@ type Config struct {
 	Prefetch uint64 `mapstructure:"prefetch"`
 }
 
-type JobBroker struct {
+type JobConsumer struct {
 	cfg           *Config
 	log           logger.Logger
 	eh            events.Handler
@@ -35,10 +35,10 @@ type JobBroker struct {
 	stopCh chan struct{}
 }
 
-func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, eh events.Handler, pq priorityqueue.Queue) (*JobBroker, error) {
+func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, eh events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
 	const op = errors.Op("new_ephemeral_pipeline")
 
-	jb := &JobBroker{
+	jb := &JobConsumer{
 		log:              log,
 		pq:               pq,
 		eh:               eh,
@@ -64,8 +64,8 @@ func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, eh
 	return jb, nil
 }
 
-func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, eh events.Handler, pq priorityqueue.Queue) (*JobBroker, error) {
-	jb := &JobBroker{
+func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, eh events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
+	jb := &JobConsumer{
 		log:              log,
 		pq:               pq,
 		eh:               eh,
@@ -82,7 +82,7 @@ func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, eh events.Hand
 	return jb, nil
 }
 
-func (j *JobBroker) Push(jb *job.Job) error {
+func (j *JobConsumer) Push(jb *job.Job) error {
 	const op = errors.Op("ephemeral_push")
 
 	// check if the pipeline registered
@@ -127,7 +127,7 @@ func (j *JobBroker) Push(jb *job.Job) error {
 	return errors.E(op, errors.Errorf("no such pipeline: %s", jb.Options.Pipeline))
 }
 
-func (j *JobBroker) consume() {
+func (j *JobConsumer) consume() {
 	// redirect
 	for {
 		select {
@@ -139,7 +139,7 @@ func (j *JobBroker) consume() {
 	}
 }
 
-func (j *JobBroker) Register(pipeline *pipeline.Pipeline) error {
+func (j *JobConsumer) 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))
@@ -150,7 +150,7 @@ func (j *JobBroker) Register(pipeline *pipeline.Pipeline) error {
 	return nil
 }
 
-func (j *JobBroker) Pause(pipeline string) {
+func (j *JobConsumer) Pause(pipeline string) {
 	if q, ok := j.pipeline.Load(pipeline); ok {
 		if q == true {
 			// mark pipeline as turned off
@@ -166,7 +166,7 @@ func (j *JobBroker) Pause(pipeline string) {
 	})
 }
 
-func (j *JobBroker) Resume(pipeline string) {
+func (j *JobConsumer) Resume(pipeline string) {
 	if q, ok := j.pipeline.Load(pipeline); ok {
 		if q == false {
 			// mark pipeline as turned on
@@ -183,7 +183,7 @@ func (j *JobBroker) Resume(pipeline string) {
 }
 
 // Run is no-op for the ephemeral
-func (j *JobBroker) Run(pipe *pipeline.Pipeline) error {
+func (j *JobConsumer) Run(pipe *pipeline.Pipeline) error {
 	j.eh.Push(events.JobEvent{
 		Event:    events.EventPipeActive,
 		Driver:   pipe.Driver(),
@@ -193,7 +193,7 @@ func (j *JobBroker) Run(pipe *pipeline.Pipeline) error {
 	return nil
 }
 
-func (j *JobBroker) Stop() error {
+func (j *JobConsumer) Stop() error {
 	var pipe string
 	j.pipeline.Range(func(key, _ interface{}) bool {
 		pipe = key.(string)
diff --git a/plugins/jobs/drivers/sqs/consumer.go b/plugins/jobs/drivers/sqs/consumer.go
index f6311715..08a6170e 100644
--- a/plugins/jobs/drivers/sqs/consumer.go
+++ b/plugins/jobs/drivers/sqs/consumer.go
@@ -256,9 +256,12 @@ func (j *JobConsumer) Push(jb *job.Job) error {
 
 	msg := fromJob(jb)
 
+	// 10 seconds deadline to make a request TODO ???
+	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
+	defer cancel()
 	// The new value for the message's visibility timeout (in seconds). Values range: 0
 	// to 43200. Maximum: 12 hours.
-	_, err := j.client.SendMessage(context.Background(), msg.pack(j.queueURL))
+	_, err := j.client.SendMessage(ctx, msg.pack(j.queueURL))
 	if err != nil {
 		return errors.E(op, err)
 	}