4 files changed, 102 insertions, 63 deletions
diff --git a/plugins/jobs/drivers/amqp/consumer.go b/plugins/jobs/drivers/amqp/consumer.go
index 429953e1..b89cdc82 100644
--- a/plugins/jobs/drivers/amqp/consumer.go
+++ b/plugins/jobs/drivers/amqp/consumer.go
@@ -12,6 +12,7 @@ import (
 	"github.com/spiral/errors"
 	"github.com/spiral/roadrunner/v2/pkg/events"
 	priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
+	jobState "github.com/spiral/roadrunner/v2/pkg/state/job"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/job"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -361,6 +362,29 @@ func (j *JobConsumer) Run(_ context.Context, p *pipeline.Pipeline) error {
 	return nil
 }
 
+func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
+	const op = errors.Op("amqp_driver_state")
+	select {
+	case pch := <-j.publishChan:
+		defer func() {
+			j.publishChan <- pch
+		}()
+
+		q, err := pch.QueueInspect(j.queue)
+		if err != nil {
+			return nil, errors.E(op, err)
+		}
+
+		return &jobState.State{
+			Queue:  q.Name,
+			Active: int64(q.Messages),
+		}, nil
+
+	case <-ctx.Done():
+		return nil, errors.E(op, errors.TimeOut, ctx.Err())
+	}
+}
+
 func (j *JobConsumer) Pause(_ context.Context, p string) {
 	pipe := j.pipeline.Load().(*pipeline.Pipeline)
 	if pipe.Name() != p {
diff --git a/plugins/jobs/drivers/beanstalk/consumer.go b/plugins/jobs/drivers/beanstalk/consumer.go
index eaf99be1..7e81e6d9 100644
--- a/plugins/jobs/drivers/beanstalk/consumer.go
+++ b/plugins/jobs/drivers/beanstalk/consumer.go
@@ -10,6 +10,7 @@ import (
 	"github.com/spiral/errors"
 	"github.com/spiral/roadrunner/v2/pkg/events"
 	priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
+	jobState "github.com/spiral/roadrunner/v2/pkg/state/job"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/job"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -213,12 +214,16 @@ func (j *JobConsumer) handleItem(ctx context.Context, item *Item) error {
 	return nil
 }
 
-func (j *JobConsumer) Register(ctx context.Context, p *pipeline.Pipeline) error {
+func (j *JobConsumer) Register(_ context.Context, p *pipeline.Pipeline) error {
 	// register the pipeline
 	j.pipeline.Store(p)
 	return nil
 }
 
+func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
+	return nil, nil
+}
+
 func (j *JobConsumer) Run(_ context.Context, p *pipeline.Pipeline) error {
 	const op = errors.Op("beanstalk_run")
 	// check if the pipeline registered
diff --git a/plugins/jobs/drivers/ephemeral/consumer.go b/plugins/jobs/drivers/ephemeral/consumer.go
index 95ad6ecd..d801b7b4 100644
--- a/plugins/jobs/drivers/ephemeral/consumer.go
+++ b/plugins/jobs/drivers/ephemeral/consumer.go
@@ -9,6 +9,7 @@ import (
 	"github.com/spiral/errors"
 	"github.com/spiral/roadrunner/v2/pkg/events"
 	priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
+	jobState "github.com/spiral/roadrunner/v2/pkg/state/job"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/job"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -106,57 +107,8 @@ func (j *JobConsumer) Push(ctx context.Context, jb *job.Job) error {
 	return nil
 }
 
-func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
-	const op = errors.Op("ephemeral_handle_request")
-	// handle timeouts
-	// theoretically, some bad user may send millions requests with a delay and produce a billion (for example)
-	// goroutines here. We should limit goroutines here.
-	if msg.Options.Delay > 0 {
-		// if we have 1000 goroutines waiting on the delay - reject 1001
-		if atomic.LoadUint64(&j.goroutines) >= goroutinesMax {
-			return errors.E(op, errors.Str("max concurrency number reached"))
-		}
-
-		go func(jj *Item) {
-			atomic.AddUint64(&j.goroutines, 1)
-			time.Sleep(jj.Options.DelayDuration())
-
-			// send the item after timeout expired
-			j.localPrefetch <- jj
-
-			atomic.AddUint64(&j.goroutines, ^uint64(0))
-		}(msg)
-
-		return nil
-	}
-
-	// insert to the local, limited pipeline
-	select {
-	case j.localPrefetch <- msg:
-		return nil
-	case <-ctx.Done():
-		return errors.E(op, errors.Errorf("local pipeline is full, consider to increase prefetch number, current limit: %d, context error: %v", j.cfg.Prefetch, ctx.Err()))
-	}
-}
-
-func (j *JobConsumer) consume() {
-	// redirect
-	for {
-		select {
-		case item, ok := <-j.localPrefetch:
-			if !ok {
-				j.log.Warn("ephemeral local prefetch queue was closed")
-				return
-			}
-
-			// set requeue channel
-			item.Options.requeueFn = j.handleItem
-
-			j.pq.Insert(item)
-		case <-j.stopCh:
-			return
-		}
-	}
+func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
+	return nil, nil
 }
 
 func (j *JobConsumer) Register(_ context.Context, pipeline *pipeline.Pipeline) error {
@@ -242,3 +194,56 @@ func (j *JobConsumer) Stop(ctx context.Context) error {
 		return errors.E(op, ctx.Err())
 	}
 }
+
+func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
+	const op = errors.Op("ephemeral_handle_request")
+	// handle timeouts
+	// theoretically, some bad user may send millions requests with a delay and produce a billion (for example)
+	// goroutines here. We should limit goroutines here.
+	if msg.Options.Delay > 0 {
+		// if we have 1000 goroutines waiting on the delay - reject 1001
+		if atomic.LoadUint64(&j.goroutines) >= goroutinesMax {
+			return errors.E(op, errors.Str("max concurrency number reached"))
+		}
+
+		go func(jj *Item) {
+			atomic.AddUint64(&j.goroutines, 1)
+			time.Sleep(jj.Options.DelayDuration())
+
+			// send the item after timeout expired
+			j.localPrefetch <- jj
+
+			atomic.AddUint64(&j.goroutines, ^uint64(0))
+		}(msg)
+
+		return nil
+	}
+
+	// insert to the local, limited pipeline
+	select {
+	case j.localPrefetch <- msg:
+		return nil
+	case <-ctx.Done():
+		return errors.E(op, errors.Errorf("local pipeline is full, consider to increase prefetch number, current limit: %d, context error: %v", j.cfg.Prefetch, ctx.Err()))
+	}
+}
+
+func (j *JobConsumer) consume() {
+	// redirect
+	for {
+		select {
+		case item, ok := <-j.localPrefetch:
+			if !ok {
+				j.log.Warn("ephemeral local prefetch queue was closed")
+				return
+			}
+
+			// set requeue channel
+			item.Options.requeueFn = j.handleItem
+
+			j.pq.Insert(item)
+		case <-j.stopCh:
+			return
+		}
+	}
+}
diff --git a/plugins/jobs/drivers/sqs/consumer.go b/plugins/jobs/drivers/sqs/consumer.go
index 9ce37543..4fb684f8 100644
--- a/plugins/jobs/drivers/sqs/consumer.go
+++ b/plugins/jobs/drivers/sqs/consumer.go
@@ -15,6 +15,7 @@ import (
 	"github.com/spiral/errors"
 	"github.com/spiral/roadrunner/v2/pkg/events"
 	priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
+	jobState "github.com/spiral/roadrunner/v2/pkg/state/job"
 	cfgPlugin "github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/job"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -261,17 +262,8 @@ func (j *JobConsumer) Push(ctx context.Context, jb *job.Job) error {
 	return nil
 }
 
-func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
-	d, err := msg.pack(j.queueURL)
-	if err != nil {
-		return err
-	}
-	_, err = j.client.SendMessage(ctx, d)
-	if err != nil {
-		return err
-	}
-
-	return nil
+func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
+	return nil, nil
 }
 
 func (j *JobConsumer) Register(_ context.Context, p *pipeline.Pipeline) error {
@@ -374,3 +366,16 @@ func (j *JobConsumer) Resume(_ context.Context, p string) {
 		Start:    time.Now(),
 	})
 }
+
+func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
+	d, err := msg.pack(j.queueURL)
+	if err != nil {
+		return err
+	}
+	_, err = j.client.SendMessage(ctx, d)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}