diff options
Diffstat (limited to 'plugins/jobs/drivers/amqp')
| -rw-r--r-- | plugins/jobs/drivers/amqp/consumer.go | 7 |
1 files changed, 5 insertions, 2 deletions
diff --git a/plugins/jobs/drivers/amqp/consumer.go b/plugins/jobs/drivers/amqp/consumer.go index d592a17a..f6442b42 100644 --- a/plugins/jobs/drivers/amqp/consumer.go +++ b/plugins/jobs/drivers/amqp/consumer.go @@ -384,7 +384,7 @@ func (j *JobsConsumer) Pause(p string) { } j.eh.Push(events.JobEvent{ - Event: events.EventPipeStopped, + Event: events.EventPipePaused, Driver: pipe.Driver(), Pipeline: pipe.Name(), Start: time.Now(), @@ -404,7 +404,7 @@ func (j *JobsConsumer) Resume(p string) { l := atomic.LoadUint32(&j.listeners) // no active listeners if l == 1 { - j.log.Warn("sqs listener already in the active state") + j.log.Warn("amqp listener already in the active state") return } @@ -439,6 +439,9 @@ func (j *JobsConsumer) Resume(p string) { // run listener j.listener(deliv) + // increase number of listeners + atomic.AddUint32(&j.listeners, 1) + j.eh.Push(events.JobEvent{ Event: events.EventPipeActive, Driver: pipe.Driver(), |