disque

Documentation

Index

• type Disque
  • func NewDisque(servers []string, cycle int) *Disque
  • func (d *Disque) Ack(jobId string) (err error)
  • func (d *Disque) Close()
  • func (d *Disque) Delete(jobId string) (err error)
  • func (d *Disque) Fetch(queueName string, timeout time.Duration) (job *Job, err error)
  • func (d *Disque) FetchMultiple(queueName string, count int, timeout time.Duration) (jobs []*Job, err error)
  • func (d *Disque) GetJobDetails(jobId string) (jobDetails *JobDetails, err error)
  • func (d *Disque) Initialize() (err error)
  • func (d *Disque) Nack(jobId string) (err error)
  • func (d *Disque) Push(queueName string, job string, timeout time.Duration) (jobId string, err error)
  • func (d *Disque) PushWithOptions(queueName string, job string, timeout time.Duration, options map[string]string) (jobId string, err error)
  • func (d *Disque) QueueLength(queueName string) (queueLength int, err error)
• type DisquePool
  • func NewDisquePool(servers []string, cycle int, capacity int, maxCapacity int, ...) (d *DisquePool)
  • func (d *DisquePool) Close()
  • func (d *DisquePool) Get(ctx context.Context) (conn *Disque, err error)
  • func (d *DisquePool) IsClosed() (closed bool)
  • func (d *DisquePool) Put(conn *Disque)
  • func (d *DisquePool) SetCapacity(capacity int)
• type Job
• type JobDetails

Types

type Disque

type Disque struct {
	// contains filtered or unexported fields
}

func NewDisque

func NewDisque(servers []string, cycle int) *Disque

Instantiate a new Disque connection

func (*Disque) Ack

func (d *Disque) Ack(jobId string) (err error)

Acknowledge receipt and processing of a message

func (*Disque) Close

func (d *Disque) Close()

Close the main connection maintained by this Disque instance

func (*Disque) Delete

func (d *Disque) Delete(jobId string) (err error)

Delete a job that was enqueued on the cluster

func (*Disque) Fetch

func (d *Disque) Fetch(queueName string, timeout time.Duration) (job *Job, err error)

Fetch a single job from a Disque queue.

func (*Disque) FetchMultiple

func (d *Disque) FetchMultiple(queueName string, count int, timeout time.Duration) (jobs []*Job, err error)

Fetch jobs from a Disque queue.

func (*Disque) GetJobDetails

func (d *Disque) GetJobDetails(jobId string) (jobDetails *JobDetails, err error)

Retrieve details for an existing job

func (*Disque) Initialize

func (d *Disque) Initialize() (err error)

Initialize the connection, including the exploration of nodes participating in the cluster.

func (*Disque) Nack

func (d *Disque) Nack(jobId string) (err error)

Tells Disque to put back the job in the queue ASAP.

func (*Disque) Push

func (d *Disque) Push(queueName string, job string, timeout time.Duration) (jobId string, err error)

Push job onto a Disque queue with the default set of options

func (*Disque) PushWithOptions

func (d *Disque) PushWithOptions(queueName string, job string, timeout time.Duration, options map[string]string) (jobId string, err error)

Push job onto a Disque queue with options given in the options map

ADDJOB queue_name job <ms-timeout>
  [REPLICATE <count>]
  [DELAY <sec>]
  [RETRY <sec>]
  [TTL <sec>]
  [MAXLEN <count>]
  [ASYNC]

Example:

options := make(map[string]string)
options["DELAY"] = 30
options["ASYNC"] = true
d.PushWithOptions("queue_name", "job", 1*time.Second, options)

func (*Disque) QueueLength

func (d *Disque) QueueLength(queueName string) (queueLength int, err error)

Retrieve length of queue

type DisquePool

type DisquePool struct {
	// contains filtered or unexported fields
}

func NewDisquePool

func NewDisquePool(servers []string, cycle int, capacity int, maxCapacity int, idleTimeout time.Duration) (d *DisquePool)

NewDisquePool creates a new pool of Disque connections. capacity is the number of active resources in the pool: there can be up to 'capacity' of these at a given time. maxCapacity specifies the extent to which the pool can be resized in the future through the SetCapacity function. You cannot resize the pool beyond maxCapacity. If a resource is unused beyond idleTimeout, it's discarded. An idleTimeout of 0 means that there is no timeout.

func (*DisquePool) Close

func (d *DisquePool) Close()

Close empties the pool calling Close on all its resources. You can call Close while there are outstanding resources. It waits for all resources to be returned (Put). After a Close, Get is not allowed.

func (*DisquePool) Get

func (d *DisquePool) Get(ctx context.Context) (conn *Disque, err error)

Get will return the next available resource. If capacity has not been reached, it will create a new one using the factory. Otherwise, it will wait until the supplied context expires.

func (*DisquePool) IsClosed

func (d *DisquePool) IsClosed() (closed bool)

IsClosed returns true if the resource pool is closed.

func (*DisquePool) Put

func (d *DisquePool) Put(conn *Disque)

Put will return a resource to the pool. For every successful Get, a corresponding Put is required. If you no longer need a resource, you will need to call Put(nil) instead of returning the closed resource. The will eventually cause a new resource to be created in its place.

func (*DisquePool) SetCapacity

func (d *DisquePool) SetCapacity(capacity int)

SetCapacity changes the capacity of the pool. You can use it to shrink or expand, but not beyond the max capacity. If the change requires the pool to be shrunk, SetCapacity waits till the necessary number of resources are returned to the pool. A SetCapacity of 0 is equivalent to closing the ResourcePool.

type Job

type Job struct {
	QueueName string
	JobId     string
	Message   string
}

type JobDetails

type JobDetails struct {
	JobId                string
	QueueName            string
	State                string
	ReplicationFactor    int
	TTL                  time.Duration
	CreatedAt            time.Time
	Delay                time.Duration
	Retry                time.Duration
	Nacks                int64
	AdditionalDeliveries int64
	NodesDelivered       []string
	NodesConfirmed       []string
	NextRequeueWithin    time.Duration
	NextAwakeWithin      time.Duration
	Message              string
}