Close connection in the amqp driver.

bytes.Buffer update in the beanstalk driver

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

1 files changed, 3 insertions, 3 deletions

diff --git a/plugins/amqp/amqpjobs/consumer.go b/plugins/amqp/amqpjobs/consumer.go
index 784a102c..1bfc4b41 100644
--- a/plugins/amqp/amqpjobs/consumer.go
+++ b/plugins/amqp/amqpjobs/consumer.go
@@ -420,17 +420,17 @@ func (c *consumer) Resume(_ context.Context, p string) {
 }
 
 func (c *consumer) Stop(context.Context) error {
-	if atomic.LoadUint32(&c.listeners) > 0 {
-		c.stopCh <- struct{}{}
-	}
+	c.stopCh <- struct{}{}
 
 	pipe := c.pipeline.Load().(*pipeline.Pipeline)
+
 	c.eh.Push(events.JobEvent{
 		Event:    events.EventPipeStopped,
 		Driver:   pipe.Driver(),
 		Pipeline: pipe.Name(),
 		Start:    time.Now(),
 	})
+
 	return nil
 }