Class: Tuber::Tube

Inherits:
Object
  • Object
show all
Defined in:
lib/tuber/tube/record.rb

Overview

Beanstalk tube which contains jobs which can be inserted, reserved, et al.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client, name) ⇒ Tube

Fetches the specified tube.

Examples:

Tuber::Tube.new(@client, 'tube-name')

Parameters:

  • client (Tuber) —

    The tuber client instance.

  • name (String) —

    The name for this tube.



20
21
22
23
24
 
# File 'lib/tuber/tube/record.rb', line 20

def initialize(client, name)
  @client = client
  @name = name.to_s
  @mutex = Mutex.new
end

Instance Attribute Details

#client ⇒ Object

Returns the value of attribute client.



11
 
# File 'lib/tuber/tube/record.rb', line 11

attr_reader :name, :client

#name ⇒ String

Returns name of the tube.

Returns:

  • (String) —

    name of the tube



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

def name
  @name
end

Instance Method Details

#clear ⇒ Object

Clears all unreserved jobs in all states from the tube

Examples:

@tube.clear


179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/tuber/tube/record.rb', line 179

def clear
  client.tubes.watch!(self.name)
  %w(delayed buried ready).each do |state|
    while job = self.peek(state.to_sym)
      begin
        job.delete
      rescue Tuber::UnexpectedResponse, Tuber::NotFoundError
        # swallow any issues
      end
    end
  end
  client.tubes.ignore(name)
rescue Tuber::NotIgnoredError
  # swallow any issues
end

#config ⇒ Tuber::Configuration (protected)

Returns configuration options for tuber

Returns:



227
228
229
 
# File 'lib/tuber/tube/record.rb', line 227

def config
  Tuber.configuration
end

#flush ⇒ Integer

Atomically deletes all jobs from the tube.

Examples:

@tube.flush # => 5

Returns:

  • (Integer) —

    Number of jobs flushed



156
157
158
159
 
# File 'lib/tuber/tube/record.rb', line 156

def flush
  res = transmit("flush-tube #{name}")
  res[:id].to_i
end

#flush_buried ⇒ Integer

Atomically deletes all buried jobs from the tube. Ready, delayed, and reserved jobs are left untouched. (Tuber only.)

Examples:

@tube.flush_buried # => 3

Returns:

  • (Integer) —

    Number of buried jobs flushed



169
170
171
172
 
# File 'lib/tuber/tube/record.rb', line 169

def flush_buried
  res = transmit("flush-buried #{name}")
  res[:id].to_i
end

#kick(bounds = 1) ⇒ Hash{String => String, Number}

Kick specified number of jobs from buried to ready state.

Examples:

@tube.kick(5)

Parameters:

  • bounds (Integer) (defaults to: 1) —

    The number of jobs to kick.

Returns:

  • (Hash{String => String, Number}) —

    Beanstalkd command response



121
122
123
 
# File 'lib/tuber/tube/record.rb', line 121

def kick(bounds=1)
  safe_use { transmit("kick #{bounds}") }
end

#pause(delay) ⇒ Array<Hash{String => String, Number}>

Pause the execution of this tube for specified delay.

Examples:

@tube.pause(10)

Parameters:

  • delay (Integer) —

    Number of seconds to delay tube execution

Returns:

  • (Array<Hash{String => String, Number}>) —

    Beanstalkd command response



145
146
147
 
# File 'lib/tuber/tube/record.rb', line 145

def pause(delay)
  transmit("pause-tube #{name} #{delay}")
end

#peek(state) ⇒ Tuber::Job

Peek at next job within this tube in given state.

Examples:

@tube.peek(:ready) # => <Tuber::Job id=5 body=foo>

Parameters:

  • state (String) —

    The job state to peek at (ready, buried, delayed)

Returns:



88
89
90
91
92
93
94
95
96
 
# File 'lib/tuber/tube/record.rb', line 88

def peek(state)
  safe_use do
    res = transmit("peek-#{state}")
    Job.new(client, res)
  end
