Class: Tuber::Tubes

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/tuber/tube/collection.rb

Overview

Represents collection of tube related commands.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client) ⇒ Tubes

Creates new tubes instance.

Examples:

Tuber::Tubes.new(@client)

Parameters:

  • client (Tuber) —

    The tuber client instance.



19
20
21
 
# File 'lib/tuber/tube/collection.rb', line 19

def initialize(client)
  @client = client
end

Instance Attribute Details

#client ⇒ Object

Returns the value of attribute client.



11
12
13
 
# File 'lib/tuber/tube/collection.rb', line 11

def client
  @client
end

Instance Method Details

#all ⇒ Array<Tuber::Tube>

List of all known beanstalk tubes.

Examples:

@client.tubes.all
  # => [<Tuber::Tube name="tube2">, <Tuber::Tube name="tube3">]

Returns:

  • (Array<Tuber::Tube>) —

    List of all beanstalk tubes.



147
148
149
150
151
 
# File 'lib/tuber/tube/collection.rb', line 147

def all
  transmit('list-tubes')[:body].map do |tube_name|
    Tube.new(client, tube_name)
  end
end

#each(&block) ⇒ Object

Calls the given block once for each known beanstalk tube, passing that element as a parameter.

Examples:

@pool.tubes.each {|t| puts t.name}

Returns:

  • An Enumerator is returned if no block is given.



160
161
162
 
# File 'lib/tuber/tube/collection.rb', line 160

def each(&block)
  all.each(&block)
end

#find(tube_name) ⇒ Tuber::Tube Also known as: []

Finds the specified beanstalk tube.

Examples:

@pool.tubes.find('tube2')
@pool.tubes['tube2']
  # => <Tuber::Tube name="tube2">

Parameters:

  • tube_name (String) —

    Name of the beanstalkd tube

Returns:



51
52
53
 
# File 'lib/tuber/tube/collection.rb', line 51

def find(tube_name)
  Tube.new(client, tube_name)
end

#ignore(*names) ⇒ Object

Ignores specified beanstalkd tubes.

Examples:

@client.tubes.ignore('foo', 'bar')

Parameters:

  • names (*String) —

    Name of tubes to ignore



232
233
234
235
236
237
 
# File 'lib/tuber/tube/collection.rb', line 232

def ignore(*names)
  names.each do |w|
    transmit "ignore #{w}"
    client.connection.remove_from_watched(w)
  end
end

#last_used ⇒ Object 



23
24
25
 
# File 'lib/tuber/tube/collection.rb', line 23

def last_used
  client.connection.tube_used
end

#last_used=(tube_name) ⇒ Object 



27
28
29
 
# File 'lib/tuber/tube/collection.rb', line 27

def last_used=(tube_name)
  client.connection.tube_used = tube_name
end

#reserve(timeout = nil, &block) {|job| ... } ⇒ Tuber::Job

Reserves a ready job looking at all watched tubes.

Examples:

@client.tubes.reserve { |job| process(job) }
  # => <Tuber::Job id=5 body="foo">

Parameters:

  • timeout (Integer) (defaults to: nil) —

    Number of seconds before timing out.

  • block (Proc) —

    Callback to perform on the reserved job.

Yields:

  • (job) —

    Reserved tuber job.

Returns:



67
68
69
70
71
72
73
 
# File 'lib/tuber/tube/collection.rb', line 67

def reserve(timeout=nil, &block)
  res = transmit(
    timeout ? "reserve-with-timeout #{timeout}" : 'reserve')
  job = Job.new(client, res)
  block.call(job) if block_given?
  job
end

#reserve_batch(count, timeout = nil) ⇒ Array<Tuber::Job>

Reserves a batch of ready jobs from watched tubes.

Without a +timeout+ the call is non-blocking and may return an empty array. With a positive +timeout+ it long-polls, blocking up to +timeout+ seconds for the first job before draining whatever is ready, up to +count+.

Examples:

@client.tubes.reserve_batch(10)       # non-blocking
@client.tubes.reserve_batch(10, 30)   # long-poll up to 30s
  # => [<Tuber::Job id=1 body="foo">, ...]

Parameters:

  • count (Integer) —

    Maximum number of jobs to reserve

  • timeout (Integer) (defaults to: nil) —

    Seconds to long-poll for the first job (nil = non-blocking)

Returns:

  • (Array<Tuber::Job>) —

    Array of reserved jobs (empty if none available)

Raises:



91
92
93
94
 
# File 'lib/tuber/tube/collection.rb', line 91

def reserve_batch(count, timeout = nil)
  results = client.connection.reserve_batch(count, timeout)
  results.map { |res| Job.new(client, res) }
end

#reserve_job(id) ⇒ Tuber::Job?

