Update beanstalk options. Fix tube priority (correctly pass to the Put

method).

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

2 files changed, 9 insertions, 8 deletions

diff --git a/plugins/jobs/drivers/beanstalk/config.go b/plugins/jobs/drivers/beanstalk/config.go
index 6a8bda1d..c1059d1e 100644
--- a/plugins/jobs/drivers/beanstalk/config.go
+++ b/plugins/jobs/drivers/beanstalk/config.go
@@ -2,6 +2,12 @@ package beanstalk
 
 import "time"
 
+const (
+	tubePriority   string = "tube_priority"
+	tube           string = "tube"
+	reserveTimeout string = "reserve_timeout"
+)
+
 type GlobalCfg struct {
 	Addr    string        `mapstructure:"addr"`
 	Timeout time.Duration `mapstructure:"timeout"`
@@ -26,7 +32,7 @@ type Config struct {
 
 func (c *Config) InitDefault() {
 	if c.Tube == "" {
-		c.Tube = "default"
+		c.Tube = "default-" + time.Now().String()
 	}
 
 	if c.ReserveTimeout == 0 {
diff --git a/plugins/jobs/drivers/beanstalk/consumer.go b/plugins/jobs/drivers/beanstalk/consumer.go
index 32ce6ef7..8bd11a58 100644
--- a/plugins/jobs/drivers/beanstalk/consumer.go
+++ b/plugins/jobs/drivers/beanstalk/consumer.go
@@ -96,11 +96,6 @@ func NewBeanstalkConsumer(configKey string, log logger.Logger, cfg config.Config
 func FromPipeline(pipe *pipeline.Pipeline, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
 	const op = errors.Op("new_beanstalk_consumer")
 
-	const (
-		tube           string = "tube"
-		reserveTimeout string = "reserve_timeout"
-	)
-
 	// PARSE CONFIGURATION -------
 	var globalCfg GlobalCfg
 
@@ -134,7 +129,7 @@ func FromPipeline(pipe *pipeline.Pipeline, log logger.Logger, cfg config.Configu
 		tout:           globalCfg.Timeout,
 		tName:          pipe.String(tube, "default"),
 		reserveTimeout: time.Second * time.Duration(pipe.Int(reserveTimeout, 5)),
-		tubePriority:   uint32(pipe.Int(tube, 10)),
+		tubePriority:   uint32(pipe.Int(tubePriority, 0)),
 		priority:       pipe.Priority(),
 
 		// buffered with two because jobs root plugin can call Stop at the same time as Pause
@@ -178,7 +173,7 @@ func (j *JobConsumer) Push(jb *job.Job) error {
 	// <ttr> seconds, the job will time out and the server will release the job.
 	//	The minimum ttr is 1. If the client sends 0, the server will silently
 	// increase the ttr to 1. Maximum ttr is 2**32-1.
-	id, err := j.pool.Put(bb.Bytes(), 0, item.Options.DelayDuration(), item.Options.TimeoutDuration())
+	id, err := j.pool.Put(bb.Bytes(), j.tubePriority, item.Options.DelayDuration(), item.Options.TimeoutDuration())
 	if err != nil {
 		errD := j.pool.Delete(id)
 		if errD != nil {