rescue Tuber::NotFoundError
  # Return nil if not found
  nil
end

#put(body, options = {}) ⇒ Hash{String => String, Number}

Inserts job with specified body onto tube.

Examples:

@tube.put "data", pri: 1000, ttr: 10, delay: 5
@tube.put "data", idp: "report", con: ["db", 3], grp: "batch-1", aft: "batch-0"

Parameters:

  • body (String) —

    The data to store with this job.

  • options (Hash{String => Integer}) (defaults to: {}) —

    The settings associated with this job.

Options Hash (options):

  • pri (Integer) —

    priority for this job

  • ttr (Integer) —

    time to respond for this job

  • delay (Integer) —

    delay for this job

  • idp (String, Array) —

    idempotency key, or [key, ttl] pair

  • con (String, Array) —

    concurrency key, or [key, limit] pair

  • grp (String) —

    group key for this job

  • aft (String) —

    after-group key (run after group completes)

Returns:

  • (Hash{String => String, Number}) —

    beanstalkd command response



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/tuber/tube/record.rb', line 53

def put(body, options={})
  safe_use do
    serialized_body = config.job_serializer.call(body)

    options = {
      :pri   => config.default_put_pri,
      :delay => config.default_put_delay,
      :ttr   => config.default_put_ttr
    }.merge(options)

    cmd_options = "#{options[:pri]} #{options[:delay]} #{options[:ttr]} #{serialized_body.bytesize}"

    tags = []
    if options[:idp]
      tags << (options[:idp].is_a?(Array) ? "idp:#{options[:idp].join(':')}" : "idp:#{options[:idp]}")
    end
    if options[:con]
      tags << (options[:con].is_a?(Array) ? "con:#{options[:con].join(':')}" : "con:#{options[:con]}")
    end
    tags << "grp:#{options[:grp]}" if options[:grp]
    tags << "aft:#{options[:aft]}" if options[:aft]

    cmd_options = "#{cmd_options} #{tags.join(' ')}" if tags.any?
    transmit("put #{cmd_options}\r\n#{serialized_body}")
  end
end

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

Reserves the next job from tube.

Examples:

@tube.reserve # => <Tuber::Job id=5 body=foo>

Parameters:

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

    Number of seconds before timing out

  • block (Proc) —

    Callback to perform on reserved job

Yields:

  • (job) —

    Job that was reserved.

Returns:



108
109
110
111
 
# File 'lib/tuber/tube/record.rb', line 108

def reserve(timeout=nil, &block)
  client.tubes.watch!(self.name)
  client.tubes.reserve(timeout, &block)
end

#safe_use(&block) ⇒ Object (protected)

Transmits a beanstalk command that requires this tube to be set as used.

Examples:

safe_use { transmit("kick 1") }
  # => "Response to kick command"

Parameters:

  • block (Proc) —

    Beanstalk command to transmit.

Returns:

  • (Object) —

    Result of block passed



216
217
218
219
220
221
222
 
# File 'lib/tuber/tube/record.rb', line 216

def safe_use(&block)
  @mutex.lock
  client.tubes.use(self.name)
  yield
ensure
  @mutex.unlock
end

#stats ⇒ Tuber::StatStruct

Returns related stats for this tube.

Examples:

@tube.stats.current_jobs_delayed # => 24

Returns:



132
133
134
135
 
# File 'lib/tuber/tube/record.rb', line 132

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

#to_s ⇒ String Also known as: inspect

String representation of tube.

Examples:

@tube.to_s # => "#<Tuber::Tube name=foo>"

Returns:

  • (String) —

    Representation of tube including name.



201
202
203
 
# File 'lib/tuber/tube/record.rb', line 201

def to_s
  "#<Tuber::Tube name=#{name.inspect}>"
end

#transmit(command, options = {}) ⇒ Object

Delegates transmit to the connection object.

See Also:



29
30
31
32
33
34
 
# File 'lib/tuber/tube/record.rb', line 29

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