Reserves a specific job by its ID.

Examples:

@client.tubes.reserve_job(123)
  # => <Tuber::Job id=123 body="foo">

Parameters:

  • id (Integer, String) —

    The job ID to reserve

Returns:

  • (Tuber::Job, nil) —

    The reserved job, or nil if not found



118
119
120
121
122
123
 
# File 'lib/tuber/tube/collection.rb', line 118

def reserve_job(id)
  res = transmit("reserve-job #{id}")
  Job.new(client, res)
rescue Tuber::NotFoundError
  nil
end

#reserve_mode(mode) ⇒ Hash

Sets the reserve mode for the connection.

Examples:

@client.tubes.reserve_mode(:weighted)
@client.tubes.reserve_mode(:fifo)

Parameters:

  • mode (String, Symbol) —

    The reserve mode ('weighted' or 'fifo')

Returns:

  • (Hash) —

    Response from beanstalkd



105
106
107
 
# File 'lib/tuber/tube/collection.rb', line 105

def reserve_mode(mode)
  transmit("reserve-mode #{mode}")
end

#stats_group(group) ⇒ Tuber::StatStruct

Returns stats for a job group.

Examples:

@client.tubes.stats_group('batch-1')
  # => #<StatStruct name="batch-1" ready=5 reserved=0 delayed=0 buried=0 waiting_jobs=0>

Parameters:

  • group (String) —

    The group name

Returns:



134
135
136
137
 
# File 'lib/tuber/tube/collection.rb', line 134

def stats_group(group)
  res = transmit("stats-group #{group}")
  StatStruct.from_hash(res[:body])
end

#transmit(command, **options) ⇒ Object

Delegates transmit to the connection object.

See Also:



34
35
36
37
38
39
 
# File 'lib/tuber/tube/collection.rb', line 34

def transmit(command, **options)
  # Empty **options must not be forwarded: on Ruby 2.7 it arrives as an
  # extra positional {} at the receiver.
  return client.connection.transmit(command) if options.empty?
  client.connection.transmit(command, **options)
end

#use(tube) ⇒ Object

Set specified tube as used.

Examples:

@conn.tubes.use("some-tube")

Parameters:

  • tube (String) —

    Tube to be used.



245
246
247
248
249
250
251
 
# File 'lib/tuber/tube/collection.rb', line 245

def use(tube)
  return tube if last_used == tube
  transmit("use #{tube}")
  self.last_used = tube
rescue BadFormatError
  raise InvalidTubeName, "Tube cannot be named '#{tube}'"
end

#used ⇒ Tuber::Tube

Currently used beanstalk tube.

Examples:

@client.tubes.used
  # => <Tuber::Tube name="tube2">

Returns:



188
189
190
191
 
# File 'lib/tuber/tube/collection.rb', line 188

def used
  last_used = transmit('list-tube-used')[:id]
  Tube.new(client, last_used)
end

#watch(*names, weight: nil) ⇒ Object

Add specified beanstalkd tubes as watched.

Examples:

@client.tubes.watch('foo', 'bar')

Parameters:

  • names (*String) —

    Name of tubes to watch

Raises:



201
202
203
204
205
206
207
208
209
 
# File 'lib/tuber/tube/collection.rb', line 201

def watch(*names, weight: nil)
  names.each do |t|
    cmd = weight ? "watch #{t} #{weight}" : "watch #{t}"
    transmit cmd
    client.connection.add_to_watched(t)
  end
rescue BadFormatError => ex
  raise InvalidTubeName, "Tube in '#{ex.cmd}' is invalid!"
end

#watch!(*names) ⇒ Object

Add specified beanstalkd tubes as watched and ignores all other tubes.

Examples:

@client.tubes.watch!('foo', 'bar')

Parameters:

  • names (*String) —

    Name of tubes to watch

Raises:



219
220
221
222
223
 
# File 'lib/tuber/tube/collection.rb', line 219

def watch!(*names)
  old_tubes = watched.map(&:name) - names.map(&:to_s)
  watch(*names)
  ignore(*old_tubes)
end

#watched ⇒ Array<Tuber::Tube>

List of watched beanstalk tubes.

Examples:

@client.tubes.watched
  # => [<Tuber::Tube name="tube2">, <Tuber::Tube name="tube3">]

Returns:

  • (Array<Tuber::Tube>) —

    List of watched beanstalk tubes.



172
173
174
175
176
177
178
 
# File 'lib/tuber/tube/collection.rb', line 172

def watched
  last_watched = transmit('list-tubes-watched')[:body]
  client.connection.tubes_watched = last_watched.dup
  last_watched.map do |tube_name|
    Tube.new(client, tube_name)
  end
end