diff options
Diffstat (limited to 'plugins/jobs/drivers/amqp/consumer.go')
| -rw-r--r-- | plugins/jobs/drivers/amqp/consumer.go | 7 |
1 files changed, 6 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 +} |