4 files changed, 21 insertions, 1 deletions
diff --git a/plugins/jobs/drivers/amqp/consumer.go b/plugins/jobs/drivers/amqp/consumer.go
index f3b40ae3..95df02ec 100644
--- a/plugins/jobs/drivers/amqp/consumer.go
+++ b/plugins/jobs/drivers/amqp/consumer.go
@@ -307,7 +307,8 @@ func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
 			Driver:   pipe.Driver(),
 			Queue:    q.Name,
 			Active:   int64(q.Messages),
-			Delayed:  *j.delayed,
+			Delayed:  atomic.LoadInt64(j.delayed),
+			Ready:    ready(atomic.LoadUint32(&j.listeners)),
 		}, nil
 
 	case <-ctx.Done():
@@ -501,3 +502,7 @@ func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
 		return errors.E(op, errors.TimeOut, ctx.Err())
 	}
 }
+
+func ready(r uint32) bool {
+	return r > 0
+}
diff --git a/plugins/jobs/drivers/beanstalk/consumer.go b/plugins/jobs/drivers/beanstalk/consumer.go
index 6a178dff..6323148b 100644
--- a/plugins/jobs/drivers/beanstalk/consumer.go
+++ b/plugins/jobs/drivers/beanstalk/consumer.go
@@ -235,6 +235,7 @@ func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
 		Pipeline: pipe.Name(),
 		Driver:   pipe.Driver(),
 		Queue:    j.tName,
+		Ready:    ready(atomic.LoadUint32(&j.listeners)),
 	}
 
 	// set stat, skip errors (replace with 0)
@@ -353,3 +354,7 @@ func (j *JobConsumer) Resume(_ context.Context, p string) {
 		Start:    time.Now(),
 	})
 }
+
+func ready(r uint32) bool {
+	return r > 0
+}
diff --git a/plugins/jobs/drivers/ephemeral/consumer.go b/plugins/jobs/drivers/ephemeral/consumer.go
index 34778642..f0992cd6 100644
--- a/plugins/jobs/drivers/ephemeral/consumer.go
+++ b/plugins/jobs/drivers/ephemeral/consumer.go
@@ -113,6 +113,7 @@ func (j *JobConsumer) State(_ context.Context) (*jobState.State, error) {
 		Queue:    pipe.Name(),
 		Active:   atomic.LoadInt64(j.active),
 		Delayed:  atomic.LoadInt64(j.delayed),
+		Ready:    ready(atomic.LoadUint32(&j.listeners)),
 	}, nil
 }
 
@@ -267,3 +268,7 @@ func (j *JobConsumer) consume() {
 		}
 	}()
 }
+
+func ready(r uint32) bool {
+	return r > 0
+}
diff --git a/plugins/jobs/drivers/sqs/consumer.go b/plugins/jobs/drivers/sqs/consumer.go
index d957b3eb..5d419b51 100644
--- a/plugins/jobs/drivers/sqs/consumer.go
+++ b/plugins/jobs/drivers/sqs/consumer.go
@@ -285,6 +285,7 @@ func (j *JobConsumer) State(ctx context.Context) (*jobState.State, error) {
 		Pipeline: pipe.Name(),
 		Driver:   pipe.Driver(),
 		Queue:    *j.queueURL,
+		Ready:    ready(atomic.LoadUint32(&j.listeners)),
 	}
 
 	nom, err := strconv.Atoi(attr.Attributes[string(types.QueueAttributeNameApproximateNumberOfMessages)])
@@ -418,3 +419,7 @@ func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
 
 	return nil
 }
+
+func ready(r uint32) bool {
+	return r > 0
+}