Class: Tuber::Jobs

Inherits:
Object
  • Object
show all
Defined in:
lib/tuber/job/collection.rb

Overview

Represents collection of job-related commands.

Constant Summary collapse

MAX_RETRIES =

Number of retries to process a job.

3
RELEASE_DELAY =

Delay in seconds before to make job ready again.

1
RESERVE_TIMEOUT =

Number of seconds to wait for a job before checking a different server.

nil

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client) ⇒ Jobs

Creates new jobs instance.

Examples:

Tuber::Jobs.new(@client)

Parameters:

  • client (Tuber) —

    The tuber client instance.



34
35
36
 
# File 'lib/tuber/job/collection.rb', line 34

def initialize(client)
  @client = client
end

Instance Attribute Details

#client ⇒ Tuber

Returns the client instance

Returns:

  • (Tuber) —

    returns the client instance



17
 
# File 'lib/tuber/job/collection.rb', line 17

attr_reader :processors, :client, :current_job

#current_job ⇒ Object

Returns the value of attribute current_job.



17
 
# File 'lib/tuber/job/collection.rb', line 17

attr_reader :processors, :client, :current_job

#processors ⇒ Array<Proc>

Returns Collection of proc to handle beanstalkd jobs

Returns:

  • (Array<Proc>) —

    returns Collection of proc to handle beanstalkd jobs



17
18
19
 
# File 'lib/tuber/job/collection.rb', line 17

def processors
  @processors
end

Instance Method Details

#find(id) ⇒ Tuber::Job Also known as: peek, []

Peek (or find) job by id from beanstalkd.

Examples:

@tuber.jobs[123] # => <Tuber::Job>
@tuber.jobs.find(123) # => <Tuber::Job>
@tuber.jobs.peek(123) # => <Tuber::Job>

Parameters:

  • id (Integer) —

    Job id to find

Returns:



58
59
60
61
62
63
 
# File 'lib/tuber/job/collection.rb', line 58

def find(id)
  res = transmit("peek #{id}")
  Job.new(client, res)
rescue Tuber::NotFoundError
  nil
end

#process!(options = {}) ⇒ Object

Watch, reserve, process and delete or bury or release jobs.

Parameters:

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

    Settings for processing

Options Hash (options):

  • release_delay (Integer) —

    Delay in seconds before to make job ready again

  • reserve_timeout (Integer) —

    Number of seconds to wait for a job before checking a different server



131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/tuber/job/collection.rb', line 131

def process!(options={})
  release_delay = options.delete(:release_delay) || RELEASE_DELAY
  reserve_timeout = options.delete(:reserve_timeout) || RESERVE_TIMEOUT
  client.tubes.watch!(*processors.keys)
  while !stop? do
    begin
      @current_job = client.tubes.reserve(reserve_timeout)
      processor = processors[@current_job.tube]
      begin
        processor[:block].call(@current_job)
        @current_job.delete
      rescue *processor[:retry_on]
        if @current_job.stats.releases < processor[:max_retries]
          @current_job.release(:delay => release_delay)
        end
      end
    rescue AbortProcessingError
      break
    rescue Tuber::JobNotReserved, Tuber::NotFoundError, Tuber::TimedOutError
      retry
    rescue StandardError # handles unspecified errors
      @current_job.bury if @current_job
    ensure # bury if still reserved
      @current_job.bury if @current_job && @current_job.exists? && @current_job.reserved?
      @current_job = nil
    end
  end
end

#register(tube_name, options = {}, &block) ⇒ Object

Register a processor to handle beanstalkd job on particular tube.

Examples:

@beanstalk.jobs.register('some-tube', :retry_on => [SomeError]) do |job|
  do_something(job)
end

@beanstalk.jobs.register('other-tube') do |job|
  do_something_else(job)
end

Parameters:

  • tube_name (String) —

    Tube name

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

    settings for processor

  • block (Proc) —

    Process beanstalkd job

Options Hash (options):

  • max_retries (Integer) —

    Number of retries to process a job

  • retry_on (Array<RuntimeError>) —

    Collection of errors to rescue and re-run processor



105
106
107
108
109
110
 
# File 'lib/tuber/job/collection.rb', line 105

def register(tube_name, options={}, &block)
  @processors ||= {}
  max_retries = options[:max_retries] || MAX_RETRIES
  retry_on = Array(options[:retry_on])
  @processors[tube_name.to_s] = { :block => block, :retry_on => retry_on, :max_retries => max_retries }
end

#stop! ⇒ Object

Sets flag to indicate that process loop should stop after current job



113
114
115
 
# File 'lib/tuber/job/collection.rb', line 113

def stop!
  @stop = true
end

#stop? ⇒ Boolean

Returns whether the process loop should stop

Returns:

  • (Boolean) —

    if true the loop should stop after current processing



120
121
122
 
# File 'lib/tuber/job/collection.rb', line 120

def stop?
  !!@stop
end

#touch_all ⇒ Integer

Extends the ttr of every job currently reserved by this connection.

A single heartbeat for a whole +reserve_batch+ window: the server already tracks the reserved set per connection, so no ids are sent and jobs that were already deleted, released, buried or lost to a ttr timeout are simply absent. Each job keeps its own ttr; deadlines are extended individually.

The returned count is how many jobs the connection actually still holds. If it is lower than expected, jobs hit their ttr and went back to the queue while the worker was busy.

Examples:

@tuber.jobs.touch_all # => 10

Returns:

  • (Integer) —

    Number of held jobs whose deadline was extended



83
84
85
 
# File 'lib/tuber/job/collection.rb', line 83

def touch_all
  transmit("touch-all")[:id].to_i
end

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

Delegates transmit to the connection object.

See Also:



41
42
43
44
45
46
 
# File 'lib/tuber/job/collection.rb', line 41

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