Remove unneeded options, complete tests for the ephemeral, update proto

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

1 files changed, 1 insertions, 24 deletions

diff --git a/plugins/jobs/drivers/amqp/item.go b/plugins/jobs/drivers/amqp/item.go
index 9b9625b0..5990d137 100644
--- a/plugins/jobs/drivers/amqp/item.go
+++ b/plugins/jobs/drivers/amqp/item.go
@@ -41,9 +41,6 @@ type Options struct {
 	// Delay defines time duration to delay execution for. Defaults to none.
 	Delay int64 `json:"delay,omitempty"`
 
-	// Reserve defines for how broker should wait until treating job are failed. Defaults to 30 min.
-	Timeout int64 `json:"timeout,omitempty"`
-
 	// private
 	// Ack delegates an acknowledgement through the Acknowledger interface that the client or server has finished work on a delivery
 	ack func(multiply bool) error
@@ -66,15 +63,6 @@ func (o *Options) DelayDuration() time.Duration {
 	return time.Second * time.Duration(o.Delay)
 }
 
-// TimeoutDuration returns timeout duration in a form of time.Duration.
-func (o *Options) TimeoutDuration() time.Duration {
-	if o.Timeout == 0 {
-		return 30 * time.Minute
-	}
-
-	return time.Second * time.Duration(o.Timeout)
-}
-
 func (i *Item) ID() string {
 	return i.Ident
 }
@@ -96,9 +84,8 @@ func (i *Item) Context() ([]byte, error) {
 			ID       string              `json:"id"`
 			Job      string              `json:"job"`
 			Headers  map[string][]string `json:"headers"`
-			Timeout  int64               `json:"timeout"`
 			Pipeline string              `json:"pipeline"`
-		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Timeout: i.Options.Timeout, Pipeline: i.Options.Pipeline},
+		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Pipeline: i.Options.Pipeline},
 	)
 
 	if err != nil {
@@ -141,10 +128,6 @@ func (i *Item) Requeue(headers map[string][]string, delay int64) error {
 	return nil
 }
 
-func (i *Item) Recycle() {
-	i.Options = nil
-}
-
 // fromDelivery converts amqp.Delivery into an Item which will be pushed to the PQ
 func (j *JobConsumer) fromDelivery(d amqp.Delivery) (*Item, error) {
 	const op = errors.Op("from_delivery_convert")
@@ -179,7 +162,6 @@ func fromJob(job *job.Job) *Item {
 			Priority: job.Options.Priority,
 			Pipeline: job.Options.Pipeline,
 			Delay:    job.Options.Delay,
-			Timeout:  job.Options.Timeout,
 		},
 	}
 }
@@ -195,7 +177,6 @@ func pack(id string, j *Item) (amqp.Table, error) {
 		job.RRJob:      j.Job,
 		job.RRPipeline: j.Options.Pipeline,
 		job.RRHeaders:  headers,
-		job.RRTimeout:  j.Options.Timeout,
 		job.RRDelay:    j.Options.Delay,
 		job.RRPriority: j.Options.Priority,
 	}, nil
@@ -232,10 +213,6 @@ func (j *JobConsumer) unpack(d amqp.Delivery) (*Item, error) {
 		}
 	}
 
-	if _, ok := d.Headers[job.RRTimeout].(int64); ok {
-		item.Options.Timeout = d.Headers[job.RRTimeout].(int64)
-	}
-
 	if _, ok := d.Headers[job.RRDelay].(int64); ok {
 		item.Options.Delay = d.Headers[job.RRDelay].(int64)
 	}